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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
