From 0db12a655c9aecbc8e1bb830197c63ae65d03123 Mon Sep 17 00:00:00 2001 From: Robert Ji Date: Tue, 7 Jul 2026 17:22:13 -0700 Subject: [PATCH 01/22] fix: handle null-valued records in KafkaActionStateStore.rebuildState() --- .../runtime/actionstate/KafkaActionStateStore.java | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index f09e5cd29..ed34d7530 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -254,11 +254,17 @@ public void rebuildState(List recoveryMarkers) { for (ConsumerRecord record : records) { try { - actionStates.put(record.key(), record.value()); + if (record.value() == null) { + // tombstone record — remove the key from cache + actionStates.remove(record.key()); + } else { + actionStates.put(record.key(), record.value()); + } } catch (Exception e) { LOG.warn( - "Failed to deserialize action state record: {}", - record.value().toString(), + "Failed to deserialize action state record: key={}, value={}", + record.key(), + record.value(), e); } } From 98b1aaaf34691f50a56f441873cae75f3e027237 Mon Sep 17 00:00:00 2001 From: Robert Ji Date: Tue, 7 Jul 2026 17:22:57 -0700 Subject: [PATCH 02/22] feat: send tombstone records in KafkaActionStateStore.pruneState() --- .../actionstate/KafkaActionStateStore.java | 65 +++++++++++------- .../KafkaActionStateStoreTest.java | 66 +++++++++++++++++++ 2 files changed, 107 insertions(+), 24 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index ed34d7530..1914e74ae 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -282,30 +282,47 @@ public void rebuildState(List recoveryMarkers) { public void pruneState(Object key, long seqNum) { LOG.debug("Pruning state for key: {} up to sequence number: {}", key, seqNum); - // Remove states from in-memory cache for this key up to the specified sequence - // number - actionStates - .entrySet() - .removeIf( - entry -> { - String stateKey = entry.getKey(); - // Extract key and sequence number from the state key - // State key format: "key_seqNum_action_event" - if (stateKey.startsWith(key.toString() + "_")) { - try { - List parts = ActionStateUtil.parseKey(stateKey); - if (parts.size() >= 2) { - long stateSeqNum = Long.parseLong(parts.get(1)); - return stateSeqNum <= seqNum; - } - } catch (NumberFormatException e) { - LOG.warn( - "Failed to parse sequence number from state key: {}", - stateKey); - } - } - return false; - }); + // collect keys that match the prune predicate + List keysToPrune = new ArrayList<>(); + for (Map.Entry entry : actionStates.entrySet()) { + String stateKey = entry.getKey(); + if (stateKey.startsWith(key.toString() + "_")) { + try { + List parts = ActionStateUtil.parseKey(stateKey); + if (parts.size() >= 2) { + long stateSeqNum = Long.parseLong(parts.get(1)); + if (stateSeqNum <= seqNum) { + keysToPrune.add(stateKey); + } + } + } catch (Exception e) { + LOG.warn("Failed to parse state key: {}", stateKey, e); + } + } + } + + // send tombstones to kafka so log compaction can reclaim storage + if (producer != null && !keysToPrune.isEmpty()) { + try { + for (String stateKey : keysToPrune) { + producer.send(new ProducerRecord<>(topic, stateKey, null)); + } + producer.flush(); + LOG.debug( + "Sent {} tombstone records to Kafka for key: {}", keysToPrune.size(), key); + } catch (Exception e) { + LOG.warn( + "Failed to send tombstone records to Kafka for key: {}. " + + "Records will persist in the topic until manual cleanup.", + key, + e); + } + } + + // remove from in-memory cache (always, regardless of tombstone success) + for (String stateKey : keysToPrune) { + actionStates.remove(stateKey); + } LOG.debug("Pruned state for key: {} up to sequence number: {}", key, seqNum); } diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 7cf829c00..659d1f769 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -34,6 +34,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import static org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy.EARLIEST; import static org.assertj.core.api.Assertions.assertThat; @@ -196,6 +197,71 @@ 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 record : history) { + assertThat(record.topic()).isEqualTo(TEST_TOPIC); + assertThat(record.key()).startsWith(TEST_KEY + "_"); + assertThat(record.value()).isNull(); + } + } + + @Test + void testPruneStateSendsTombstonesWithCorrectKeys() throws Exception { + // Arrange + 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).hasSize(2); + List tombstoneKeys = + history.stream().map(ProducerRecord::key).sorted().collect(Collectors.toList()); + List expectedKeys = + List.of(key1, key2).stream().sorted().collect(Collectors.toList()); + assertThat(tombstoneKeys).isEqualTo(expectedKeys); + assertThat(history).allMatch(r -> r.value() == null); + } + + @Test + void testPruneStateNoMatchingKeys() throws Exception { + // Arrange - add states for a different key + actionStates.put( + ActionStateUtil.generateKey("other-key", 1L, testAction, testEvent), + testActionState); + + // Act + actionStateStore.pruneState(TEST_KEY, 2L); + + // Assert - no tombstones sent, other key's state remains + assertThat(mockProducer.history()).isEmpty(); + assertThat(actionStates).hasSize(1); + } + + @Test + void testPruneStateWithNullProducer() throws Exception { + // Arrange - store with null producer + Map localStates = new HashMap<>(); + KafkaActionStateStore nullProducerStore = + new KafkaActionStateStore( + localStates, new AgentConfiguration(), null, mockConsumer, TEST_TOPIC); + localStates.put( + ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent), testActionState); + + // Act - should not throw + nullProducerStore.pruneState(TEST_KEY, 1L); + + // Assert - in-memory removal still works + assertThat(localStates).isEmpty(); } @Test From 2c033e7b73eb3ec3c3398d784643e17b9314e4c1 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Tue, 14 Jul 2026 13:08:49 -0700 Subject: [PATCH 03/22] feat: make tombstone emission opt-in and harden pruning path - add kafkaActionStateTombstoneEnabled (default false) so tombstone emission is opt-in; rebuildState still honors tombstones already in the topic regardless of the flag - match the parsed key part exactly in pruneState so pruning key "a_1" can no longer tombstone state of the distinct key "a" - report async tombstone send failures via producer callback (flush() does not surface per-record errors) - narrow the prune catch to IllegalArgumentException and state the retention consequence in the warning - remove dead inner try/catch in rebuildState (deserialization errors throw from poll(), not from the map ops it wrapped) - document the durable-deletion replay constraint on ActionStateStore.pruneState and add the new option to the config docs - tests: default-off pruning, prefix-collision regression, tombstone replay in rebuildState; simplify assertions --- .../api/configuration/AgentConfigOptions.java | 13 +++ docs/content/docs/operations/configuration.md | 1 + .../runtime/actionstate/ActionStateStore.java | 7 ++ .../actionstate/KafkaActionStateStore.java | 85 +++++++++++-------- .../KafkaActionStateStoreTest.java | 79 ++++++++++++++--- 5 files changed, 139 insertions(+), 46 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index c39997da1..7d9f9fcc9 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -64,6 +64,19 @@ public class AgentConfigOptions { public static final ConfigOption KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR = new ConfigOption<>("kafkaActionStateTopicReplicationFactor", Integer.class, 1); + /** + * The config parameter determines whether pruning sends tombstone (null-valued) records to the + * Kafka action state topic so log compaction can reclaim pruned keys. Defaults to {@code + * false}: without tombstones the topic grows unboundedly, but restoring any checkpoint or + * savepoint replays correctly. When enabled, restoring from the latest completed checkpoint is + * unaffected, but restoring an older checkpoint or savepoint may replay tombstones written + * after that restore point, erasing action state the replay still needs and causing already + * completed actions to re-execute. Enable only if the job never restores from non-latest + * checkpoints or savepoints, or if re-executing actions is acceptable. + */ + public static final ConfigOption KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = + new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false); + /** The config parameter specifies the Fluss bootstrap servers. */ public static final ConfigOption FLUSS_BOOTSTRAP_SERVERS = new ConfigOption<>("flussBootstrapServers", String.class, "localhost:9123"); diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index 44ebbc0fe..e15305bf8 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -168,6 +168,7 @@ Here are the configuration options for Kafka-based Action State Store. | `kafkaActionStateTopic` | (none) | String | The config parameter specifies the Kafka topic for action state. | | `kafkaActionStateTopicNumPartitions`| 64 | Integer | The config parameter specifies the number of partitions for the Kafka action state topic. | | `kafkaActionStateTopicReplicationFactor` | 1 | Integer | The config parameter specifies the replication factor for the Kafka action state topic. | +| `kafkaActionStateTombstoneEnabled` | false | Boolean | Whether pruning sends tombstone records so log compaction can reclaim pruned keys. Off by default: without tombstones the topic grows unboundedly, but restoring any checkpoint or savepoint replays correctly. When enabled, restoring from the latest completed checkpoint is unaffected, but restoring an older checkpoint or savepoint may replay tombstones written after that restore point and re-execute already completed actions. Enable only if the job never restores from non-latest checkpoints or savepoints, or if re-executing actions is acceptable. | #### Fluss-based Action State Store diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index e29557c0d..284b7b8d0 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -77,6 +77,13 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) /** * Prune the state for a given key. * + *

Implementations must at least evict the matching entries from the in-memory cache. + * Whether the backend storage is also cleaned up is implementation-specific. Implementations + * that durably delete backend records must not let deletion outpace the oldest checkpoint or + * savepoint that may still be restored: {@link #rebuildState(List)} replays the backend from + * the restored checkpoint's recovery marker, so a durable deletion issued after that marker + * erases state the replay still needs and causes already completed actions to re-execute. + * * @param key the key whose state should be pruned * @param seqNum the sequence number up to which the state should be pruned */ diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index 1914e74ae..1b4af81e9 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -50,6 +50,7 @@ import java.util.UUID; import java.util.concurrent.TimeUnit; +import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOMBSTONE_ENABLED; import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC; import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC_NUM_PARTITIONS; import static org.apache.flink.agents.api.configuration.AgentConfigOptions.KAFKA_ACTION_STATE_TOPIC_REPLICATION_FACTOR; @@ -90,6 +91,9 @@ public class KafkaActionStateStore implements ActionStateStore { // Kafka topic that stores action states private final String topic; + // Whether pruning sends tombstone records for log compaction + private final boolean tombstoneEnabled; + @VisibleForTesting KafkaActionStateStore( Map actionStates, @@ -103,6 +107,7 @@ public class KafkaActionStateStore implements ActionStateStore { this.topic = topic; this.latestKeySeqNum = new HashMap<>(); this.agentConfiguration = agentConfiguration; + this.tombstoneEnabled = agentConfiguration.get(KAFKA_ACTION_STATE_TOMBSTONE_ENABLED); } /** Constructs a new KafkaActionStateStore with custom Kafka configuration. */ @@ -110,6 +115,7 @@ public KafkaActionStateStore(AgentConfiguration agentConfiguration) { this.actionStates = new HashMap<>(); this.latestKeySeqNum = new HashMap<>(); this.agentConfiguration = agentConfiguration; + this.tombstoneEnabled = agentConfiguration.get(KAFKA_ACTION_STATE_TOMBSTONE_ENABLED); this.topic = Preconditions.checkArgumentNotNull( agentConfiguration.get(KAFKA_ACTION_STATE_TOPIC), @@ -252,20 +258,14 @@ public void rebuildState(List recoveryMarkers) { break; } + // Deserialization failures throw from poll() itself and are handled by the + // outer catch, so records here are always fully deserialized. for (ConsumerRecord record : records) { - try { - if (record.value() == null) { - // tombstone record — remove the key from cache - actionStates.remove(record.key()); - } else { - actionStates.put(record.key(), record.value()); - } - } catch (Exception e) { - LOG.warn( - "Failed to deserialize action state record: key={}, value={}", - record.key(), - record.value(), - e); + if (record.value() == null) { + // Tombstone record - remove the key from cache + actionStates.remove(record.key()); + } else { + actionStates.put(record.key(), record.value()); } } @@ -282,30 +282,49 @@ public void rebuildState(List recoveryMarkers) { public void pruneState(Object key, long seqNum) { LOG.debug("Pruning state for key: {} up to sequence number: {}", key, seqNum); - // collect keys that match the prune predicate + // Collect state keys belonging to this key with sequence number <= seqNum. The parsed + // key part must match exactly: prefix matching alone would let pruning key "a_1" match + // state keys of the distinct key "a" (whose keys also start with "a_1_"). + String keyStr = key.toString(); + String keyPrefix = keyStr + "_"; List keysToPrune = new ArrayList<>(); - for (Map.Entry entry : actionStates.entrySet()) { - String stateKey = entry.getKey(); - if (stateKey.startsWith(key.toString() + "_")) { - try { - List parts = ActionStateUtil.parseKey(stateKey); - if (parts.size() >= 2) { - long stateSeqNum = Long.parseLong(parts.get(1)); - if (stateSeqNum <= seqNum) { - keysToPrune.add(stateKey); - } - } - } catch (Exception e) { - LOG.warn("Failed to parse state key: {}", stateKey, e); + for (String stateKey : actionStates.keySet()) { + if (!stateKey.startsWith(keyPrefix)) { + continue; + } + try { + List parts = ActionStateUtil.parseKey(stateKey); + if (parts.get(0).equals(keyStr) && Long.parseLong(parts.get(1)) <= seqNum) { + keysToPrune.add(stateKey); } + } catch (IllegalArgumentException e) { + LOG.warn( + "Cannot parse state key: {}. The entry cannot be pruned and will be " + + "retained in memory and in the topic.", + stateKey, + e); } } - // send tombstones to kafka so log compaction can reclaim storage - if (producer != null && !keysToPrune.isEmpty()) { + // Send tombstones to Kafka so log compaction can reclaim storage; opt-in because + // tombstones break replay when restoring a checkpoint/savepoint older than the prune + // (see KAFKA_ACTION_STATE_TOMBSTONE_ENABLED). Send failures surface asynchronously, + // so report them via callback; the records then persist until manual cleanup. + if (tombstoneEnabled && producer != null && !keysToPrune.isEmpty()) { try { for (String stateKey : keysToPrune) { - producer.send(new ProducerRecord<>(topic, stateKey, null)); + producer.send( + new ProducerRecord<>(topic, stateKey, null), + (metadata, exception) -> { + if (exception != null) { + LOG.warn( + "Failed to send tombstone record for state key: {}. " + + "The record will persist in the topic " + + "until manual cleanup.", + stateKey, + exception); + } + }); } producer.flush(); LOG.debug( @@ -319,10 +338,8 @@ public void pruneState(Object key, long seqNum) { } } - // remove from in-memory cache (always, regardless of tombstone success) - for (String stateKey : keysToPrune) { - actionStates.remove(stateKey); - } + // Remove from in-memory cache (always, regardless of tombstone success) + actionStates.keySet().removeAll(keysToPrune); LOG.debug("Pruned state for key: {} up to sequence number: {}", key, seqNum); } diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 659d1f769..362f7a412 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -19,11 +19,13 @@ import org.apache.flink.agents.api.Event; import org.apache.flink.agents.api.InputEvent; +import org.apache.flink.agents.api.configuration.AgentConfigOptions; import org.apache.flink.agents.plan.AgentConfiguration; import org.apache.flink.agents.plan.actions.Action; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.MockConsumer; import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; @@ -34,7 +36,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.stream.Collectors; import static org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy.EARLIEST; import static org.assertj.core.api.Assertions.assertThat; @@ -80,6 +81,14 @@ void setUp() throws Exception { testActionState = new ActionState(testEvent); } + /** Builds a store sharing this test's mock consumer but with tombstone emission enabled. */ + private KafkaActionStateStore tombstoneEnabledStore( + Map states, Producer producer) { + AgentConfiguration config = new AgentConfiguration(); + config.set(AgentConfigOptions.KAFKA_ACTION_STATE_TOMBSTONE_ENABLED, true); + return new KafkaActionStateStore(states, config, producer, mockConsumer, TEST_TOPIC); + } + @Test void testPutActionState() throws Exception { // Act @@ -176,6 +185,7 @@ void testRecoveryMarker() throws Exception { @Test void testPruneState() throws Exception { // Arrange + actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); actionStates.put( ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent), testActionState); actionStates.put( @@ -211,6 +221,7 @@ void testPruneState() throws Exception { @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); @@ -223,18 +234,46 @@ void testPruneStateSendsTombstonesWithCorrectKeys() throws Exception { // Assert - exactly keys for seqNum 1 and 2 appear as tombstones var history = mockProducer.history(); - assertThat(history).hasSize(2); - List tombstoneKeys = - history.stream().map(ProducerRecord::key).sorted().collect(Collectors.toList()); - List expectedKeys = - List.of(key1, key2).stream().sorted().collect(Collectors.toList()); - assertThat(tombstoneKeys).isEqualTo(expectedKeys); - assertThat(history).allMatch(r -> r.value() == null); + 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__", 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 testPruneStateNoTombstonesByDefault() throws Exception { + // Arrange - setUp store uses a default AgentConfiguration (tombstones disabled) + actionStates.put( + ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent), testActionState); + actionStates.put( + ActionStateUtil.generateKey(TEST_KEY, 2L, testAction, testEvent), testActionState); + + // Act + actionStateStore.pruneState(TEST_KEY, 2L); + + // Assert - no tombstones sent, but in-memory entries are still evicted + assertThat(mockProducer.history()).isEmpty(); + assertThat(actionStates).isEmpty(); } @Test void testPruneStateNoMatchingKeys() throws Exception { // Arrange - add states for a different key + actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); actionStates.put( ActionStateUtil.generateKey("other-key", 1L, testAction, testEvent), testActionState); @@ -249,11 +288,9 @@ void testPruneStateNoMatchingKeys() throws Exception { @Test void testPruneStateWithNullProducer() throws Exception { - // Arrange - store with null producer + // Arrange - tombstones enabled but producer is null Map localStates = new HashMap<>(); - KafkaActionStateStore nullProducerStore = - new KafkaActionStateStore( - localStates, new AgentConfiguration(), null, mockConsumer, TEST_TOPIC); + KafkaActionStateStore nullProducerStore = tombstoneEnabledStore(localStates, null); localStates.put( ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent), testActionState); @@ -320,4 +357,22 @@ void testRebuildState() throws Exception { ActionStateUtil.generateKey(TEST_KEY, 3L, testAction, testEvent))) .isEqualTo(thirdState); } + + @Test + void testRebuildStateRemovesTombstonedKeys() throws Exception { + // Arrange - two state records followed by a tombstone for the first key + List recoveryMarkers = List.of(Map.of(0, 0L)); + String key1 = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent); + String key2 = ActionStateUtil.generateKey(TEST_KEY, 2L, testAction, testEvent); + mockConsumer.addRecord(new ConsumerRecord<>(TEST_TOPIC, 0, 0L, key1, testActionState)); + mockConsumer.addRecord(new ConsumerRecord<>(TEST_TOPIC, 0, 1L, key2, testActionState)); + mockConsumer.addRecord(new ConsumerRecord<>(TEST_TOPIC, 0, 2L, key1, null)); + + // Act + actionStateStore.rebuildState(recoveryMarkers); + + // Assert - the tombstoned key is removed, the other key is restored + assertThat(actionStates).doesNotContainKey(key1); + assertThat(actionStates.get(key2)).isEqualTo(testActionState); + } } From e43475365bb86360a74ad433414edfc84f6a528f Mon Sep 17 00:00:00 2001 From: rob-9 Date: Wed, 15 Jul 2026 11:16:09 -0700 Subject: [PATCH 04/22] test: cover async tombstone send failures and unparseable keys in pruneState - testPruneStateEvictsCacheEvenWhenTombstoneSendFails: verifies pruneState degrades gracefully and still evicts the in-memory entry when a tombstone send fails asynchronously (the callback-reporting fix from the prior commit) - testPruneStateSkipsUnparseableKeys: verifies a state key that cannot be parsed into 4 parts is retained rather than pruned (the narrowed IllegalArgumentException catch) Both were verified to fail when the corresponding fix is reverted. --- .../KafkaActionStateStoreTest.java | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 362f7a412..554143c65 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -254,6 +254,38 @@ void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws Exception { 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")); + + // Act - should not throw despite the async send failure + actionStateStore.pruneState(TEST_KEY, 1L); + + // Assert - in-memory entry is still evicted regardless of tombstone delivery + assertThat(actionStates).doesNotContainKey(stateKey); + } + + @Test + void testPruneStateSkipsUnparseableKeys() throws Exception { + // Arrange - a state key with the right prefix but the wrong number of parts, which + // ActionStateUtil.parseKey cannot split into exactly 4 parts + actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); + String malformedKey = TEST_KEY + "_1_onlythreeparts"; + actionStates.put(malformedKey, testActionState); + + // Act - should not throw despite the unparseable key + actionStateStore.pruneState(TEST_KEY, 10L); + + // Assert - the unparseable entry is retained, and no tombstone was sent for it + assertThat(actionStates).containsKey(malformedKey); + assertThat(mockProducer.history()).isEmpty(); + } + @Test void testPruneStateNoTombstonesByDefault() throws Exception { // Arrange - setUp store uses a default AgentConfiguration (tombstones disabled) From a02a28a1f166f7c618f29955504e6ad975ef7137 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Wed, 15 Jul 2026 11:25:40 -0700 Subject: [PATCH 05/22] style: fix spotless formatting violations --- .../flink/agents/runtime/actionstate/ActionStateStore.java | 6 +++--- .../runtime/actionstate/KafkaActionStateStoreTest.java | 3 +-- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 284b7b8d0..84dc7baf6 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -77,9 +77,9 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) /** * Prune the state for a given key. * - *

Implementations must at least evict the matching entries from the in-memory cache. - * Whether the backend storage is also cleaned up is implementation-specific. Implementations - * that durably delete backend records must not let deletion outpace the oldest checkpoint or + *

Implementations must at least evict the matching entries from the in-memory cache. Whether + * the backend storage is also cleaned up is implementation-specific. Implementations that + * durably delete backend records must not let deletion outpace the oldest checkpoint or * savepoint that may still be restored: {@link #rebuildState(List)} replays the backend from * the restored checkpoint's recovery marker, so a durable deletion issued after that marker * erases state the replay still needs and causes already completed actions to re-execute. diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 554143c65..04b674eed 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -258,8 +258,7 @@ void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws Exception { 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); + String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent); actionStates.put(stateKey, testActionState); mockProducer.errorNext(new RuntimeException("simulated broker failure")); From 41c9fee7d32769ce930f09770b536e6e0bf0acb1 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 3 Aug 2026 11:23:16 -0700 Subject: [PATCH 06/22] feat: add kafkaActionStateTombstoneEnabled to python core options --- python/flink_agents/api/core_options.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/python/flink_agents/api/core_options.py b/python/flink_agents/api/core_options.py index 64f6a727a..adac20ade 100644 --- a/python/flink_agents/api/core_options.py +++ b/python/flink_agents/api/core_options.py @@ -140,6 +140,12 @@ class AgentConfigOptions: default=1, ) + KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = ConfigOption( + key="kafkaActionStateTombstoneEnabled", + config_type=bool, + default=False, + ) + FLUSS_BOOTSTRAP_SERVERS = ConfigOption( key="flussBootstrapServers", config_type=str, From 07958ffcef7aabf3b55c7cb809e4a40593888516 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 3 Aug 2026 11:21:11 -0700 Subject: [PATCH 07/22] test: exercise the real async tombstone send failure path --- .../KafkaActionStateStoreTest.java | 35 +++++++++++++++++-- 1 file changed, 32 insertions(+), 3 deletions(-) diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 834b7ba7f..4936250ce 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -25,9 +25,11 @@ import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.MockProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringSerializer; @@ -37,6 +39,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy.EARLIEST; import static org.assertj.core.api.Assertions.assertThat; @@ -261,16 +265,41 @@ void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws Exception { @Test void testPruneStateEvictsCacheEvenWhenTombstoneSendFails() throws Exception { - // Arrange - the next send() will fail asynchronously (e.g. broker unavailable) - actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); + // Arrange - a producer whose send() completes its callback with an exception, exercising + // the async failure path (mockProducer's autoComplete=true completes sends successfully + // before errorNext() can take effect, so a dedicated producer is needed here) + AtomicBoolean failureCallbackInvoked = new AtomicBoolean(); + MockProducer failingProducer = + new MockProducer<>( + false, + new ActionStateKeyPartitioner(), + new StringSerializer(), + new ActionStateKafkaSeder()) { + @Override + public synchronized Future send( + ProducerRecord record, Callback callback) { + assertThat(callback).isNotNull(); + Future future = + super.send( + record, + (metadata, exception) -> { + failureCallbackInvoked.set(exception != null); + callback.onCompletion(metadata, exception); + }); + assertThat(errorNext(new RuntimeException("simulated broker failure"))) + .isTrue(); + return future; + } + }; + actionStateStore = tombstoneEnabledStore(actionStates, failingProducer); String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent); actionStates.put(stateKey, testActionState); - mockProducer.errorNext(new RuntimeException("simulated broker failure")); // Act - should not throw despite the async send failure actionStateStore.pruneState(TEST_KEY, 1L); // Assert - in-memory entry is still evicted regardless of tombstone delivery + assertThat(failureCallbackInvoked).isTrue(); assertThat(actionStates).doesNotContainKey(stateKey); } From f46cffc97c00d41bd34522a2f498bb04a6d4f03d Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 3 Aug 2026 11:21:39 -0700 Subject: [PATCH 08/22] docs: document that keys containing '_' are never pruned or tombstoned --- .../api/configuration/AgentConfigOptions.java | 5 +++++ docs/content/docs/operations/configuration.md | 2 ++ .../runtime/actionstate/ActionStateStore.java | 4 ++++ .../actionstate/FlussActionStateStoreTest.java | 11 +++++++++++ .../actionstate/KafkaActionStateStoreTest.java | 14 ++++++++------ 5 files changed, 30 insertions(+), 6 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index fd9d0b92c..359313964 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -90,6 +90,11 @@ public enum ConditionEvaluationFailureStrategy { * after that restore point, erasing action state the replay still needs and causing already * completed actions to re-execute. Enable only if the job never restores from non-latest * checkpoints or savepoints, or if re-executing actions is acceptable. + * + *

Also note: an agent key that itself contains the {@code _} character (e.g. {@code + * user_123}) is never pruned or tombstoned regardless of this setting. Durable action state + * stores join key parts with an unescaped {@code _}, so such a key fails to parse back into its + * parts and its state is retained in memory and in backend storage instead. */ public static final ConfigOption KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false); diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index 1ffe365da..f985b99ee 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -170,6 +170,8 @@ The eight `memory.generate-event*` options have no raw `ConfigOption` default. W |------------------------------|------------------|---------|------------------------------------------------------------------------------------------| | `actionStateStoreBackend` | (none) | String | The backend for action state store. Supported values: `"kafka"`, `"fluss"`. | +Durable action state stores currently join raw agent keys and other key parts with an unescaped `_`. Agent keys containing `_` cannot be parsed safely during pruning, so both Kafka and Fluss retain their state in memory and backend storage. Kafka also emits no tombstones for those keys. + #### Kafka-based Action State Store Here are the configuration options for Kafka-based Action State Store. diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 84dc7baf6..915baabd4 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -84,6 +84,10 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) * the restored checkpoint's recovery marker, so a durable deletion issued after that marker * erases state the replay still needs and causes already completed actions to re-execute. * + *

The current durable stores encode raw agent keys using an unescaped {@code _} separator. + * Agent keys containing {@code _} therefore cannot be parsed safely during pruning and are + * retained in both the in-memory cache and backend storage. + * * @param key the key whose state should be pruned * @param seqNum the sequence number up to which the state should be pruned */ diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java index f6ba5fcc3..b73b78a2a 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java @@ -121,6 +121,17 @@ void testGetTriggersDivergenceCleanup() throws Exception { assertThat(store.get(TEST_KEY, 3L, testAction, testEvent)).isNull(); } + @Test + void testPruneStateRetainsKeyContainingUnderscore() throws Exception { + String agentKey = "user_123"; + String stateKey = ActionStateUtil.generateKey(agentKey, 1L, testAction, testEvent); + actionStates.put(stateKey, testActionState); + + store.pruneState(agentKey, 1L); + + assertThat(actionStates).containsKey(stateKey); + } + // ==================== rebuildState tests ==================== @Test diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 4936250ce..371d0a9a5 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -305,17 +305,19 @@ public synchronized Future send( @Test void testPruneStateSkipsUnparseableKeys() throws Exception { - // Arrange - a state key with the right prefix but the wrong number of parts, which - // ActionStateUtil.parseKey cannot split into exactly 4 parts + // Arrange - an agent key that itself contains the "_" separator (e.g. "user_123") + // produces a state key with 5 "_"-separated parts once seqNum and the two UUIDs are + // appended, which ActionStateUtil.parseKey cannot split into exactly 4 parts actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); - String malformedKey = TEST_KEY + "_1_onlythreeparts"; - actionStates.put(malformedKey, testActionState); + String agentKey = "user_123"; + String stateKey = ActionStateUtil.generateKey(agentKey, 1L, testAction, testEvent); + actionStates.put(stateKey, testActionState); // Act - should not throw despite the unparseable key - actionStateStore.pruneState(TEST_KEY, 10L); + actionStateStore.pruneState(agentKey, 10L); // Assert - the unparseable entry is retained, and no tombstone was sent for it - assertThat(actionStates).containsKey(malformedKey); + assertThat(actionStates).containsKey(stateKey); assertThat(mockProducer.history()).isEmpty(); } From 2cb557c46e44cda2b48c7b552e31f7115aaf84e4 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 3 Aug 2026 11:21:47 -0700 Subject: [PATCH 09/22] refactor: drop redundant producer.flush() in pruneState --- .../agents/runtime/actionstate/KafkaActionStateStore.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index 6748a9d89..91b74aabc 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -327,9 +327,10 @@ public void pruneState(Object key, long seqNum) { } }); } - producer.flush(); LOG.debug( - "Sent {} tombstone records to Kafka for key: {}", keysToPrune.size(), key); + "Queued {} tombstone records to Kafka for key: {}", + keysToPrune.size(), + key); } catch (Exception e) { LOG.warn( "Failed to send tombstone records to Kafka for key: {}. " From 2cb89d81d072600cdcd72db708f974695f733c1c Mon Sep 17 00:00:00 2001 From: rob-9 Date: Fri, 21 Aug 2026 15:03:18 -0700 Subject: [PATCH 10/22] docs: clarify tombstone recovery scope --- .../agents/api/configuration/AgentConfigOptions.java | 10 +++++----- docs/content/docs/operations/configuration.md | 2 +- .../agents/runtime/actionstate/ActionStateStore.java | 11 ++++++----- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index 359313964..3698c9840 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -84,11 +84,11 @@ public enum ConditionEvaluationFailureStrategy { /** * The config parameter determines whether pruning sends tombstone (null-valued) records to the * Kafka action state topic so log compaction can reclaim pruned keys. Defaults to {@code - * false}: without tombstones the topic grows unboundedly, but restoring any checkpoint or - * savepoint replays correctly. When enabled, restoring from the latest completed checkpoint is - * unaffected, but restoring an older checkpoint or savepoint may replay tombstones written - * after that restore point, erasing action state the replay still needs and causing already - * completed actions to re-execute. Enable only if the job never restores from non-latest + * false}: disabling this option does not invalidate older restore points through pruning, but + * the topic continues to grow. When enabled, the checkpoint whose completion triggers pruning + * remains usable, but restoring an earlier checkpoint or savepoint may replay tombstones + * written after that restore point, erasing action state the replay still needs and causing + * already completed actions to re-execute. Enable only if the job never restores from earlier * checkpoints or savepoints, or if re-executing actions is acceptable. * *

Also note: an agent key that itself contains the {@code _} character (e.g. {@code diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index f985b99ee..cad45ba11 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -182,7 +182,7 @@ Here are the configuration options for Kafka-based Action State Store. | `kafkaActionStateTopic` | (none) | String | The config parameter specifies the Kafka topic for action state. | | `kafkaActionStateTopicNumPartitions`| 64 | Integer | The config parameter specifies the number of partitions for the Kafka action state topic. | | `kafkaActionStateTopicReplicationFactor` | 1 | Integer | The config parameter specifies the replication factor for the Kafka action state topic. | -| `kafkaActionStateTombstoneEnabled` | false | Boolean | Whether pruning sends tombstone records so log compaction can reclaim pruned keys. Off by default: without tombstones the topic grows unboundedly, but restoring any checkpoint or savepoint replays correctly. When enabled, restoring from the latest completed checkpoint is unaffected, but restoring an older checkpoint or savepoint may replay tombstones written after that restore point and re-execute already completed actions. Enable only if the job never restores from non-latest checkpoints or savepoints, or if re-executing actions is acceptable. | +| `kafkaActionStateTombstoneEnabled` | false | Boolean | Whether pruning sends tombstone records so log compaction can reclaim pruned keys on a compacted action-state topic. Off by default: pruning does not invalidate older restore points, but the topic continues to grow. When enabled, the checkpoint whose completion triggers pruning remains usable, but restoring an earlier checkpoint or savepoint may replay later tombstones and re-execute already completed actions. Enable only if the job never restores from earlier checkpoints or savepoints, or if re-executing actions is acceptable. | #### Fluss-based Action State Store diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 915baabd4..67380594c 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -78,11 +78,12 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) * Prune the state for a given key. * *

Implementations must at least evict the matching entries from the in-memory cache. Whether - * the backend storage is also cleaned up is implementation-specific. Implementations that - * durably delete backend records must not let deletion outpace the oldest checkpoint or - * savepoint that may still be restored: {@link #rebuildState(List)} replays the backend from - * the restored checkpoint's recovery marker, so a durable deletion issued after that marker - * erases state the replay still needs and causes already completed actions to re-execute. + * the backend storage is also cleaned up is implementation-specific. Durable deletion can + * invalidate checkpoints or savepoints whose recovery markers precede the deletion: {@link + * #rebuildState(List)} replays the backend from the restored recovery marker, so records + * deleted after that marker may be state the replay still needs. Implementations must either + * enforce a recovery boundary that protects every supported restore point or clearly document + * the recovery trade-off of advancing beyond that boundary. * *

The current durable stores encode raw agent keys using an unescaped {@code _} separator. * Agent keys containing {@code _} therefore cannot be parsed safely during pruning and are From 14b894f6842a9a5cf1c86f227672ab7e6e01acce Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 3 Aug 2026 11:23:15 -0700 Subject: [PATCH 11/22] test: keep tombstone assertions in the dedicated test, revert testPruneState to cache eviction only --- .../actionstate/KafkaActionStateStoreTest.java | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 371d0a9a5..c6e623da0 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -194,7 +194,6 @@ void testRecoveryMarker() throws Exception { @Test void testPruneState() throws Exception { // Arrange - actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); actionStates.put( ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent), testActionState); actionStates.put( @@ -216,15 +215,6 @@ 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 record : history) { - assertThat(record.topic()).isEqualTo(TEST_TOPIC); - assertThat(record.key()).startsWith(TEST_KEY + "_"); - assertThat(record.value()).isNull(); - } } @Test @@ -243,6 +233,7 @@ void testPruneStateSendsTombstonesWithCorrectKeys() throws Exception { // Assert - exactly keys for seqNum 1 and 2 appear as tombstones var history = mockProducer.history(); + assertThat(history).extracting(ProducerRecord::topic).containsOnly(TEST_TOPIC); assertThat(history).extracting(ProducerRecord::key).containsExactlyInAnyOrder(key1, key2); assertThat(history).extracting(ProducerRecord::value).containsOnlyNulls(); } From 16b44b61015ddd8383456e5231e12068275a462d Mon Sep 17 00:00:00 2001 From: rob-9 Date: Fri, 21 Aug 2026 15:24:36 -0700 Subject: [PATCH 12/22] [runtime][docs] Harden tombstone recovery edge cases --- .../api/configuration/AgentConfigOptions.java | 2 +- docs/content/docs/operations/configuration.md | 2 +- docs/content/docs/operations/deployment.md | 6 +- .../runtime/actionstate/ActionStateStore.java | 4 +- .../actionstate/KafkaActionStateStore.java | 9 +- .../KafkaActionStateStoreTest.java | 85 ++++++++++++++++--- 6 files changed, 89 insertions(+), 19 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index 3698c9840..77ceea8ef 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -91,7 +91,7 @@ public enum ConditionEvaluationFailureStrategy { * already completed actions to re-execute. Enable only if the job never restores from earlier * checkpoints or savepoints, or if re-executing actions is acceptable. * - *

Also note: an agent key that itself contains the {@code _} character (e.g. {@code + *

Also note: a Flink key that itself contains the {@code _} character (e.g. {@code * user_123}) is never pruned or tombstoned regardless of this setting. Durable action state * stores join key parts with an unescaped {@code _}, so such a key fails to parse back into its * parts and its state is retained in memory and in backend storage instead. diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index cad45ba11..15e71eb28 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -170,7 +170,7 @@ The eight `memory.generate-event*` options have no raw `ConfigOption` default. W |------------------------------|------------------|---------|------------------------------------------------------------------------------------------| | `actionStateStoreBackend` | (none) | String | The backend for action state store. Supported values: `"kafka"`, `"fluss"`. | -Durable action state stores currently join raw agent keys and other key parts with an unescaped `_`. Agent keys containing `_` cannot be parsed safely during pruning, so both Kafka and Fluss retain their state in memory and backend storage. Kafka also emits no tombstones for those keys. +Durable action state stores currently join raw Flink keys and other key parts with an unescaped `_`. Flink keys containing `_` cannot be parsed safely during pruning, so both Kafka and Fluss retain their state in memory and backend storage. Kafka also emits no tombstones for those keys. #### Kafka-based Action State Store diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md index 384e2ab0e..5610500fe 100644 --- a/docs/content/docs/operations/deployment.md +++ b/docs/content/docs/operations/deployment.md @@ -92,7 +92,7 @@ After recovery from a checkpoint, Flink Agents reprocess events that arrived aft ### Exactly-Once Action Consistency -To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. After recovering from a checkpoint, Flink Agents consult the external store and will not re-execute actions that were already completed. This guarantees each action is executed exactly once after recovering from a checkpoint. +To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. After recovering from a checkpoint, Flink Agents consult the external store and reuse completed action state when its backing record remains available. This prevents re-execution for checkpoints supported by the store's retained recovery history. When a completed Action is reused during recovery, its stored output Events keep their original Event IDs, while their lineage is rebound to the Event that triggers the reused Action in the recovered execution. @@ -104,6 +104,10 @@ The same persisted action state is also used by fine-grained durable execution. See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for configuration options. +{{< hint warning >}} +**Note**: Enabling Kafka action-state tombstones can invalidate checkpoints or savepoints older than the prune and cause completed actions to execute again. See [Action State Store Configuration]({{< ref "docs/operations/configuration#action-state-store" >}}) for the recovery trade-off. +{{< /hint >}} + {{< hint info >}} **Note**: Exactly-once action consistency is guaranteed only if, after recovering from the same checkpoint, inputs for each key arrive in the same order as before recovery. If this ordering requirement is not met, the system falls back to exactly-once output consistency. {{< /hint >}} diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 67380594c..3b546c572 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -85,8 +85,8 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) * enforce a recovery boundary that protects every supported restore point or clearly document * the recovery trade-off of advancing beyond that boundary. * - *

The current durable stores encode raw agent keys using an unescaped {@code _} separator. - * Agent keys containing {@code _} therefore cannot be parsed safely during pruning and are + *

The current durable stores encode raw Flink keys using an unescaped {@code _} separator. + * Flink keys containing {@code _} therefore cannot be parsed safely during pruning and are * retained in both the in-memory cache and backend storage. * * @param key the key whose state should be pruned diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index 91b74aabc..fc2df4bfd 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -181,10 +181,13 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro // the requested seqNum return stateSeqNum > seqNum; } - } catch (NumberFormatException e) { + } catch (IllegalArgumentException e) { LOG.warn( - "Failed to parse sequence number from state key: {}", - stateKey); + "Cannot parse state key: {}. The entry cannot be " + + "considered for divergence cleanup and will " + + "be retained.", + entry.getKey(), + e); } return false; }); diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index c6e623da0..c87749d89 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -33,9 +33,19 @@ import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Configuration; +import org.apache.logging.log4j.core.config.LoggerConfig; +import org.apache.logging.log4j.core.layout.PatternLayout; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -155,6 +165,16 @@ void testGetActionStateWithDiverge() throws Exception { assertNull(actionStateStore.get(TEST_KEY, 4L, testAction, testEvent)); } + @Test + void testGetRetainsUnparseableKey() throws Exception { + String flinkKey = "user_123"; + String stateKey = ActionStateUtil.generateKey(flinkKey, 1L, testAction, testEvent); + actionStates.put(stateKey, testActionState); + + assertThat(actionStateStore.get(flinkKey, 2L, testAction, testEvent)).isNull(); + assertThat(actionStates).containsKey(stateKey); + } + @Test void testRecoveryMarker() throws Exception { // Test getting initial recovery marker @@ -240,16 +260,16 @@ void testPruneStateSendsTombstonesWithCorrectKeys() throws Exception { @Test void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws Exception { - // Arrange - agent key "a" seq 1 yields state key "a_1__", which is a - // prefix match for pruning agent key "a_1" + // Arrange - Flink key "a" seq 1 yields state key "a_1__", which is a + // prefix match for pruning Flink 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 + // Act - prune a DIFFERENT Flink 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 + // Assert - Flink key "a"'s state is untouched and no tombstones were sent assertThat(actionStates).containsKey(otherKeyState); assertThat(mockProducer.history()).isEmpty(); } @@ -286,17 +306,42 @@ public synchronized Future send( String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent); actionStates.put(stateKey, testActionState); - // Act - should not throw despite the async send failure - actionStateStore.pruneState(TEST_KEY, 1L); - - // Assert - in-memory entry is still evicted regardless of tombstone delivery - assertThat(failureCallbackInvoked).isTrue(); - assertThat(actionStates).doesNotContainKey(stateKey); + TestAppender appender = new TestAppender("KafkaActionStateStoreTestAppender"); + appender.start(); + LoggerContext loggerContext = (LoggerContext) LogManager.getContext(false); + Configuration loggerConfiguration = loggerContext.getConfiguration(); + String loggerName = KafkaActionStateStore.class.getName(); + LoggerConfig loggerConfig = loggerConfiguration.getLoggerConfig(loggerName); + boolean addedLoggerConfig = !loggerConfig.getName().equals(loggerName); + if (addedLoggerConfig) { + loggerConfig = new LoggerConfig(loggerName, Level.WARN, false); + loggerConfiguration.addLogger(loggerName, loggerConfig); + } + loggerConfig.addAppender(appender, Level.WARN, null); + loggerContext.updateLoggers(); + + try { + // Act - should not throw despite the async send failure + actionStateStore.pruneState(TEST_KEY, 1L); + + // Assert - the failure is reported and the in-memory entry is still evicted + assertThat(failureCallbackInvoked).isTrue(); + assertThat(appender.getMessages()) + .anyMatch(message -> message.contains("Failed to send tombstone record")); + assertThat(actionStates).doesNotContainKey(stateKey); + } finally { + loggerConfig.removeAppender(appender.getName()); + if (addedLoggerConfig) { + loggerConfiguration.removeLogger(loggerName); + } + appender.stop(); + loggerContext.updateLoggers(); + } } @Test void testPruneStateSkipsUnparseableKeys() throws Exception { - // Arrange - an agent key that itself contains the "_" separator (e.g. "user_123") + // Arrange - a Flink key that itself contains the "_" separator (e.g. "user_123") // produces a state key with 5 "_"-separated parts once seqNum and the two UUIDs are // appended, which ActionStateUtil.parseKey cannot split into exactly 4 parts actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); @@ -434,6 +479,24 @@ void testRebuildStateRemovesTombstonedKeys() throws Exception { assertThat(actionStates.get(key2)).isEqualTo(testActionState); } + private static class TestAppender extends AbstractAppender { + + private final List messages = Collections.synchronizedList(new ArrayList<>()); + + private TestAppender(String name) { + super(name, null, PatternLayout.newBuilder().withPattern("%msg").build(), true, null); + } + + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + + private List getMessages() { + return messages; + } + } + /** Contract: the consumer is closed even when closing the producer throws. */ @Test @SuppressWarnings("unchecked") From d09d94c7e0f79475e99d0c4a4b522fb998022d2f Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 01:05:54 -0700 Subject: [PATCH 13/22] [runtime][java] Protect Fluss cleanup from key collisions --- .../actionstate/FlussActionStateStore.java | 22 ++++++++++--------- .../FlussActionStateStoreTest.java | 20 +++++++++++++++++ 2 files changed, 32 insertions(+), 10 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java index 0a20fe2bd..c7d491942 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java @@ -216,12 +216,12 @@ public void put(Object key, long seqNum, Action action, Event event, ActionState @Override public ActionState get(Object key, long seqNum, Action action, Event event) throws Exception { String stateKey = generateKey(key, seqNum, action, event); - String keyPrefix = key.toString() + "_"; + String keyStr = key.toString(); - boolean hasDivergence = checkDivergence(key.toString(), seqNum); + boolean hasDivergence = checkDivergence(keyStr, seqNum); if (!actionStates.containsKey(stateKey) || hasDivergence) { - removeStateEntries(keyPrefix, stateSeqNum -> stateSeqNum > seqNum); + removeStateEntries(keyStr, stateSeqNum -> stateSeqNum > seqNum); } ActionState state = actionStates.get(stateKey); @@ -237,10 +237,11 @@ private boolean checkDivergence(String key, long seqNum) { } /** - * Removes cached state entries whose key starts with {@code keyPrefix} and whose parsed - * sequence number satisfies {@code seqNumFilter}. + * Removes cached state entries belonging to {@code key} whose parsed sequence number satisfies + * {@code seqNumFilter}. */ - private void removeStateEntries(String keyPrefix, LongPredicate seqNumFilter) { + private void removeStateEntries(String key, LongPredicate seqNumFilter) { + String keyPrefix = key + "_"; actionStates .entrySet() .removeIf( @@ -250,10 +251,11 @@ private void removeStateEntries(String keyPrefix, LongPredicate seqNumFilter) { } try { List parts = ActionStateUtil.parseKey(entry.getKey()); - if (parts.size() >= 2) { - long stateSeqNum = Long.parseLong(parts.get(1)); - return seqNumFilter.test(stateSeqNum); + if (!parts.get(0).equals(key)) { + return false; } + long stateSeqNum = Long.parseLong(parts.get(1)); + return seqNumFilter.test(stateSeqNum); } catch (Exception e) { LOG.warn("Failed to parse state key: {}", entry.getKey(), e); } @@ -486,7 +488,7 @@ public Object getRecoveryMarker() { @Override public void pruneState(Object key, long seqNum) { LOG.debug("Pruning in-memory state for key: {} up to seqNum: {}", key, seqNum); - removeStateEntries(key.toString() + "_", stateSeqNum -> stateSeqNum <= seqNum); + removeStateEntries(key.toString(), stateSeqNum -> stateSeqNum <= seqNum); } @Override diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java index b73b78a2a..50fff1c13 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java @@ -132,6 +132,26 @@ void testPruneStateRetainsKeyContainingUnderscore() throws Exception { assertThat(actionStates).containsKey(stateKey); } + @Test + void testPruneStateDoesNotPruneOtherKeysWithMatchingPrefix() throws Exception { + String otherKeyState = ActionStateUtil.generateKey("a", 1L, testAction, testEvent); + actionStates.put(otherKeyState, testActionState); + + store.pruneState("a_1", 10L); + + assertThat(actionStates).containsKey(otherKeyState); + } + + @Test + void testGetDoesNotEvictOtherKeysWithMatchingPrefix() throws Exception { + String otherKeyState = ActionStateUtil.generateKey("a", 1L, testAction, testEvent); + actionStates.put(otherKeyState, testActionState); + + assertThat(store.get("a_1", 0L, testAction, testEvent)).isNull(); + + assertThat(actionStates).containsKey(otherKeyState); + } + // ==================== rebuildState tests ==================== @Test From a8bf18a6a82f3b32deb51a7184f3e70a85f72d61 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 01:06:10 -0700 Subject: [PATCH 14/22] [runtime][java] Strengthen unparseable Kafka key test --- .../agents/runtime/actionstate/KafkaActionStateStoreTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index c87749d89..fa68c608a 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -168,10 +168,10 @@ void testGetActionStateWithDiverge() throws Exception { @Test void testGetRetainsUnparseableKey() throws Exception { String flinkKey = "user_123"; - String stateKey = ActionStateUtil.generateKey(flinkKey, 1L, testAction, testEvent); + String stateKey = ActionStateUtil.generateKey(flinkKey, 3L, testAction, testEvent); actionStates.put(stateKey, testActionState); - assertThat(actionStateStore.get(flinkKey, 2L, testAction, testEvent)).isNull(); + assertThat(actionStateStore.get(flinkKey, 1L, testAction, testEvent)).isNull(); assertThat(actionStates).containsKey(stateKey); } From fca59c2271b6a832a35a10065606c4f5d9aabf51 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 01:06:27 -0700 Subject: [PATCH 15/22] [runtime][java] Quiet repeated Kafka key parse failures --- .../runtime/actionstate/KafkaActionStateStore.java | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index fc2df4bfd..2877581c9 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -181,13 +181,11 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro // the requested seqNum return stateSeqNum > seqNum; } - } catch (IllegalArgumentException e) { - LOG.warn( - "Cannot parse state key: {}. The entry cannot be " - + "considered for divergence cleanup and will " - + "be retained.", - entry.getKey(), - e); + } catch (IllegalArgumentException ignored) { + LOG.debug( + "Retaining unparseable state key during divergence " + + "cleanup: {}", + entry.getKey()); } return false; }); From 8b8db63c774ba347eb75c2c82c672102936de9c0 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 01:06:46 -0700 Subject: [PATCH 16/22] [api][runtime][docs] Clarify action state key limitation --- .../flink/agents/api/configuration/AgentConfigOptions.java | 6 +++--- docs/content/docs/operations/configuration.md | 2 +- .../flink/agents/runtime/actionstate/ActionStateStore.java | 7 +++++-- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index 77ceea8ef..08cd300c6 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -92,9 +92,9 @@ public enum ConditionEvaluationFailureStrategy { * checkpoints or savepoints, or if re-executing actions is acceptable. * *

Also note: a Flink key that itself contains the {@code _} character (e.g. {@code - * user_123}) is never pruned or tombstoned regardless of this setting. Durable action state - * stores join key parts with an unescaped {@code _}, so such a key fails to parse back into its - * parts and its state is retained in memory and in backend storage instead. + * user_123}) cannot be parsed safely during pruning or divergence cleanup. Its cached state is + * retained, so stale higher-sequence state may survive a detected divergence. Pruning also + * emits no tombstones for that key, so its Kafka topic records remain. */ public static final ConfigOption KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false); diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index 15e71eb28..b2ed07453 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -170,7 +170,7 @@ The eight `memory.generate-event*` options have no raw `ConfigOption` default. W |------------------------------|------------------|---------|------------------------------------------------------------------------------------------| | `actionStateStoreBackend` | (none) | String | The backend for action state store. Supported values: `"kafka"`, `"fluss"`. | -Durable action state stores currently join raw Flink keys and other key parts with an unescaped `_`. Flink keys containing `_` cannot be parsed safely during pruning, so both Kafka and Fluss retain their state in memory and backend storage. Kafka also emits no tombstones for those keys. +Durable action state stores currently join raw Flink keys and other key parts with an unescaped `_`. Flink keys containing `_` cannot be parsed safely during pruning or divergence cleanup. Both backends retain the affected in-memory entries, so stale higher-sequence state may survive a detected divergence. Kafka also retains the corresponding topic records and emits no tombstones for these keys. The Fluss log is append-only, so its physical cleanup always relies on Fluss retention configuration. #### Kafka-based Action State Store diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 3b546c572..2da4bee88 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -86,8 +86,11 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) * the recovery trade-off of advancing beyond that boundary. * *

The current durable stores encode raw Flink keys using an unescaped {@code _} separator. - * Flink keys containing {@code _} therefore cannot be parsed safely during pruning and are - * retained in both the in-memory cache and backend storage. + * Flink keys containing {@code _} therefore cannot be parsed safely during pruning or + * divergence cleanup. Affected entries remain in the in-memory cache, so stale higher-sequence + * state may survive a detected divergence. Kafka pruning also retains their topic records and + * emits no tombstones. The Fluss log is append-only and its physical cleanup always relies on + * Fluss retention configuration. * * @param key the key whose state should be pruned * @param seqNum the sequence number up to which the state should be pruned From a6c68e39f61e1e1acdf20491ca85f5b8d3aa12d8 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 01:07:03 -0700 Subject: [PATCH 17/22] [docs] Restore exactly-once action wording --- docs/content/docs/operations/deployment.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md index 5610500fe..bfb0e6695 100644 --- a/docs/content/docs/operations/deployment.md +++ b/docs/content/docs/operations/deployment.md @@ -92,7 +92,7 @@ After recovery from a checkpoint, Flink Agents reprocess events that arrived aft ### Exactly-Once Action Consistency -To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. After recovering from a checkpoint, Flink Agents consult the external store and reuse completed action state when its backing record remains available. This prevents re-execution for checkpoints supported by the store's retained recovery history. +To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. This guarantees each action is executed exactly once after recovering from a checkpoint. When a completed Action is reused during recovery, its stored output Events keep their original Event IDs, while their lineage is rebound to the Event that triggers the reused Action in the recovered execution. From 298e44b60da844b33686d39fdbf3a3b7cb7849cb Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 23:53:29 -0700 Subject: [PATCH 18/22] [runtime][java] Scope Kafka divergence cleanup by key --- .../runtime/actionstate/KafkaActionStateStore.java | 5 +++-- .../runtime/actionstate/KafkaActionStateStoreTest.java | 10 ++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index 2877581c9..741d327d5 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -157,6 +157,7 @@ public void put(Object key, long seqNum, Action action, Event event, ActionState @Override public ActionState get(Object key, long seqNum, Action action, Event event) throws Exception { String stateKey = generateKey(key, seqNum, action, event); + String keyStr = key.toString(); LOG.debug( "Looking up action state: key={}, seqNum={}, stateKey={}, cachedStates={}", @@ -165,7 +166,7 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro stateKey, actionStates.keySet()); - boolean hasDivergence = checkDivergence(key.toString(), seqNum); + boolean hasDivergence = checkDivergence(keyStr, seqNum); if (!actionStates.containsKey(stateKey) || hasDivergence) { actionStates @@ -175,7 +176,7 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro // Extract key and sequence number from the state key try { List parts = ActionStateUtil.parseKey(entry.getKey()); - if (parts.size() >= 2) { + if (parts.get(0).equals(keyStr)) { long stateSeqNum = Long.parseLong(parts.get(1)); // clean up any states with sequence number greater than // the requested seqNum diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index fa68c608a..8c4a9a2fd 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -175,6 +175,16 @@ void testGetRetainsUnparseableKey() throws Exception { assertThat(actionStates).containsKey(stateKey); } + @Test + void testGetDoesNotEvictOtherKeysWithMatchingPrefix() throws Exception { + String otherKeyState = ActionStateUtil.generateKey("a", 1L, testAction, testEvent); + actionStates.put(otherKeyState, testActionState); + + assertThat(actionStateStore.get("a_1", 0L, testAction, testEvent)).isNull(); + + assertThat(actionStates).containsKey(otherKeyState); + } + @Test void testRecoveryMarker() throws Exception { // Test getting initial recovery marker From 53ca82629842cd00b61a97a7fc68ba460d6ad331 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 23:54:32 -0700 Subject: [PATCH 19/22] [runtime][java] Quiet retained Fluss state keys --- .../agents/runtime/actionstate/FlussActionStateStore.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java index c7d491942..7ab09a563 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java @@ -256,8 +256,10 @@ private void removeStateEntries(String key, LongPredicate seqNumFilter) { } long stateSeqNum = Long.parseLong(parts.get(1)); return seqNumFilter.test(stateSeqNum); - } catch (Exception e) { - LOG.warn("Failed to parse state key: {}", entry.getKey(), e); + } catch (Exception ignored) { + LOG.debug( + "Retaining unparseable state key during cleanup: {}", + entry.getKey()); } return false; }); From 3435e079bc441b739ce0c045cb951088cafdf15c Mon Sep 17 00:00:00 2001 From: rob-9 Date: Sun, 23 Aug 2026 23:55:12 -0700 Subject: [PATCH 20/22] [docs] Restore action recovery explanation --- docs/content/docs/operations/deployment.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/content/docs/operations/deployment.md b/docs/content/docs/operations/deployment.md index bfb0e6695..ce9a14923 100644 --- a/docs/content/docs/operations/deployment.md +++ b/docs/content/docs/operations/deployment.md @@ -92,7 +92,7 @@ After recovery from a checkpoint, Flink Agents reprocess events that arrived aft ### Exactly-Once Action Consistency -To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. This guarantees each action is executed exactly once after recovering from a checkpoint. +To ensure exactly-once action consistency, you must configure an external action state store. Flink Agents record action state in this store on a per-action basis. After recovering from a checkpoint, Flink Agents consult the external store and will not re-execute actions that were already completed. This guarantees each action is executed exactly once after recovering from a checkpoint. When a completed Action is reused during recovery, its stored output Events keep their original Event IDs, while their lineage is rebound to the Event that triggers the reused Action in the recovered execution. From ad615173554f5693a47f413bb93989b27c0d7913 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 24 Aug 2026 00:00:18 -0700 Subject: [PATCH 21/22] [runtime][java] Scope divergence detection by key --- .../runtime/actionstate/FlussActionStateStore.java | 10 ++++++++++ .../runtime/actionstate/KafkaActionStateStore.java | 10 ++++++++++ .../actionstate/FlussActionStateStoreTest.java | 14 ++++++++++++++ .../actionstate/KafkaActionStateStoreTest.java | 14 ++++++++++++++ 4 files changed, 48 insertions(+) diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java index 7ab09a563..a597895c2 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java @@ -232,6 +232,16 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro private boolean checkDivergence(String key, long seqNum) { return actionStates.keySet().stream() .filter(k -> k.startsWith(key + "_" + seqNum + "_")) + .filter( + stateKey -> { + try { + List parts = ActionStateUtil.parseKey(stateKey); + return parts.get(0).equals(key) + && Long.parseLong(parts.get(1)) == seqNum; + } catch (IllegalArgumentException ignored) { + return false; + } + }) .count() > 1; } diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java index 741d327d5..db2357d70 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java @@ -205,6 +205,16 @@ public ActionState get(Object key, long seqNum, Action action, Event event) thro private boolean checkDivergence(String key, long seqNum) { return actionStates.keySet().stream() .filter(k -> k.startsWith(key + "_" + seqNum + "_")) + .filter( + stateKey -> { + try { + List parts = ActionStateUtil.parseKey(stateKey); + return parts.get(0).equals(key) + && Long.parseLong(parts.get(1)) == seqNum; + } catch (IllegalArgumentException ignored) { + return false; + } + }) .count() > 1; } diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java index 50fff1c13..ac91fa850 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java @@ -152,6 +152,20 @@ void testGetDoesNotEvictOtherKeysWithMatchingPrefix() throws Exception { assertThat(actionStates).containsKey(otherKeyState); } + @Test + void testGetDoesNotTreatOtherKeyPrefixAsDivergence() throws Exception { + String currentState = ActionStateUtil.generateKey("a", 1L, testAction, testEvent); + String futureState = ActionStateUtil.generateKey("a", 2L, testAction, testEvent); + String collidingState = ActionStateUtil.generateKey("a_1", 0L, testAction, testEvent); + actionStates.put(currentState, testActionState); + actionStates.put(futureState, testActionState); + actionStates.put(collidingState, testActionState); + + assertThat(store.get("a", 1L, testAction, testEvent)).isEqualTo(testActionState); + + assertThat(actionStates).containsKey(futureState); + } + // ==================== rebuildState tests ==================== @Test diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 8c4a9a2fd..1cc4759a2 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -185,6 +185,20 @@ void testGetDoesNotEvictOtherKeysWithMatchingPrefix() throws Exception { assertThat(actionStates).containsKey(otherKeyState); } + @Test + void testGetDoesNotTreatOtherKeyPrefixAsDivergence() throws Exception { + String currentState = ActionStateUtil.generateKey("a", 1L, testAction, testEvent); + String futureState = ActionStateUtil.generateKey("a", 2L, testAction, testEvent); + String collidingState = ActionStateUtil.generateKey("a_1", 0L, testAction, testEvent); + actionStates.put(currentState, testActionState); + actionStates.put(futureState, testActionState); + actionStates.put(collidingState, testActionState); + + assertThat(actionStateStore.get("a", 1L, testAction, testEvent)).isEqualTo(testActionState); + + assertThat(actionStates).containsKey(futureState); + } + @Test void testRecoveryMarker() throws Exception { // Test getting initial recovery marker From 69fa8562c0a09bc30e1deaf32a990fce1482eae4 Mon Sep 17 00:00:00 2001 From: rob-9 Date: Mon, 24 Aug 2026 09:56:00 -0700 Subject: [PATCH 22/22] [runtime][java] Support underscores in action state keys --- .../api/configuration/AgentConfigOptions.java | 5 --- docs/content/docs/operations/configuration.md | 2 -- .../ActionStateKeyPartitioner.java | 7 ++-- .../runtime/actionstate/ActionStateStore.java | 7 ---- .../runtime/actionstate/ActionStateUtil.java | 18 +++++++++-- .../ActionStateKeyPartitionerTest.java | 32 +++++++++++++------ .../actionstate/ActionStateUtilTest.java | 32 +++++++++++++++---- .../FlussActionStateStoreTest.java | 14 ++++++-- .../KafkaActionStateStoreTest.java | 18 +++++------ 9 files changed, 86 insertions(+), 49 deletions(-) diff --git a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java index 08cd300c6..782b4e486 100644 --- a/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java +++ b/api/src/main/java/org/apache/flink/agents/api/configuration/AgentConfigOptions.java @@ -90,11 +90,6 @@ public enum ConditionEvaluationFailureStrategy { * written after that restore point, erasing action state the replay still needs and causing * already completed actions to re-execute. Enable only if the job never restores from earlier * checkpoints or savepoints, or if re-executing actions is acceptable. - * - *

Also note: a Flink key that itself contains the {@code _} character (e.g. {@code - * user_123}) cannot be parsed safely during pruning or divergence cleanup. Its cached state is - * retained, so stale higher-sequence state may survive a detected divergence. Pruning also - * emits no tombstones for that key, so its Kafka topic records remain. */ public static final ConfigOption KAFKA_ACTION_STATE_TOMBSTONE_ENABLED = new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false); diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md index b2ed07453..c38b8eddf 100644 --- a/docs/content/docs/operations/configuration.md +++ b/docs/content/docs/operations/configuration.md @@ -170,8 +170,6 @@ The eight `memory.generate-event*` options have no raw `ConfigOption` default. W |------------------------------|------------------|---------|------------------------------------------------------------------------------------------| | `actionStateStoreBackend` | (none) | String | The backend for action state store. Supported values: `"kafka"`, `"fluss"`. | -Durable action state stores currently join raw Flink keys and other key parts with an unescaped `_`. Flink keys containing `_` cannot be parsed safely during pruning or divergence cleanup. Both backends retain the affected in-memory entries, so stale higher-sequence state may survive a detected divergence. Kafka also retains the corresponding topic records and emits no tombstones for these keys. The Fluss log is append-only, so its physical cleanup always relies on Fluss retention configuration. - #### Kafka-based Action State Store Here are the configuration options for Kafka-based Action State Store. diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitioner.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitioner.java index 7fc8b175b..e7dad2f96 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitioner.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitioner.java @@ -45,10 +45,9 @@ public int partition( throw new IllegalArgumentException("Key format is invalid"); } - if ("".equalsIgnoreCase(keyParts[0])) { - throw new IllegalArgumentException("First part of the key cannot be empty"); - } - + // Preserve the existing first-segment partitioning for compatibility with records already + // in the topic. Keys beginning with '_' previously could not be written; accepting the + // empty first segment makes them usable without moving any existing records. return MathUtils.murmurHash(keyParts[0].hashCode()) % numPartitions; } diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java index 2da4bee88..b2f603c73 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateStore.java @@ -85,13 +85,6 @@ void put(Object key, long seqNum, Action action, Event event, ActionState state) * enforce a recovery boundary that protects every supported restore point or clearly document * the recovery trade-off of advancing beyond that boundary. * - *

The current durable stores encode raw Flink keys using an unescaped {@code _} separator. - * Flink keys containing {@code _} therefore cannot be parsed safely during pruning or - * divergence cleanup. Affected entries remain in the in-memory cache, so stale higher-sequence - * state may survive a detected divergence. Kafka pruning also retains their topic records and - * emits no tombstones. The Fluss log is append-only and its physical cleanup always relies on - * Fluss retention configuration. - * * @param key the key whose state should be pruned * @param seqNum the sequence number up to which the state should be pruned */ diff --git a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java index 24d849bac..7b4a505e0 100644 --- a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java +++ b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtil.java @@ -55,11 +55,23 @@ public static String generateKey( generateUUIDForAction(action)); } + /** + * Parses an action-state key from its fixed sequence/event/action suffix. The Flink key itself + * may contain the separator. + */ public static List parseKey(String key) { Preconditions.checkNotNull(key, "key cannot be null."); - String[] parts = key.split(KEY_SEPARATOR); - Preconditions.checkArgument(parts.length == 4, "Invalid key format."); - return List.of(parts); + + int actionSeparator = key.lastIndexOf(KEY_SEPARATOR); + int eventSeparator = key.lastIndexOf(KEY_SEPARATOR, actionSeparator - 1); + int sequenceSeparator = key.lastIndexOf(KEY_SEPARATOR, eventSeparator - 1); + Preconditions.checkArgument(sequenceSeparator >= 0, "Invalid key format."); + + return List.of( + key.substring(0, sequenceSeparator), + key.substring(sequenceSeparator + 1, eventSeparator), + key.substring(eventSeparator + 1, actionSeparator), + key.substring(actionSeparator + 1)); } private static String generateUUIDForEvent(Event event) throws IOException { diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitionerTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitionerTest.java index 3f45bcf77..24fbac1eb 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitionerTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateKeyPartitionerTest.java @@ -99,6 +99,19 @@ void testSameKeyFirstPartConsistentPartitioning() { assertEquals(partition1, partition3); } + @Test + void testSameKeyContainingSeparatorUsesConsistentPartition() { + String key1 = "user_123_1_action1_event1"; + String key2 = "user_123_2_action2_event2"; + + int partition1 = + partitioner.partition(TEST_TOPIC, key1, key1.getBytes(), null, null, cluster); + int partition2 = + partitioner.partition(TEST_TOPIC, key2, key2.getBytes(), null, null, cluster); + + assertEquals(partition1, partition2); + } + @Test void testNullKeyThrowsException() { IllegalArgumentException exception = @@ -155,15 +168,16 @@ void testInvalidKeyFormatThrowsException() { } @Test - void testEmptyFirstKeyPartThrowException() { - String invalidKey = "_1_action_event"; - IllegalArgumentException exception = - assertThrows( - IllegalArgumentException.class, - () -> - partitioner.partition( - TEST_TOPIC, invalidKey, null, null, null, cluster)); - assertEquals("First part of the key cannot be empty", exception.getMessage()); + void testKeyStartingWithSeparatorUsesConsistentPartition() { + String key1 = "_user_1_action_event"; + String key2 = "_user_2_action_event"; + + int partition1 = + partitioner.partition(TEST_TOPIC, key1, key1.getBytes(), null, null, cluster); + int partition2 = + partitioner.partition(TEST_TOPIC, key2, key2.getBytes(), null, null, cluster); + + assertEquals(partition1, partition2); } @Test diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java index 2a90c1f15..c396a79f7 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/ActionStateUtilTest.java @@ -136,6 +136,31 @@ public void testParseKeyRoundTrip() throws Exception { assertEquals(String.valueOf(seqNum), parsedParts.get(1)); } + @Test + public void testParseKeyRoundTripWithSeparatorInFlinkKey() throws Exception { + Object originalKey = "_user_123_with_separator_"; + Action action = new NoOpAction("round-trip-action"); + InputEvent inputEvent = new InputEvent("round-trip-input"); + long seqNum = 456; + + String generatedKey = ActionStateUtil.generateKey(originalKey, seqNum, action, inputEvent); + List parsedParts = ActionStateUtil.parseKey(generatedKey); + + assertEquals(originalKey.toString(), parsedParts.get(0)); + assertEquals(String.valueOf(seqNum), parsedParts.get(1)); + } + + @Test + public void testParsePreviouslyGeneratedKeyWithSeparatorInFlinkKey() { + String eventId = "00000000-0000-0000-0000-000000000001"; + String actionId = "00000000-0000-0000-0000-000000000002"; + String persistedKey = "user_123_456_" + eventId + "_" + actionId; + + assertEquals( + List.of("user_123", "456", eventId, actionId), + ActionStateUtil.parseKey(persistedKey)); + } + @Test public void testParseKeyWithNullInput() { assertThrows( @@ -154,13 +179,6 @@ public void testParseKeyWithInvalidFormat() { ActionStateUtil.parseKey("only_three_parts"); }); - // Test with too many parts - assertThrows( - IllegalArgumentException.class, - () -> { - ActionStateUtil.parseKey("one_two_three_four_five_six"); - }); - // Test with empty string assertThrows( IllegalArgumentException.class, diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java index ac91fa850..8d82277f0 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java @@ -122,14 +122,24 @@ void testGetTriggersDivergenceCleanup() throws Exception { } @Test - void testPruneStateRetainsKeyContainingUnderscore() throws Exception { + void testPruneStateSupportsKeysContainingUnderscore() throws Exception { String agentKey = "user_123"; String stateKey = ActionStateUtil.generateKey(agentKey, 1L, testAction, testEvent); actionStates.put(stateKey, testActionState); store.pruneState(agentKey, 1L); - assertThat(actionStates).containsKey(stateKey); + assertThat(actionStates).doesNotContainKey(stateKey); + } + + @Test + void testGetCleansFutureStateForKeyContainingUnderscore() throws Exception { + String flinkKey = "user_123"; + String stateKey = ActionStateUtil.generateKey(flinkKey, 3L, testAction, testEvent); + actionStates.put(stateKey, testActionState); + + assertThat(store.get(flinkKey, 1L, testAction, testEvent)).isNull(); + assertThat(actionStates).doesNotContainKey(stateKey); } @Test diff --git a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java index 1cc4759a2..02a3ebb72 100644 --- a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java +++ b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java @@ -166,13 +166,13 @@ void testGetActionStateWithDiverge() throws Exception { } @Test - void testGetRetainsUnparseableKey() throws Exception { + void testGetCleansFutureStateForKeyContainingUnderscore() throws Exception { String flinkKey = "user_123"; String stateKey = ActionStateUtil.generateKey(flinkKey, 3L, testAction, testEvent); actionStates.put(stateKey, testActionState); assertThat(actionStateStore.get(flinkKey, 1L, testAction, testEvent)).isNull(); - assertThat(actionStates).containsKey(stateKey); + assertThat(actionStates).doesNotContainKey(stateKey); } @Test @@ -364,21 +364,19 @@ public synchronized Future send( } @Test - void testPruneStateSkipsUnparseableKeys() throws Exception { - // Arrange - a Flink key that itself contains the "_" separator (e.g. "user_123") - // produces a state key with 5 "_"-separated parts once seqNum and the two UUIDs are - // appended, which ActionStateUtil.parseKey cannot split into exactly 4 parts + void testPruneStateSupportsKeysContainingUnderscore() throws Exception { actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); String agentKey = "user_123"; String stateKey = ActionStateUtil.generateKey(agentKey, 1L, testAction, testEvent); actionStates.put(stateKey, testActionState); - // Act - should not throw despite the unparseable key actionStateStore.pruneState(agentKey, 10L); - // Assert - the unparseable entry is retained, and no tombstone was sent for it - assertThat(actionStates).containsKey(stateKey); - assertThat(mockProducer.history()).isEmpty(); + assertThat(actionStates).doesNotContainKey(stateKey); + assertThat(mockProducer.history()) + .extracting(ProducerRecord::key) + .containsExactly(stateKey); + assertThat(mockProducer.history()).extracting(ProducerRecord::value).containsOnlyNulls(); } @Test