pnowojski commented on a change in pull request #8853: [FLINK-12958] Integrate
AsyncWaitOperator with mailbox
URL: https://github.com/apache/flink/pull/8853#discussion_r296743568
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java
##########
@@ -394,18 +403,37 @@ 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();
+ // TODO: once we no longer support legacy sources, the
following `if` can simply become
+ // `while(!queue.tryPut(...)) { mainThreadExecutor.yield() }'.
Currently, this case makes a difference between
+ // new mailbox and the legacy source compatibility loop that
requires us to give up the lock. 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 (!queue.tryPut(streamElementQueueEntry)) {
Review comment:
Could we encapsulate differences between `isMailboxThread` check inside a
mailbox class? Or some wrapper around mailbox?
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
