[runtime] send Kafka tombstones on state pruning for durable log compaction by rob-9 · Pull Request #885 · apache/flink-agents

rob-9 · 2026-07-08T00:36:20Z

Addresses #691

Purpose of change

KafkaActionStateStore.pruneState() only removed entries from the in-memory cache. the underlying Kafka topic was never updated, so:

the topic grew unbounded, meaning Kafka log compaction had no tombstones to trigger key deletion
after a restart, rebuildState() replayed pruned records back into memory, resurrecting state that should have been discarded

This PR sends null-valued records to the state topic during pruning, and updates rebuildState() to skip tombstoned keys during replay instead of inserting them into the cache. Also fixes an NPE in the error-logging path that called record.value().toString() on a null value.

Tests

mvn --batch-mode --no-transfer-progress -pl runtime -am -Dtest=KafkaActionStateStoreTest test

testPruneState verifies tombstones are produced and in-memory entries are evicted
testPruneStateSendsTombstonesWithCorrectKeys asserts exact composite keys on produced tombstones
testPruneStateNoMatchingKeys verifies no tombstones sent when no keys match
testPruneStateWithNullProducer verifies graceful degradation when producer is null

API

No public API changes.

Documentation

doc-not-needed

wenjin272 · 2026-07-13T09:02:37Z

@joeyutong Can you take a look at this PR?

joeyutong · 2026-07-14T04:00:13Z

Thanks for addressing the growth of Kafka action-state records. However, writing tombstones unconditionally may break recovery from older checkpoints or savepoints.

For example, if C0 is followed by action state S and then C1 tombstones S, restoring C0 will replay both S and the tombstone. rebuildState() removes S, so the action may execute again. This happens even when Kafka compaction is disabled.

Could tombstone emission be opt-in, for example through kafkaActionStateTombstoneEnabled defaulting to false?

rob-9 · 2026-07-14T18:51:41Z

Could tombstone emission be opt-in, for example through kafkaActionStateTombstoneEnabled defaulting to false?

Yeah, agreed.

Can we make cleanup safe by default though? Maybe we could stamp tombstones with the checkpoint id that triggered the prune; rebuildState() applies one only if id <= restored checkpoint, so restoring C0 skips C1's tombstones. But this would only protect the record until compaction deletes it.

This can be a follow-up. I'll add the opt-in flag to this PR for now.

- 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

joeyutong · 2026-07-15T03:23:44Z

Thanks for raising the question of whether cleanup can be safe by default. A checkpoint-aligned, user-controlled cleanup path may be a safer general solution:

Persist and expose the mapping from each checkpoint to its effective Kafka recovery marker—the per-partition minimum across all subtasks.
Leave the irreversible cleanup decision to users. Based on their recovery requirements, they can choose the oldest checkpoint to retain and delete records before its marker offsets through Kafka's deleteRecords API.
During rebuildState(), validate the requested offsets against Kafka's current beginning offsets. If a recovery offset is older than the available data, fail fast with a clear error instead of silently rebuilding incomplete state.

Since rebuildState() already starts from the recovery marker, the same marker is a natural boundary for physical cleanup. This aligns cleanup with the actual recovery semantics without automatically invalidating older checkpoints. Once users advance that boundary, fail-fast validation makes any incompatible recovery attempt explicit.

This could be explored as a follow-up.

…neState - 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.

joeyutong · 2026-07-16T03:56:08Z

+     * 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<Boolean> KAFKA_ACTION_STATE_TOMBSTONE_ENABLED =


Could we also add the corresponding option to python/flink_agents/api/core_options.py? The cross-language option parity check currently fails because this field exists only on the Java side.

joeyutong · 2026-07-16T03:56:19Z

+        actionStateStore = tombstoneEnabledStore(actionStates, mockProducer);
+        String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent);
+        actionStates.put(stateKey, testActionState);
+        mockProducer.errorNext(new RuntimeException("simulated broker failure"));


mockProducer uses autoComplete=true, so errorNext() is called before any pending completion exists and does not make the subsequent send() fail. This test therefore exercises the successful send path. Could we inject an exception through the callback so the async failure path is actually covered?

weiqingy · 2026-07-17T07:17:47Z

+            }
+            try {
+                List<String> parts = ActionStateUtil.parseKey(stateKey);
+                if (parts.get(0).equals(keyStr) && Long.parseLong(parts.get(1)) <= seqNum) {


The collision you caught here feels like a thread worth pulling one turn further — I think the same root cause has a second face that this check can't reach.

Both this guard and the collision it fixes trace back to generateKey joining on _ while embedding the raw key unescaped (ActionStateUtil.java:44-56). That's what let pruning key a_1 match agent key a's state key a_1_<uuid>_<uuid>, and the equality check handles that case correctly.

The sibling case is when the agent key itself contains _. For an ordinary keyBy("user_123") the state key is user_123_1_<uuid>_<uuid>, so split("_") yields 5 parts (UUIDs use hyphens) and parseKey's checkArgument(parts.length == 4) throws (ActionStateUtil.java:58-63). This line never gets evaluated, and the entry lands in the catch on line 300 — so it's never evicted from the cache, never tombstoned, and never reclaimed by compaction. For any key containing _, the growth this PR targets would stay as it is. FlussActionStateStore.removeStateEntries catches Exception and skips the same way (FlussActionStateStore.java:242-261).

Worth saying that catching IllegalArgumentException on line 300 is an improvement in its own right — the old code caught only NumberFormatException, so a bad part count escaped pruneState and could take down notifyCheckpointComplete. Warn-and-retain is strictly better.

Escaping the separator properly would touch three files, which feels like more than this PR signed up for — would you rather note the limitation in the option's javadoc and file a follow-up? Or do you see a narrower angle I'm missing?

And on the test: testPruneStateSkipsUnparseableKeys pins this branch with TEST_KEY + "_1_onlythreeparts". Any appetite for a realistic fixture like "user_123" — same branch, but it'd document the case that actually reaches it?

weiqingy · 2026-07-17T07:17:47Z

-                        });
+                            });
+                }
+                producer.flush();


