wenjin272 opened a new issue, #1301: URL: https://github.com/apache/flink-agents/issues/1301
### Search before asking - [x] I searched the existing issues. Related #1084 concerns distinct event occurrences with identical attributes; this issue tracks divergence detection and invalidation of execution history. ### Description Both KafkaActionStateStore and FlussActionStateStore treat more than one cached ActionState for the same business key and sequence number as execution divergence. A lookup then removes cached states for later sequence numbers of that key. However, seqNum advances once per top-level InputEvent, not once per Action. Multiple root Actions, an InputAction followed by an internal Event and another Action, and fan-out within one input are normal ways to produce multiple ActionStates under the same key and seqNum. Consequently, recovery can find and reuse a saved completion for an early input while unnecessarily evicting the saved completions for later inputs of the same key. Those later Actions execute again, potentially repeating expensive model calls or external side effects. Affected code in the audited 0.4 snapshot: - [KafkaActionStateStore.get/checkDivergence](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java#L249-L292) - [FlussActionStateStore.get/checkDivergence](https://github.com/apache/flink-agents/blob/47daff57b5c4189c80dd7037660f2a7832270879/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java#L247-L269) The Kafka predicate was also rechecked on the default branch when filing. ### How to reproduce 1. Enable Kafka or Fluss ActionStateStore and take checkpoint C0 before processing two inputs for the same key. 2. For each input, run InputAction -> WorkEvent -> WorkAction. Wait until all four Action completions have been persisted. 3. Fail before another checkpoint completes. 4. Restore C0 with a fresh store instance and replay the same two inputs in the same order. 5. Count actual Action body invocations across attempts. Expected: all four completed Actions are reused during recovery; their bodies are not invoked again. Actual: the first input has two cached ActionStates. Its lookup satisfies count > 1 and evicts the second input's saved states; the second input's Actions execute again. The cleanup affects the same key's later sequence numbers in the in-memory cache. It does not delete the backend log records or directly evict other keys. A recovery window containing only one input does not demonstrate the later-input eviction. ### Preserve genuine divergence protection Simply deleting the count check is insufficient. The original protection addresses histories such as: ```text attempt 1: input A at seq=s -> failure attempt 2: input B at seq=s -> read at seq=s+1 -> failure attempt 3: input A at seq=s -> read at seq=s+1 ``` If A and B update Memory differently, the read result saved under B cannot be reused under A. Also, clearing only seqNum > s leaves same-round children that may depend on the old Memory. Invalidated records remain in the backend log, so another cold recovery must not resurrect them. ### Prototype findings A local feasibility prototype used existing Event/ActionState fields and backend log order, without changing the persisted key format or user API: 1. Distinguish multiple Actions from conflicting root inputs by comparing the existing input attribute fingerprints for the same key/seqNum. Do not use regenerated random InputEvent IDs as replay identity. 2. On a confirmed root-input conflict, invalidate the current round as well as its later suffix. 3. Apply history invalidation while reconstructing the cache from the ordered backend log, so stale suffixes do not reappear after another failure/checkpoint. 4. Where a completed parent's saved outputEvents provides evidence of an internal event occurrence, compare that identity with the candidate child's taskEvent before reusing its result. Do not blindly apply random-ID comparison to regenerated subgraph bootstraps. This is feasibility evidence for the tested histories, not a general detector for arbitrary code changes or nondeterminism, and not a complete fix for #1084. The same final 28-case regression fixture produced: - Original implementation: 22 assertion failures, 0 errors/skips. - Prototype: 28/28 passing on both JDK 11 and JDK 21. - Related suites with the prototype: 246/246 on JDK 11; 189/189 on JDK 21. The tests used the real Java operator, Flink snapshot/restore, production store lookup/write and serialization paths, Kafka rebuildState, and Fluss replayRecords. Broker transport was mocked. Some replay boundaries were explicitly constructed. Live Kafka/Fluss, full Python/Pemja recovery, retention/rescaling, and the complete recovery-marker matrix remain unverified. The prototype currently scans the cache and can make log reconstruction quadratic. A production implementation needs indexes or equivalent incremental checks, plus performance validation. ### Acceptance criteria - Normal multi-root Action, cascading Action, and fan-out histories retain later saved completions on replay for both backends. - Genuine conflicting histories cannot reuse results or Memory updates from an incompatible current round or suffix. - Repeated failures and a new checkpoint after a branch change cannot resurrect stale cached history, including when the replayed log no longer contains the old root record. - Other keys remain isolated; regenerated subgraph bootstrap behavior remains valid. - Add regression coverage for the real operator recovery path and backend cache reconstruction, and targeted live-backend/Python recovery validation. - Avoid an unbounded full-cache scan for every replayed record. ### Version and environment - Flink Agents 0.4-SNAPSHOT; prototype baseline 47daff57b5c4189c80dd7037660f2a7832270879. - Kafka and Fluss ActionStateStore. - JDK 11 serial/worker execution and JDK 21 continuation execution. The JDK 21 parallel-option combinations are not independent worker-engine coverage. - The prototype has not been merged or published as a PR. ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- 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]
