You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "rob-9 (via GitHub)" <gi...@apache.org> on 2026/09/28 22:34:10 UTC
[PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
rob-9 opened a new pull request, #1175:
URL: https://github.com/apache/flink-agents/pull/1175
Linked issue: #1174
### Purpose of change
Recovery should reuse a durable call result saved before a checkpoint when its action is still pending. Fluss previously saved bucket end offsets as the recovery marker, which could place that result before the replay window and cause the external call to run again.
Closes #1174. Related Kafka fix: #1161; this PR does not depend on it.
#### Runtime flow
When the operator snapshots its recovery marker, `FlussActionStateStore` first reads the bucket end offsets, then synchronously appends the latest cached action states. It returns the captured offsets only after every append succeeds. Recovery reads those refreshed states and any subsequent writes through the existing replay path. Checkpoint completion still prunes completed input sequences from the cache.
#### Design and comparison with Kafka
The Kafka implementation in #1161, reviewed at `14ee1fe`, tracks exact offsets returned by producer writes and uses per-key completed-sequence boundaries to decide which records need replay. Fluss 0.9.0's `AppendResult` does not expose the written record's offset, so this PR refreshes states after the marker instead.
| Approach | Benefit | Cost |
| --- | --- | --- |
| Kafka #1161: track offsets and rewind | No checkpoint refresh writes; completed sequences need not hold the replay window back | Additional offset and sequence bookkeeping, including recovery reconstruction and eviction |
| This PR: capture offsets and refresh cached Fluss states | Uses the existing cache and recovery-marker format; no offset guesses or new per-key maps | One synchronous append per cached state per snapshot, including completed states awaiting pruning; more checkpoint latency and log traffic |
The review decision for this draft is whether that checkpoint write cost is acceptable. Performance under large caches and frequent checkpoints has not been measured. Filtering solely on `ActionState.isCompleted()` would be unsafe: a completed action can still belong to an input sequence whose other actions are pending.
### Behavioral Semantics
| State at snapshot | Behavior |
| --- | --- |
| Pending action with saved call results | Refresh into the recovery window |
| Completed action still in the cache | Refresh until its input sequence is pruned |
| State reconstructed during recovery | Refresh again at the next snapshot |
| Pruned or divergence-evicted entry | No refresh write |
| Empty cache | Capture offsets without appending records |
A metadata lookup, serialization, or append failure prevents the marker from being returned and fails the snapshot. Partial refresh writes do not evict cached state; a retry refreshes all remaining entries again. Interrupted offset lookups and appends preserve the interrupt flag. Fluss retention must still preserve the required replay window.
### Tests
| Contract | Coverage |
| --- | --- |
| Pending results survive repeated checkpoints and replay completes without repeating the call | `FlussActionStateRecoveryIntegrationTest#pendingCallSurvivesRepeatedCheckpointRestoreAndCompletes` |
| Pre-checkpoint states remain reachable across buckets | `FlussActionStateStoreIntegrationTest#testRebuildStateWithRecoveryMarkers`, `#testMultiBucketRecovery` |
| Refresh uses latest cached states, including completed actions | `FlussActionStateStoreTest#testCheckpointRefreshesLatestStatesAfterCapturingOffsets` |
| Pruning and divergence stop refresh writes | `#testCheckpointDoesNotRefreshPrunedOrDivergedStates` |
| Partial refresh failure retains state for retry | `#testFailedCheckpointRefreshRetainsCacheForRetry` |
| Cancellation preserves interruption | `#testInterruptedRefreshRestoresInterruptFlag`, `#testInterruptedOffsetLookupDoesNotRefresh` |
All 99 focused tests passed on JDK 17 with an embedded Fluss 0.9.0-incubating cluster. A review rerun of the 23 store and operator recovery tests also passed. The operator regression fails against the original implementation at `2e97add7` because the saved state is missing after restore, and passes with the fix. The Maven reactor rebuilt the affected modules from source; Spotless and Apache RAT checks passed.
The operator test uses an embedded Fluss cluster and real keyed-state snapshots/restores. A test task deterministically yields after executing a real durable call, so it runs on every supported JDK; subsequent recovery executes the normal Java action through completion. It does not exercise JDK continuation scheduling, a full distributed Flink restart, or Python-specific suspension. Retention expiry and checkpoint performance have not been tested.
Before marking this draft ready, benchmark checkpoint cost with a representative cache size and add a regression that restores the previous successful checkpoint after a later checkpoint's refresh partially succeeds and then fails. The current failure test verifies cache preservation and retry, but not that recovery scenario.
<details>
<summary>Kafka review considerations and verification command</summary>
The Kafka review identified interrupted writes clearing the interrupt flag and completed-sequence bookkeeping growing after state eviction. Both were addressed in its latest reviewed revision. This implementation preserves interruption and adds no extra per-key bookkeeping.
The Kafka operator test restores the saved call result but does not run the pending action through completion. This PR's regression test additionally restores twice, executes the pending action, verifies its output and a call count of one, and checks recovery after completion and pruning.
- Interrupt handling: https://github.com/apache/flink-agents/pull/1161#discussion_r4108563928
- Bookkeeping cleanup: https://github.com/apache/flink-agents/pull/1161#discussion_r4108564256
- Replay test coverage: https://github.com/apache/flink-agents/pull/1161#discussion_r4108564565
```sh
mvn -pl runtime -am test -Dtest=FlussActionStateStoreTest,FlussActionStateStoreIntegrationTest,FlussActionStateRecoveryIntegrationTest,DurableExecutionManagerTest,ActionExecutionOperatorTest -Dsurefire.failIfNoSpecifiedTests=false
```
</details>
### API
No new configuration, public API, table schema, or serialized marker format. Fluss marker capture now performs synchronous writes. Existing markers remain readable, but this change cannot repair results already excluded by a checkpoint taken before the fix.
### Documentation
- [ ] `doc-needed`
- [ ] `doc-not-needed`
- [x] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 merged PR #1175:
URL: https://github.com/apache/flink-agents/pull/1175
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #1175:
URL: https://github.com/apache/flink-agents/pull/1175#issuecomment-6142026588
@joeyutong Could you please review this PR when you have a chance?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
Posted by "rob-9 (via GitHub)" <gi...@apache.org>.
rob-9 commented on code in PR #1175:
URL: https://github.com/apache/flink-agents/pull/1175#discussion_r4245394346
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java:
##########
@@ -483,19 +493,41 @@ private Map<Integer, Long> getBucketOffsets(OffsetSpec offsetSpec) {
buckets.add(b);
}
return admin.listOffsets(tablePath, buckets, offsetSpec).all().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Interrupted getting offsets for Fluss table: " + tablePath, e);
} catch (Exception e) {
throw new RuntimeException("Failed to get offsets for Fluss table: " + tablePath, e);
}
}
/**
- * Returns the end offsets of each bucket as a recovery marker. Similar to Kafka's
- * implementation, this captures the current log position so that {@link #rebuildState} can
- * resume from these offsets instead of scanning from the beginning.
+ * Captures bucket end offsets, then synchronously appends the cached states after those offsets
+ * before returning the recovery marker.
+ *
+ * <p>A pending action may already have persisted call results before the checkpoint. Starting
+ * recovery at the unmodified log end would lose those results. Fluss append acknowledgements do
+ * not expose record offsets, so refreshing the cache into the recovery window avoids guessing
+ * where the earlier writes landed. Completed actions are also retained until the enclosing
+ * input sequence is pruned after a completed checkpoint.
+ *
+ * <p>Called on the operator thread, serially with puts and pruning. A failed refresh fails the
+ * checkpoint; partial appends remain safe for earlier recovery markers. This adds one append
+ * per cached state per snapshot and requires Fluss retention to preserve the recovery window.
*/
@Override
public Object getRecoveryMarker() {
- return getBucketEndOffsets();
+ Map<Integer, Long> offsets = getBucketEndOffsets();
+ try {
+ for (Map.Entry<String, ActionState> entry : actionStates.entrySet()) {
Review Comment:
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.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
Posted by "sangkyoonnam (via GitHub)" <gi...@apache.org>.
sangkyoonnam commented on code in PR #1175:
URL: https://github.com/apache/flink-agents/pull/1175#discussion_r4235327035
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java:
##########
@@ -483,19 +493,41 @@ private Map<Integer, Long> getBucketOffsets(OffsetSpec offsetSpec) {
buckets.add(b);
}
return admin.listOffsets(tablePath, buckets, offsetSpec).all().get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "Interrupted getting offsets for Fluss table: " + tablePath, e);
} catch (Exception e) {
throw new RuntimeException("Failed to get offsets for Fluss table: " + tablePath, e);
}
}
/**
- * Returns the end offsets of each bucket as a recovery marker. Similar to Kafka's
- * implementation, this captures the current log position so that {@link #rebuildState} can
- * resume from these offsets instead of scanning from the beginning.
+ * Captures bucket end offsets, then synchronously appends the cached states after those offsets
+ * before returning the recovery marker.
+ *
+ * <p>A pending action may already have persisted call results before the checkpoint. Starting
+ * recovery at the unmodified log end would lose those results. Fluss append acknowledgements do
+ * not expose record offsets, so refreshing the cache into the recovery window avoids guessing
+ * where the earlier writes landed. Completed actions are also retained until the enclosing
+ * input sequence is pruned after a completed checkpoint.
+ *
+ * <p>Called on the operator thread, serially with puts and pruning. A failed refresh fails the
+ * checkpoint; partial appends remain safe for earlier recovery markers. This adds one append
+ * per cached state per snapshot and requires Fluss retention to preserve the recovery window.
*/
@Override
public Object getRecoveryMarker() {
- return getBucketEndOffsets();
+ Map<Integer, Long> offsets = getBucketEndOffsets();
+ try {
+ for (Map.Entry<String, ActionState> entry : actionStates.entrySet()) {
Review Comment:
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](https://github.com/apache/flink-agents/pull/1161#discussion_r4226684419), 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?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]
Posted by "joeyutong (via GitHub)" <gi...@apache.org>.
joeyutong commented on PR #1175:
URL: https://github.com/apache/flink-agents/pull/1175#issuecomment-6210433801
Looks good to me overall.
cc @beryllw for awareness: exposing the bucket and an append offset (even the batch base offset) through Fluss `AppendResult` could potentially let this backend use a Kafka-style recovery marker and avoid the extra state refresh writes at checkpoint time. The log-produce response already carries [`base_offset`](https://github.com/apache/fluss/blob/v0.9.0-incubating/fluss-rpc/src/main/proto/FlussApi.proto#L793-L799), but it isn't exposed through `AppendResult` today.
Just a possible future enhancement; the current approach looks reasonable for this 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org