[GitHub] [flink] pnowojski commented on a change in pull request #18651: [FLINK-25792][connectors] Only flushing the async sink base if it is …

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

pnowojski commented on a change in pull request #18651:
URL: https://github.com/apache/flink/pull/18651#discussion_r803467988



##########
File path: flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java
##########
@@ -267,18 +267,31 @@ private void registerCallback() {
 
     @Override
     public void write(InputT element, Context context) throws IOException, InterruptedException {
+        while (mailboxExecutor.tryYield()) {}

Review comment:
       Kind of @dmvk . One small caveat is that "yield" means "yield to downstream". So it only yields to mailbox actions enqueued by self OR downstream operators.
   
   I have however a couple of suspicious about this:
   1. We shouldn't be yielding on the hot path. Only if internal operator's buffer is full 
   2. I don't understand why it's `tryYield()` in the first place? IMO It should have been `yield()` the loop below from the get go:
   ```
           while (bufferedRequestEntries.size() >= maxBufferedRequests) {
               mailboxExecutor.yield();
           }
   ```
   Otherwise we have active/busy waiting loop wasting CPU resources.
   
   
   A good example how I believe similar pattern is correctly implemented is `AsyncWaitOperator`.
   
   I am kind of lost here and I would need to understand why it was changed to `tryYield` as part of https://issues.apache.org/jira/browse/FLINK-24234. I'm afraid that I won't be able to dig into this until next week.




-- 
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: issues-unsubscribe@flink.apache.org

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