[runtime] send Kafka tombstones on state pruning for durable log compaction#885
[runtime] send Kafka tombstones on state pruning for durable log compaction#885rob-9 wants to merge 5 commits into
Conversation
github-actions
Bot
added
doc-not-needed
|
@joeyutong Can you take a look at this PR? |
|
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. Could tombstone emission be opt-in, for example through |
Yeah, agreed. Can we make cleanup safe by default though? Maybe we could stamp tombstones with the checkpoint id that triggered the prune; 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
|
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:
Since 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.
| * 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 = |
There was a problem hiding this comment.
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.
| actionStateStore = tombstoneEnabledStore(actionStates, mockProducer); | ||
| String stateKey = ActionStateUtil.generateKey(TEST_KEY, 1L, testAction, testEvent); | ||
| actionStates.put(stateKey, testActionState); | ||
| mockProducer.errorNext(new RuntimeException("simulated broker failure")); |
There was a problem hiding this comment.
mockProducer uses autoComplete=true, so errorNext() is called before any pending completion exists and does not make the subsequent send() fail. This test therefore exercises the successful send path. Could we inject an exception through the callback so the async failure path is actually covered?
3 participants
Addresses #691
Purpose of change
KafkaActionStateStore.pruneState()only removed entries from the in-memory cache. the underlying Kafka topic was never updated, so:rebuildState()replayed pruned records back into memory, resurrecting state that should have been discardedThis PR sends
null-valued records to the state topic during pruning, and updatesrebuildState()to skip tombstoned keys during replay instead of inserting them into the cache. Also fixes an NPE in the error-logging path that calledrecord.value().toString()on a null value.Tests
mvn --batch-mode --no-transfer-progress -pl runtime -am -Dtest=KafkaActionStateStoreTest testtestPruneStateverifies tombstones are produced and in-memory entries are evictedtestPruneStateSendsTombstonesWithCorrectKeysasserts exact composite keys on produced tombstonestestPruneStateNoMatchingKeysverifies no tombstones sent when no keys matchtestPruneStateWithNullProducerverifies graceful degradation when producer is nullAPI
No public API changes.
Documentation
doc-not-needed