Re: [PR] [FLINK-38543] Change the overall UC restore process, JM and task initialization [flink]

["pnowojski (via GitHub)" <gi...@apache.org>]

pnowojski commented on code in PR #27862:
URL: https://github.com/apache/flink/pull/27862#discussion_r3032853913


##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -913,8 +913,14 @@ private CompletableFuture<Void> restoreStateAndGates(
                                             "Input gate request partitions"));
         }
 
-        return CompletableFuture.allOf(recoveredFutures.toArray(new CompletableFuture[0]))
-                .thenRun(mailboxProcessor::suspend);
+        // Return allOf result instead of thenRun result.
+        // thenRun returns a NEW future that completes after the callback finishes.
+        // Since suspend() runs on the async thread and just sends a poison mail,
+        // the mailbox loop can exit before suspend() returns, causing isDone() to be false.
+        CompletableFuture<Void> allRecoveredFuture =
+                CompletableFuture.allOf(recoveredFutures.toArray(new CompletableFuture[0]));
+        allRecoveredFuture.thenRun(mailboxProcessor::suspend);
+        return allRecoveredFuture;

Review Comment:
   Please add a test coverage for this issue.



##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java:
##########
@@ -109,8 +116,16 @@ public void setChannelStateWriter(ChannelStateWriter channelStateWriter) {
     }
 
     public final InputChannel toInputChannel() throws IOException {
-        Preconditions.checkState(
-                stateConsumedFuture.isDone(), "recovered state is not fully consumed");
+        // Check the appropriate future based on configuration:
+        // - When unaligned during recovery is enabled: check bufferFilteringCompleteFuture
+        // - When disabled: check stateConsumedFuture (original behavior)
+        if (inputGate.isCheckpointingDuringRecoveryEnabled()) {
+            Preconditions.checkState(
+                    bufferFilteringCompleteFuture.isDone(), "buffer filtering is not complete");

Review Comment:
   I think it would be better invariant to always `checkState` this, regarldess if `isCheckpointingDuringRecoveryEnabled` is enabled or not and that `bufferFilteringCompleteFuture` should be always completed.
   
   Otherwise we might have an incosistent state where `bufferFilteringCompleteFuture.isDone() == false` while `stateConsumedFuture.isDone() == true`, which doesn't make sense.



##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -913,8 +913,14 @@ private CompletableFuture<Void> restoreStateAndGates(
                                             "Input gate request partitions"));
         }
 
-        return CompletableFuture.allOf(recoveredFutures.toArray(new CompletableFuture[0]))
-                .thenRun(mailboxProcessor::suspend);
+        // Return allOf result instead of thenRun result.
+        // thenRun returns a NEW future that completes after the callback finishes.
+        // Since suspend() runs on the async thread and just sends a poison mail,
+        // the mailbox loop can exit before suspend() returns, causing isDone() to be false.
+        CompletableFuture<Void> allRecoveredFuture =
+                CompletableFuture.allOf(recoveredFutures.toArray(new CompletableFuture[0]));
+        allRecoveredFuture.thenRun(mailboxProcessor::suspend);
+        return allRecoveredFuture;

Review Comment:
   Also, isn't this worthy of a separate bug fix? Can not this cause some critical problems?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/InputGate.java:
##########
@@ -192,5 +192,12 @@ public String toString() {
 
     public abstract CompletableFuture<Void> getStateConsumedFuture();
 
+    /**
+     * Returns a future that completes when buffer filtering is complete for all channels. This
+     * future completes before {@link #getStateConsumedFuture()}, enabling earlier RUNNING state
+     * transition when unaligned checkpoint during recovery is enabled.
+     */
+    public abstract CompletableFuture<Void> getBufferFilteringCompleteFuture();

Review Comment:
   This commit lacks a test coverage. You should add some a single simple test to for example for `UnionInputGate` - that would indirectly test for `RecoveredInputChannel` and `SingleInputGate`



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