[hotfix] Introduce OperatorSnapshotFinalizer factory method

Author: rkhachatryan

flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SnapshotUtils.java

@@ -72,7 +72,7 @@ public static <OUT, OP extends StreamOperator<OUT>> TaggedOperatorSubtaskState s
                 operator.snapshotState(checkpointId, timestamp, options, storage);
 
         OperatorSubtaskState state =
-                new OperatorSnapshotFinalizer(snapshotInProgress).getJobManagerOwnedState();
+                OperatorSnapshotFinalizer.create(snapshotInProgress).getJobManagerOwnedState();
 
         operator.notifyCheckpointComplete(checkpointId);
         return new TaggedOperatorSubtaskState(index, state);

flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizer.java

@@ -27,12 +27,11 @@
 import org.apache.flink.runtime.state.SnapshotResult;
 import org.apache.flink.util.concurrent.FutureUtils;
 
-import javax.annotation.Nonnull;
-
 import java.util.concurrent.ExecutionException;
 
 import static org.apache.flink.runtime.checkpoint.StateObjectCollection.emptyIfNull;
 import static org.apache.flink.runtime.checkpoint.StateObjectCollection.singletonOrEmpty;
+import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
  * This class finalizes {@link OperatorSnapshotFutures}. Each object is created with a {@link
@@ -47,8 +46,9 @@ public class OperatorSnapshotFinalizer {
     /** Secondary replica of the operator subtask state for faster, local recovery on TM. */
     private final OperatorSubtaskState taskLocalState;
 
-    public OperatorSnapshotFinalizer(@Nonnull OperatorSnapshotFutures snapshotFutures)
+    public static OperatorSnapshotFinalizer create(OperatorSnapshotFutures snapshotFutures)
             throws ExecutionException, InterruptedException {
+        checkNotNull(snapshotFutures);
 
         SnapshotResult<KeyedStateHandle> keyedManaged =
                 FutureUtils.runIfNotDoneAndGet(snapshotFutures.getKeyedStateManagedFuture());
@@ -68,7 +68,7 @@ public OperatorSnapshotFinalizer(@Nonnull OperatorSnapshotFutures snapshotFuture
         SnapshotResult<StateObjectCollection<OutputStateHandle>> resultSubpartition =
                 snapshotFutures.getResultSubpartitionStateFuture().get();
 
-        jobManagerOwnedState =
+        OperatorSubtaskState jobManagerOwnedState =
                 OperatorSubtaskState.builder()
                         .setManagedOperatorState(
                                 singletonOrEmpty(operatorManaged.getJobManagerOwnedSnapshot()))
@@ -83,7 +83,7 @@ public OperatorSnapshotFinalizer(@Nonnull OperatorSnapshotFutures snapshotFuture
                                 emptyIfNull(resultSubpartition.getJobManagerOwnedSnapshot()))
                         .build();
 
-        taskLocalState =
+        OperatorSubtaskState taskLocalState =
                 OperatorSubtaskState.builder()
                         .setManagedOperatorState(
                                 singletonOrEmpty(operatorManaged.getTaskLocalSnapshot()))
@@ -94,6 +94,14 @@ public OperatorSnapshotFinalizer(@Nonnull OperatorSnapshotFutures snapshotFuture
                         .setResultSubpartitionState(
                                 emptyIfNull(resultSubpartition.getTaskLocalSnapshot()))
                         .build();
+
+        return new OperatorSnapshotFinalizer(jobManagerOwnedState, taskLocalState);
+    }
+
+    public OperatorSnapshotFinalizer(
+            OperatorSubtaskState jobManagerOwnedState, OperatorSubtaskState taskLocalState) {
+        this.jobManagerOwnedState = checkNotNull(jobManagerOwnedState);
+        this.taskLocalState = checkNotNull(taskLocalState);
     }
 
     public OperatorSubtaskState getTaskLocalState() {

flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java

@@ -189,7 +189,7 @@ private SnapshotsFinalizeResult finalizeNonFinishedSnapshots() throws Exception
 
             // finalize the async part of all by executing all snapshot runnables
             OperatorSnapshotFinalizer finalizedSnapshots =
-                    new OperatorSnapshotFinalizer(snapshotInProgress);
+                    OperatorSnapshotFinalizer.create(snapshotInProgress);
 
             jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(
                     operatorID, finalizedSnapshots.getJobManagerOwnedState());

flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/OperatorSnapshotFinalizerTest.java

@@ -93,7 +93,7 @@ void testRunAndExtract() throws Exception {
             assertThat(f).isNotDone();
         }
 
-        OperatorSnapshotFinalizer finalizer = new OperatorSnapshotFinalizer(snapshotFutures);
+        OperatorSnapshotFinalizer finalizer = OperatorSnapshotFinalizer.create(snapshotFutures);
 
         for (Future<?> f : snapshotFutures.getAllFutures()) {
             assertThat(f).isDone();

flink-runtime/src/test/java/org/apache/flink/streaming/util/AbstractStreamOperatorTestHarness.java

@@ -761,7 +761,7 @@ public OperatorSnapshotFinalizer snapshotWithLocalState(
                         checkpointStorageAccess.resolveCheckpointStorageLocation(
                                 checkpointId, locationReference));
 
-        return new OperatorSnapshotFinalizer(operatorStateResult);
+        return OperatorSnapshotFinalizer.create(operatorStateResult);
     }
 
     /**