[runtime][java] Preserve pending Fluss action state across checkpoints

[rob-9]

Purpose of change

Fluss recovery can repeat a durable call whose result was saved before a checkpoint while its action remained pending. The checkpoint's bucket end offsets place that result outside the replay window.

This change keeps cached results reachable from the checkpoint. Closes #1174. Related Kafka fix: #1161; this PR does not depend on it.

Runtime flow

At snapshot time, FlussActionStateStore captures the bucket end offsets, then synchronously appends the latest cached action states. It returns the marker only after every write succeeds. Recovery uses the existing replay path; checkpoint completion still prunes completed input sequences from the cache.

Design and tradeoffs

Kafka #1161 (reference revision 14ee1fe) tracks exact producer offsets and completed-sequence boundaries to rewind its marker. Fluss 0.9.0's append acknowledgement does not expose the record offset, so this implementation refreshes states after the marker instead.

Approach	Benefit	Cost
Kafka: track offsets and rewind	No checkpoint refresh writes; tighter replay window	Offset and per-key sequence bookkeeping must survive recovery and follow eviction
Fluss: capture offsets and refresh	Existing cache and marker format; no offset guesses or extra per-key maps	One acknowledged write per cached state per snapshot, including completed states awaiting pruning

The Kafka review's interruption and bookkeeping concerns are addressed here by preserving the interrupt flag and using cache membership directly. Filtering only on action completion would be unsafe: other actions in the same input sequence may still be pending.

Behavioral Semantics

Snapshot condition	Behavior
Pending or completed action still cached	Refresh it into the replay window
State loaded during recovery	Refresh it again at the next checkpoint
Entry pruned or removed by divergence handling	No refresh write
Empty cache	Capture offsets without writes
Metadata, serialization, or append failure	Fail the snapshot; retain cached state for retry

Interrupted offset lookups and writes preserve interruption. A partially failed refresh does not invalidate the previous successful checkpoint.

Tests

Contract	Coverage
Pending results survive repeated restore and execute only once	pendingCallSurvivesRepeatedCheckpointRestoreAndCompletes: real Fluss writes, two restores, output and call-count assertions
Previous checkpoint survives a later partial refresh failure	partialRefreshFailurePreservesPreviousCheckpoint: C1 succeeds, C2 acknowledges one refresh then fails, restoring C1 completes both actions without repeating either call
Refresh preserves pre-checkpoint states across buckets and with SASL	FlussActionStateStoreIntegrationTest, FlussActionStateStoreSaslIntegrationTest
Latest state, completed actions, pruning, divergence, retry and interruption	FlussActionStateStoreTest

All 103 focused tests passed on JDK 17 with embedded Fluss. Spotless and Apache RAT passed. The original recovery regression fails against the pre-fix source at 2e97add7 and passes with the fix.

The operator tests use real keyed-state snapshots/restores and an embedded Fluss cluster. Only suspension is simulated; recovery executes the normal Java action through completion. JDK continuation scheduling, Python-specific suspension, retention expiry and a distributed Flink restart were not exercised.

Checkpoint cost

Local measurements on macOS arm64, JDK 17, embedded Fluss 0.9.0-incubating, one tablet server and four buckets. Each cached state contains one durable result. Times include offset lookup, serialization and acknowledged refresh writes; medians of three samples after one warm-up, excluding initial state writes.

Cached states	1 KiB result	16 KiB result
100	63 ms	48 ms
1,000	917 ms	1,015 ms
5,000	1,348 ms	1,604 ms

These are synthetic local measurements, not production bounds. Remote latency, replication and competing traffic can increase checkpoint cost. The benchmark harness and raw logs are excluded from this PR.

Verification command

mvn -pl runtime -am test '-Dtest=Fluss*Test,DurableExecutionManagerTest,ActionExecutionOperatorTest' -Dsurefire.failIfNoSpecifiedTests=false

API

No new public API, configuration, table schema or serialized marker format. Marker capture now writes synchronously. Existing markers remain readable, but the fix cannot repair results already excluded by older checkpoints. Fluss retention must still preserve every supported restore window.

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

[sangkyoonnam]

LGTM. The fix addresses #1174, and the one suggestion below doesn't block it.

All 39 Fluss tests pass at 08add04 on JDK 17. With only the refresh removed (getRecoveryMarker() returning getBucketEndOffsets()), both new operator tests fail, and the repeated-restore test loses the pending state on the first restore (L120). The test also checkpoints with the action still queued and asserts the output and call count after restore, which is what joeyutong asked for on #1161.

I checked the ordering in the code. Refresh rows carry the same agent_key as the original put(), so they land in the same bucket after the earlier writes and last-write-wins replay keeps the newest copy. Per-bucket minimum markers cover every subtask's refresh, and a failed refresh throws before recoveryMarkerOpState.update.

[sangkyoonnam]

The PR already documents this cost and explains why isCompleted() isn't a safe filter. A narrower filter is safe: skip states whose input sequence is at or below the key's lastCompletedSequenceNumber in this checkpoint. No replay after restoring this checkpoint needs them. Without it, refreshes include every completed input still awaiting pruning, not only in-flight ones. With timely checkpoint completion, a subtask completing 10 inputs/s, each leaving two cached action states, accumulates about 1,200 states over a 60 s interval. Each is an acknowledged append on the mailbox thread during the synchronous snapshot.

snapshotLastCompletedSequenceNumbers already collects those boundaries, immediately after snapshotRecoveryMarker(). Collecting them before the refresh and registering them in checkpointIdToSeqNums after it succeeds would let this loop use them. Keys without a boundary keep all their states. That's the needsReplay rule in #1161 and joeyutong's suggestion there, so I'd add it as a follow-up shared by both stores once #1161 settles, unless you'd rather fold it in here. A test with states at, below and above the boundary, including a completed action in an unfinished input, would pin it down.

Does a shared hook after #1161 work for you?

[rob-9]

folded this in using the same hook as #1161; refresh now skips completed input sequences, keeps states needed by unfinished inputs, and registers pruning boundaries only after the marker succeeds.

[sangkyoonnam]

Verified 1e88f8e, re-approving. Fluss getRecoveryMarker(Map) now skips states at or below each key's checkpoint boundary, and the shared snapshotState registers checkpointIdToSeqNums only after marker capture and any marker-state update succeed, which tightens the Kafka path from #1161 too. testCheckpointRefreshesOnlySequencesNeedingReplay covers below, at and above the boundary, including a completed action inside an unfinished input, and failedSnapshotDoesNotRegisterPruningBoundaries covers both the capture and the update failure. Ready for a committer pass.

[wenjin272]

@joeyutong Could you please review this PR when you have a chance?