[PR] KAFKA-20780: Fix to clear completed inflight poll on empty fetch response and time left [kafka]

["lianetm (via GitHub)" <gi...@apache.org>]

lianetm opened a new pull request, #22979:
URL: https://github.com/apache/kafka/pull/22979

   Fix to ensure that a new fetch request (new inflightPoll event) is sent right away in the case where a previous fetch event completed without records and there is still time to poll internally. 
   
   Before this PR, a second iteration of the internal poll loop would not clear the previous event that was already done (and generated no data), so it would block on the buffer before sending a next fetch, introducing latency.
   
   Fix by ensuring that the inflight poll is cleared if completed before checking the need to send a new one.
   


-- 
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: jira-unsubscribe@kafka.apache.org

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

Re: [PR] KAFKA-20780: Fix to clear completed inflight poll on empty fetch response and time left [kafka]

["lianetm (via GitHub)" <gi...@apache.org>]

lianetm commented on code in PR #22979:
URL: https://github.com/apache/kafka/pull/22979#discussion_r3667584607


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -982,15 +982,20 @@ public ConsumerRecords<K, V> poll(final Duration timeout) {
     }
 
     /**
-     * {@code checkInflightPoll()} manages the lifetime of the {@link AsyncPollEvent} processing. If it is
-     * called when no event is currently processing, it will start a new event processing asynchronously. A check
-     * is made during each invocation to see if the <em>inflight</em> event has completed. If it has, it will be
-     * processed accordingly.
+     * {@code checkInflightPoll()} manages the lifecycle of the {@link AsyncPollEvent}. If no event is
+     * currently processing, a new one is started asynchronously. Each invocation checks whether the
+     * <em>inflight</em> event has completed; if so, a new event is submitted in its place so a fetch request
+     * stays in flight. If the completed event left records buffered no new
+     * event is submitted here (it would gate those records behind a fresh validate-positions stage).
+     * Instead the buffered records are returned and the next fetch is pipelined by {@link #poll(Duration)} via
+     * {@link #sendPrefetches(Timer)}.
      */
     private void checkInflightPoll(Timer timer, boolean firstPass) {
-        if (firstPass && inflightPoll != null) {
-            // Handle the case where there's a remaining inflight poll from the *previous* invocation
-            // of AsyncKafkaConsumer.poll().
+        // Clear the current inflight poll if we can, so a new one (and a new fetch) is submitted below.
+        // On the first pass this may clear a leftover from the previous poll(). On later passes it clears
+        // inflights that have completed. A completed poll that filled the buffer is kept, so its records
+        // are returned first (see maybeClearPreviousInflightPoll).
+        if (inflightPoll != null && (firstPass || inflightPoll.isComplete())) {

Review Comment:
   this is the fix, keeping it minimal, but as follow-up I will refactor this path to make it less error prone, ideas here https://issues.apache.org/jira/browse/KAFKA-20844, I will take it separately right after this PR 



-- 
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: jira-unsubscribe@kafka.apache.org

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

Re: [PR] KAFKA-20780: Fix to clear completed inflight poll on empty fetch response and time left [kafka]

["lianetm (via GitHub)" <gi...@apache.org>]

lianetm merged PR #22979:
URL: https://github.com/apache/kafka/pull/22979


-- 
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: jira-unsubscribe@kafka.apache.org

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