[GitHub] [flink] StephanEwen commented on pull request #12510: [FLINK-18039]

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

StephanEwen commented on pull request #12510:
URL: https://github.com/apache/flink/pull/12510#issuecomment-643396266


   Thanks for the patch! I like the general direction, but am unsure about a few details.
   
   The earlier version of the code had the re-creation separately in the `RecreateOnResetOperatorCoordinator` class. In hind-sight, I think that was better.
     - The `OperatorCoordinatorHolder` has already a somewhat complex task (threading model, checkpoint guarantees, in the future probably execution versioning). I think it would be nice to keep future responsibilities out of this class. The "recreate on reset" sounds like it can fit into a separate class well (like in the earlier version).
     - This also allows us to make this optional. Coordinator implementations that want that behavior can use it by wrapping themselves into the `RecreateOnResetOperatorCoordinator`, others simply go as they are.
   
   Does the "quiesced context" have to throw exceptions? In other quiescing implementations the components turn their methods into no-ops instead, which means no unexpected exceptions during closing/recreation.
   
   The quiesced context uses a way of counting in-progress message sends.
     - Is the purpose here to avoid locks because of dangers of deadlocks?
     - I think the same mechanism we use in "sendEvent()" is also needed in "failJob()". Both are interactions with outside components.
     - Could we avoid having the `Thread.sleep(1)` statement?
   
   The code below is a suggestion for a quiesced context that does not throw an exception and uses wait/notify instead of sleeping.
   
   ```java
   class QuiesceableContext implements OperatorCoordinator.Context {
   
       private final OperatorCoordinator.Context context;
       private Object lock;
       private int callsInProgress;
       private boolean quiesced;
   
       @Override
       public CompletableFuture<Acknowledge> sendEvent(
               OperatorEvent evt,
               int targetSubtask) throws TaskNotRunningException {
   
           if (enterAndCheckQuiesced()) {
               return CompletableFuture.completedFuture(null);
           }
   
           try {
               return context.sendEvent(evt, targetSubtask);
           } finally {
               exit();
           }
       }
   
       private boolean enterAndCheckQuiesced() {
           synchronized (lock) {
               if (quiesced) {
                   return true;
               }
               callsInProgress++;
               return false;
       }
   
       private booleanexit() {
           synchronized (lock) {
               callsInProgress--;
               if (callsInProgress == 0) {
                   lock.notifyAll();
               }
           }
       }
   
       private void quiesce() {
           synchronized (lock) {
               quiesced = true;
               while (callsInProgress > 0) {
                   lock.wait();
               }
           }
       }
   }
   ```


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