Does this flush earn its place? The callback just above already reports failures, the eviction below runs regardless of the outcome, and ordering looks covered — put() flushes before returning (line 146), so a tombstone is always sent after its record is acked.

Curious whether there's a case it's protecting that I'm not seeing.

weiqingy · 2026-07-17T07:17:47Z

+     * checkpoints or savepoints, or if re-executing actions is acceptable.
+     */
+    public static final ConfigOption<Boolean> KAFKA_ACTION_STATE_TOMBSTONE_ENABLED =
+            new ConfigOption<>("kafkaActionStateTombstoneEnabled", Boolean.class, false);


Defaulting to false reads as the right call to me, and I think for a sharper reason than the javadoc gives itself credit for: with cleanup.policy=compact on the topic (KafkaActionStateStore.java:396), a tombstone is a durable delete instruction to the log cleaner — so the older-checkpoint erasure you documented above isn't something the replay side could have absorbed. Gating emission is the lever that actually controls it. A safe default-on would seem to need emission gated on the oldest still-restorable checkpoint, and I can't see a cheap way to know that here, so I'm not suggesting you flip it.

That does leave #691's original ask — the unbounded growth — unaddressed while the option is off. Where do you see this landing: is opt-in the endpoint, or would a follow-up be worth filing so the issue has somewhere to point?

weiqingy · 2026-07-17T07:17:47Z

 | `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. |


Small one: this row and the new public AgentConfigOptions.KAFKA_ACTION_STATE_TOMBSTONE_ENABLED are both in the diff, while the description checks doc-not-needed and says "No public API changes". It also still describes the unconditional design rather than the opt-in one that shipped. Mind refreshing it before merge?

weiqingy · 2026-07-17T07:17:47Z

+
+        // Assert - tombstones should have been sent to Kafka
+        var history = mockProducer.history();
+        assertThat(history).hasSize(2);


testPruneStateSendsTombstonesWithCorrectKeys pins this same contract more precisely — containsExactlyInAnyOrder(key1, key2) on the keys, containsOnlyNulls() on the values. A bug in tombstone keys, count, or nullness would fail both together, so I'm not sure this block catches anything the dedicated test would miss.

Would it read cleaner to leave testPruneState as it was — default store, cache eviction only, which also drops the tombstoneEnabledStore swap at the top — and keep the tombstone contract in one place? testPruneStateNoTombstonesByDefault already pins the default-off behavior.

rob-9 · 2026-07-17T16:27:43Z

thanks for the reviews @weiqingy @joeyutong . i'll try to address these soon.

rob-9 added 2 commits July 7, 2026 17:22

fix: handle null-valued records in KafkaActionStateStore.rebuildState()

0db12a6

feat: send tombstone records in KafkaActionStateStore.pruneState()

98b1aaa

github-actions Bot added doc-not-needed Your PR changes do not impact docs fixVersion/0.4.0 priority/major Default priority of the PR or issue. labels Jul 8, 2026

rob-9 added 2 commits July 15, 2026 11:16

style: fix spotless formatting violations

a02a28a

joeyutong reviewed Jul 16, 2026

View reviewed changes

weiqingy reviewed Jul 17, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[runtime] send Kafka tombstones on state pruning for durable log compaction#885

[runtime] send Kafka tombstones on state pruning for durable log compaction#885
rob-9 wants to merge 5 commits into
apache:mainfrom
rob-9:feat/691-kafka-prune-tombstones

rob-9 commented Jul 8, 2026

Uh oh!

wenjin272 commented Jul 13, 2026

Uh oh!

joeyutong commented Jul 14, 2026

Uh oh!

rob-9 commented Jul 14, 2026

Uh oh!

joeyutong commented Jul 15, 2026

Uh oh!

joeyutong Jul 16, 2026

Uh oh!

joeyutong Jul 16, 2026

Uh oh!

weiqingy Jul 17, 2026

Uh oh!

weiqingy Jul 17, 2026

Uh oh!

weiqingy Jul 17, 2026

Uh oh!

weiqingy Jul 17, 2026

Uh oh!

weiqingy Jul 17, 2026

Uh oh!

rob-9 commented Jul 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

rob-9 commented Jul 8, 2026

Purpose of change

Tests

API

Documentation

Uh oh!

wenjin272 commented Jul 13, 2026

Uh oh!

joeyutong commented Jul 14, 2026

Uh oh!

rob-9 commented Jul 14, 2026

Uh oh!

joeyutong commented Jul 15, 2026

Uh oh!

joeyutong Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

joeyutong Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

weiqingy Jul 17, 2026

Choose a reason for hiding this comment

Uh oh!

rob-9 commented Jul 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants