Re: [PR] [runtime][java] Preserve pending Fluss action state across checkpoints [flink-agents]

["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