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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
