1fanwang opened a new pull request, #1161: URL: https://github.com/apache/flink-agents/pull/1161
Linked issue: https://github.com/apache/flink-agents/issues/1158 ### Purpose of change A checkpoint can capture Kafka end offsets after a durable call result is written but before that input sequence is reflected in Flink state. Recovery then starts from the end offsets, misses the saved result, and re-runs the call. This change makes Kafka recovery markers rewind only for live action-state records whose sequence is newer than the checkpointed per-key boundary. Closes https://github.com/apache/flink-agents/issues/1158. #### Runtime flow `KafkaActionStateStore` now tracks the latest Kafka offset for each live action-state key. When an input sequence finishes, `DurableExecutionManager` marks that sequence as checkpointed for the key. `getRecoveryMarker()` starts from Kafka end offsets and rewinds only for records that are still newer than that boundary. #### Key decisions - track the latest durable offset per live action-state key - use the per-key completed-sequence boundary to separate checkpointed state from pending state - drop offset bookkeeping when recovery or pruning evicts a key ### Behavioral Semantics #### Interaction decisions | Boundary for a key | Record sequence | Marker | | --- | --- | --- | | absent | any live record | replay from that record | | `>=` sequence | checkpointed | stay at Kafka end offset | | `<` sequence | pending | rewind to that record | #### Behavioral contracts - pending durable Kafka results survive recovery - checkpointed sequences stay skipped - evicted state keys do not leave stale replay offsets behind #### Failure behavior Kafka metadata and replay failures still use the existing exception paths. Malformed recovery keys still fail recovery. ### Tests - pending-result replay: `KafkaActionStateStoreTest#testRebuildStateKeepsPendingResultBeforeRecoveryMarker` - checkpointed-vs-pending boundary: `KafkaActionStateStoreTest#testRecoveryMarkerSkipsCheckpointedSequenceAndKeepsLaterPendingSequence` - nearby replay coverage: `KafkaActionStateStoreTest#testRecoveryMarker`, `#testRebuildState`, `#testRebuildStateRemovesTombstonedKeys`, `ActionStateSerializerRestoreTest` Not verified: a live Kafka broker. <details> <summary>Raw test commands and logs</summary> ```text # Red mvn -pl runtime -am -Dtest=KafkaActionStateStoreTest#testRebuildStateKeepsPendingResultBeforeRecoveryMarker -Dsurefire.failIfNoSpecifiedTests=false test expected: ActionState{..., completed=false} but was: null # Green mvn -pl runtime -am -Dtest='KafkaActionStateStoreTest#testRecoveryMarker+testRebuildState+testRebuildStateRemovesTombstonedKeys+testRebuildStateKeepsPendingResultBeforeRecoveryMarker+testRecoveryMarkerSkipsCheckpointedSequenceAndKeepsLaterPendingSequence,ActionStateSerializerRestoreTest' -Dsurefire.failIfNoSpecifiedTests=false test Tests run: 7, Failures: 0, Errors: 0, Skipped: 0 BUILD SUCCESS ``` </details> ### API No public API changes. ### Documentation - [ ] `doc-needed` - [x] `doc-not-needed` - [ ] `doc-included` ### Was this patch authored or co-authored using generative AI tooling? - [x] Yes - [ ] No Generated-by: GitHub Copilot CLI (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
