joeyutong commented on code in PR #885:
URL: https://github.com/apache/flink-agents/pull/885#discussion_r3592532182
##########
runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java:
##########
@@ -196,6 +207,129 @@ void testPruneState() throws Exception {
assertNull(
actionStates.get(ActionStateUtil.generateKey(TEST_KEY, 2L,
testAction, testEvent)));
assertNotNull(actionStateStore.get(TEST_KEY, 3L, testAction,
testEvent));
+
+ // Assert - tombstones should have been sent to Kafka
+ var history = mockProducer.history();
+ assertThat(history).hasSize(2);
+ for (ProducerRecord<String, ActionState> record : history) {
+ assertThat(record.topic()).isEqualTo(TEST_TOPIC);
+ assertThat(record.key()).startsWith(TEST_KEY + "_");
+ assertThat(record.value()).isNull();
+ }
+ }
+
+ @Test
+ void testPruneStateSendsTombstonesWithCorrectKeys() throws Exception {
+ // Arrange
+ actionStateStore = tombstoneEnabledStore(actionStates, mockProducer);
+ String key1 = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction,
testEvent);
+ String key2 = ActionStateUtil.generateKey(TEST_KEY, 2L, testAction,
testEvent);
+ String key3 = ActionStateUtil.generateKey(TEST_KEY, 3L, testAction,
testEvent);
+ actionStates.put(key1, testActionState);
+ actionStates.put(key2, testActionState);
+ actionStates.put(key3, testActionState);
+
+ // Act
+ actionStateStore.pruneState(TEST_KEY, 2L);
+
+ // Assert - exactly keys for seqNum 1 and 2 appear as tombstones
+ var history = mockProducer.history();
+
assertThat(history).extracting(ProducerRecord::key).containsExactlyInAnyOrder(key1,
key2);
+
assertThat(history).extracting(ProducerRecord::value).containsOnlyNulls();
+ }
+
+ @Test
+ void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws
Exception {
+ // Arrange - agent key "a" seq 1 yields state key "a_1_<uuid>_<uuid>",
which is a
+ // prefix match for pruning agent key "a_1"
+ actionStateStore = tombstoneEnabledStore(actionStates, mockProducer);
+ String otherKeyState = ActionStateUtil.generateKey("a", 1L,
testAction, testEvent);
+ actionStates.put(otherKeyState, testActionState);
+
+ // Act - prune a DIFFERENT agent key whose name collides with "a"'s
key prefix
+ actionStateStore.pruneState("a_1", 10L);
+
+ // Assert - agent key "a"'s state is untouched and no tombstones were
sent
+ assertThat(actionStates).containsKey(otherKeyState);
+ assertThat(mockProducer.history()).isEmpty();
+ }
+
+ @Test
+ void testPruneStateEvictsCacheEvenWhenTombstoneSendFails() throws
Exception {
+ // Arrange - the next send() will fail asynchronously (e.g. broker
unavailable)
+ actionStateStore = tombstoneEnabledStore(actionStates, mockProducer);
+ String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L,
testAction, testEvent);
+ actionStates.put(stateKey, testActionState);
+ mockProducer.errorNext(new RuntimeException("simulated broker
failure"));
Review Comment:
mockProducer uses autoComplete=true, so errorNext() is called before any
pending completion exists and does not make the subsequent send() fail. This
test therefore exercises the successful send path. Could we inject an exception
through the callback so the async failure path is actually covered?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
