diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index <HASH>..<HASH> 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -179,9 +179,6 @@ public class SourceStreamTask<
                             } else {
                                 // this is a "true" end of input regardless of whether
                                 // stop-with-savepoint was issued or not
-                                synchronized (lock) {
-                                    operatorChain.setIsStoppingBySyncSavepoint(false);
-                                }
                                 mailboxProcessor.allActionsCompleted();
                             }
                         });
@@ -254,6 +251,9 @@ public class SourceStreamTask<
         public void run() {
             try {
                 mainOperator.run(lock, getStreamStatusMaintainer(), operatorChain);
+                synchronized (lock) {
+                    operatorChain.setIsStoppingBySyncSavepoint(false);
+                }
                 completionFuture.complete(null);
             } catch (Throwable t) {
                 // Note, t can be also an InterruptedException
[FLINK-<I>][checkpointing] Reset OperatorChain.isStoppingBySyncSavepoint properly

Reset OperatorChain.isStoppingBySyncSavepoint from
LegacySourceFunctionThread so that StreamTask can read
it without relying on completionFuture callbacks which
can fire later.