wenjin272 opened a new pull request, #1306:
URL: https://github.com/apache/flink-agents/pull/1306
Linked issue: #1301
Fixes #1301
### Purpose of change
Normal fan-out and cascades now retain later saved Action completions during
recovery. Previously, both stores treated multiple ActionStates for one
key/sequence as divergence, causing later inputs to rerun unnecessarily.
#### Runtime flow
Kafka and Fluss rebuild a shared indexed cache incrementally from their
ordered logs. Each lookup checks input fingerprints and, where available,
completed producers' output Event IDs before returning a saved state. Conflicts
invalidate the affected history; ordinary misses retain the existing
conservative suffix invalidation. Kafka removes corresponding recovery offsets.
Fluss refreshes cached rounds in sequence order before checkpointing so a
subsequent rebuild preserves recovered future work.
#### Key decisions
Action counts do not identify a fork. Root matching keeps the existing
attribute fingerprint because restored InputEvents can have new random IDs.
Internal occurrence checks require persisted completed-producer evidence;
comparing every random Event ID would incorrectly reject regenerated subgraph
bootstraps. A shared per-key, ordered-sequence index avoids whole-cache scans
in lookup and log replay.
### Behavioral Semantics
#### Interaction decisions
All invalidation below is restricted to the same business key.
| Evidence and lookup result | Behavior |
| --- | --- |
| Multiple Actions, one input fingerprint, matching candidate | Reuse the
candidate; retain later rounds. |
| Conflicting input fingerprints at the same sequence | Remove the current
round and all later rounds; execute the requested Action. |
| Candidate has a different triggering Event ID, and the current internal
Event ID appears in a completed producer's outputs for that round | Remove that
candidate and later rounds; retain other states in the current round. |
| Different Event ID without completed-producer evidence | ID difference
alone does not invalidate the candidate. |
| No candidate and no input conflict | Keep other current-round states;
invalidate later rounds. |
| Earlier-round record encountered during ordered log rebuild | Discard the
previously reconstructed suffix; also discard the current round if its input
conflicts. |
#### Behavioral contracts
- Normal multi-Action histories reuse all matching completed Actions across
recovery.
- Conflicting input histories and evidenced stale child occurrences cannot
reuse the invalidated current/suffix results, including after another failure
or checkpoint.
- Other keys remain isolated, and regenerated subgraph bootstraps retain
their existing replay behavior.
- Checkpoint recovery markers and Fluss refresh ordering preserve the
surviving cached work.
#### Failure behavior
Detected conflicts and misses fall back to execution rather than failing
recovery. The existing suffix invalidation remains conservative even when a
miss does not prove divergence. Backend write/rebuild failures retain the
stores' existing exception behavior; failed writes do not add new indexed
producer evidence. No new configuration or backend schema is introduced.
### Tests
| Contract | Tests |
| --- | --- |
| Normal fan-out/cascade/repeated Action reuse |
`ActionStateDivergenceRecoveryTest` normal-history cases;
`ActionStatePythonDivergenceIT` |
| Input changes and stale same-round children cannot reuse incompatible
results | `ActionStateDivergenceRecoveryTest` changed-input, changed-root and
truncated-replay cases;
`ActionStateDivergenceIntegrationTest.changedRootCannotReuseChildComputedAgainstAnotherMemoryHistory`
|
| Another failure/checkpoint cannot revive invalidated history |
`ActionStateDivergenceRecoveryTest.checkpointAfterDivergenceMustNotReviveOldSuffix`
and repeated-restore cases |
| Other keys and regenerated bootstraps remain valid |
`ActionStateCacheTest` cross-key/boundary cases; regenerated-bootstrap recovery
case |
| Surviving work remains recoverable through marker/refresh updates |
`KafkaActionStateStoreTest.divergenceRemovesOffsetsFromRecoveryMarkers`;
`ActionStateDivergenceIntegrationTest.completedCascadeSurvivesCheckpointBeforeReplayingRecoveredInputs`
|
Java 11: 380 regression tests plus 4 native Python/Pemja tests passed, with
no skips. Java 21 clean rebuild: 394 passed, 14 skipped by existing
non-continuation assumptions in `ActionExecutionOperatorTest`; those cases
passed on Java 11. All new tests executed on both JDKs. The integration suite
used an actual Kafka container and embedded Fluss cluster. Spotless (Java 11),
Ruff, license checks and whitespace checks passed.
Not verified: the full repository suite, exhaustive retention/rescaling
combinations, a throughput or heap benchmark, and a dedicated shared-Memory
interleaving reproduction. Python/Pemja tests use simulated backend transport;
live backend tests use Java Actions. This does not solve cross-channel input
ordering (#1302), general same-attribute occurrence collisions (#1084), or
arbitrary nondeterminism.
<details>
<summary>Index invariants and reproducible test selections</summary>
Each key has an ordered sequence map with state keys, input fingerprint
counts and completed-output ID counts. Per-state evidence snapshots handle
ActionState objects mutated in place before put. Put/remove/clear, tombstones
and pruning maintain these indexes; map views cannot bypass maintenance.
Lookup costs O(log R), indexing O(log R + E), and invalidation visits
affected rounds/states. R is cached rounds for one key; E is the record's
output count. Fluss checkpoint refresh still enumerates surviving states. The
NoScanMap regression prohibits whole-map traversal over 3,000 states across 100
keys. Nine cache tests cover evidence reference counts, mutation, deletion and
sequence bounds.
Run from the checkout with the selected JDK and a supported Python on PATH.
Set PYTHONPATH to the checkout's python directory, runtime/src/test/resources
and the active Python's purelib directory. Local Mockito runs used a Byte Buddy
1.15.4 startup agent.
```sh
mvn -pl runtime -am clean test \
'-Dtest=ActionState*Test,ActionStatePythonDivergenceIT,KafkaActionState*Test,FlussActionState*Test,DurableExecutionManagerTest,ActionExecutionOperatorTest,MemoryWriteBelowValueRecoveryTest,InternalSubagentRecoveryTest,SubgraphCallTest,DecisionModelActionRecoveryTest'
\
-Dsurefire.failIfNoSpecifiedTests=false
```
This is the Java 21 expanded selection. Java 11 ran the state-store suites,
DurableExecutionManagerTest, ActionExecutionOperatorTest and
MemoryWriteBelowValueRecoveryTest, plus the Python suite separately. The native
Python test ends in IT and must be selected explicitly. Docker is required for
Kafka integration coverage.
</details>
### API
No public API, configuration, persisted key format or payload schema
changes. Existing stores gain corrected cache invalidation and reuse behavior.
Memory overhead increases for the indexes; recovery still loads the required
log window into memory.
### Documentation
- [ ] `doc-needed`
- [ ] `doc-not-needed`
- [x] `doc-included`
Consistency now documents shared Memory ordering under Known limitations,
alongside external side effects. The Actions chapter links to the combined
section.
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: Codex 0.153.4 (GPT-6)
--
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]
