[GitHub] [flink] rkhachatryan commented on a change in pull request #14892: [FLINK-21312][checkpointing] Reset OperatorChain.isStoppingBySyncSavуpoint properly

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

rkhachatryan commented on a change in pull request #14892:
URL: https://github.com/apache/flink/pull/14892#discussion_r572051109



##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java
##########
@@ -120,7 +120,7 @@
 
     private final OperatorEventDispatcherImpl operatorEventDispatcher;
 
-    private boolean isStoppingBySyncSavepoint;
+    private volatile boolean isStoppingBySyncSavepoint;

Review comment:
       I think that this would NOT introduce new "synchronisation" point because only visibility is an issue here (thread don't have to wait for each other).
   
   I tried in `LegacySourceFunctionThread` 
   ```
   mailboxProcessor
           .getMailboxExecutor(TaskMailbox.MAX_PRIORITY)
           .execute(
                   () -> operatorChain.setIsStoppingBySyncSavepoint(false),
                   "unsetIsStoppingBySyncSavepoint");
   ```
   It failed ~33% runs because mailbox loop can be terminated earlier (`mailboxProcessor.allActionsCompleted()` is called concurrently).
   
   Besides that, in my opinion it only moves the complexity to another place increasing it overall.
   
   Using `synchronized` to prevent visibility issue seems somewhat non-obvious and error-prone to me.  But if you prefer I'll replace `volatile` with `synchronized`.




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

For queries about this service, please contact Infrastructure at:
users@infra.apache.org