[GitHub] [flink] lindong28 commented on a diff in pull request #20454: [FLINK-28639][Runtime/Checkpointing] Preserve consistency of events from subtask to OC

[GitBox <gi...@apache.org>]

lindong28 commented on code in PR #20454:
URL: https://github.com/apache/flink/pull/20454#discussion_r1001273247


##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java:
##########
@@ -207,25 +241,43 @@ public void close() throws Exception {
     public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event)
             throws Exception {
         mainThreadExecutor.assertRunningInMainThread();
-        if (event instanceof AcknowledgeCheckpointEvent) {
+
+        if (event instanceof AcknowledgeCloseGatewayEvent) {
+            Preconditions.checkArgument(
+                    subtask == ((AcknowledgeCloseGatewayEvent) event).getSubtaskIndex());
+            completeAcknowledgeCloseGatewayFuture(
+                    ((AcknowledgeCloseGatewayEvent) event).getCheckpointID(), subtask);
+            return;
+        } else if (event instanceof OpenGatewayEvent) {
+            Preconditions.checkArgument(subtask == ((OpenGatewayEvent) event).getSubtaskIndex());
             subtaskGatewayMap
                     .get(subtask)
-                    .openGatewayAndUnmarkCheckpoint(
-                            ((AcknowledgeCheckpointEvent) event).getCheckpointID());
+                    .openGatewayAndUnmarkCheckpoint(((OpenGatewayEvent) event).getCheckpointID());
             return;
         }
+
         coordinator.handleEventFromOperator(subtask, attemptNumber, event);
     }
 
     public void executionAttemptFailed(int subtask, int attemptNumber, @Nullable Throwable reason) {
         mainThreadExecutor.assertRunningInMainThread();
+
+        if (!context.isConcurrentExecutionAttemptsSupported()) {
+            abortAcknowledgeCloseGatewayFutures(
+                    x -> x.f1 == subtask, String.format("Subtask %d has failed.", subtask), reason);
+        }
+
         coordinator.executionAttemptFailed(subtask, attemptNumber, reason);
     }
 
     @Override
     public void subtaskReset(int subtask, long checkpointId) {
         mainThreadExecutor.assertRunningInMainThread();
 
+        if (!context.isConcurrentExecutionAttemptsSupported()) {
+            checkNoSuchAcknowledgeCloseGatewayFutures(x -> x.f1 == subtask);

Review Comment:
   Is it possible that `subtaskReset(...)` is invoked after the coordinator sends the `closeGatewayEvent` but before the subtask sends the `openGatewayEvent`?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OpenGatewayEvent.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.operators.coordination;
+
+import java.util.Objects;
+
+/**
+ * An {@link OperatorEvent} sent between coordinators and subtasks to notify the other size that the

Review Comment:
   size -> side
   
   Can you update the PR description to make it clear when/who will send this event?
   
   Can you also update the JIRA description to provide information around when/who/which events are sent in this design?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorHolder.java:
##########
@@ -397,12 +580,19 @@ public void abortCurrentTriggering() {
         // we can remove the delegation once the checkpoint coordinator runs fully in the
         // scheduler's main thread executor
         mainThreadExecutor.execute(
-                () ->
-                        subtaskGatewayMap
-                                .values()
-                                .forEach(
-                                        SubtaskGatewayImpl
-                                                ::openGatewayAndUnmarkLastCheckpointIfAny));
+                () -> {
+                    abortAcknowledgeCloseGatewayFutures(

Review Comment:
   Just to double check, this method may be invoked without `notifyCheckpointAborted()` being invoked for the `latestAttemptedCheckpointId`, right?



##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorEventDispatcherImpl.java:
##########
@@ -75,6 +103,36 @@ void dispatchEventToHandlers(
         }
     }
 
+    void initializeOperatorEventGatewayIfExists(StreamOperator<?> operator) throws Exception {
+        if (gatewayMap.containsKey(operator.getOperatorID())) {
+            getOperatorEventGateway(operator.getOperatorID())
+                    .initializeState(getOperatorStateBackend(operator));
+        }
+    }
+
+    void snapshotOperatorEventGatewayIfExists(StreamOperator<?> operator) throws Exception {
+        if (gatewayMap.containsKey(operator.getOperatorID())) {
+            getOperatorEventGateway(operator.getOperatorID())
+                    .snapshotState(getOperatorStateBackend(operator));
+        }
+    }
+
+    void notifyCheckpointAbortedIfExists(StreamOperator<?> operator, long checkpointId) {
+        if (gatewayMap.containsKey(operator.getOperatorID())) {
+            OperatorEventGatewayImpl gateway = getOperatorEventGateway(operator.getOperatorID());
+            gateway.openGateway(checkpointId);
+        }
+    }
+
+    void notifyOperatorSnapshotCreatedIfExists(StreamOperator<?> operator, long checkpointId) {

Review Comment:
   Since this method is called right after the snapshot future is created, should it be named `notifyOperatorSnapshotStartedIfExists()`?



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