dawidwys commented on pull request #15055:
URL: https://github.com/apache/flink/pull/15055#issuecomment-872827247
I looked into the test failures and I think they're actually related to a
bug in the `MailboxProcessor`. The `MailboxProcessor#runMailboxLoop` expects
that `processMail(localMailbox, false);` won't return if there is a suspended
action. However, if there are no mails in the queue, the `processMail` will
return because a `batch` could not have been created and therefore it won't go
into the loop while default action is unavailable.
I'd suggest to fix it as follows:
```
/**
* This helper method handles all special actions from the mailbox. In
the current design, this
* method also evaluates all control flag changes. This keeps the hot
path in {@link
* #runMailboxLoop()} free from any other flag checking, at the cost
that all flag changes must
* make sure that the mailbox signals mailbox#hasMail.
*
* @return true if a mail has been processed.
*/
private boolean processMail(TaskMailbox mailbox, boolean singleStep)
throws Exception {
// Doing this check is an optimization to only have a volatile read
in the expected hot
// path, locks are only
// acquired after this point.
boolean isBatchAvailable = mailbox.createBatch();
// Take mails in a non-blockingly and execute them.
boolean processed = isBatchAvailable &&
processMailsNonBlocking(singleStep);
if (singleStep) {
return processed;
}
// If the default action is currently not available, we can run a
blocking mailbox execution
// until the default action becomes available again.
processed |= processMailsWhenDefaultActionUnavailable();
return processed;
}
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
