pnowojski commented on a change in pull request #9426: [FLINK-12958] Integrate 
AsyncWaitOperator with mailbox (preserving compatibility with legacy sources)
URL: https://github.com/apache/flink/pull/9426#discussion_r314195257
 
 

 ##########
 File path: 
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
 ##########
 @@ -397,26 +404,41 @@ private void stopResources(boolean waitForShutdown) 
throws InterruptedException
         * @throws InterruptedException if the current thread has been 
interrupted
         */
        private <T> void addAsyncBufferEntry(StreamElementQueueEntry<T> 
streamElementQueueEntry) throws InterruptedException {
-               assert(Thread.holdsLock(checkpointingLock));
-
                pendingStreamElementQueueEntry = streamElementQueueEntry;
 
-               while (!queue.tryPut(streamElementQueueEntry)) {
-                       // we wait for the emitter to notify us if the queue 
has space left again
-                       checkpointingLock.wait();
+               if (!queue.tryPut(streamElementQueueEntry)) {
+                       do {
+                               // We will receive a notify on the lock at the 
end of each processing-letter execution (see {Emitter})
+                               // so we can try to put again.
+                               if (!tryYieldMainThreadExecutor()) {
 
 Review comment:
   This logic (difference of mailbox vs non mailbox thread) should be 
encapsulated in the mailbox. It's already duplicated here, it will be 
duplicated in other places (like `ContinousFileReaderOperator`) and mailbox 
should leak the details of the threading model to the operators.
   
   something like `mailbox.yield()` called from here, should do this black 
magic in a secret.

----------------------------------------------------------------
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:
us...@infra.apache.org


With regards,
Apache Git Services

Reply via email to