This is an automated email from the ASF dual-hosted git repository.
lianetm pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 20e952c7838 KAFKA-20780: Fix to clear completed inflight poll on empty
fetch response and time left (#22979)
20e952c7838 is described below
commit 20e952c78383e18e9102dd436dc5a8071a08bb05
Author: Lianet Magrans <[email protected]>
AuthorDate: Tue Jul 28 16:20:39 2026 -0400
KAFKA-20780: Fix to clear completed inflight poll on empty fetch response
and time left (#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.
Reviewers: Andrew Schofield <[email protected]>
---
.../consumer/internals/AsyncKafkaConsumer.java | 19 +--
.../consumer/internals/AsyncKafkaConsumerTest.java | 135 ++++++++++++++++++++-
.../internals/FetchRequestManagerTest.java | 39 ++++++
3 files changed, 180 insertions(+), 13 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 38dd6cb23c7..13a94c4bba8 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -982,15 +982,20 @@ public class AsyncKafkaConsumer<K, V> implements
ConsumerDelegate<K, V> {
}
/**
- * {@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())) {
maybeClearPreviousInflightPoll();
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 5ddca753702..59d9f345f56 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -44,6 +44,7 @@ import
org.apache.kafka.clients.consumer.internals.events.CompletableApplication
import
org.apache.kafka.clients.consumer.internals.events.CompletableBackgroundEvent;
import
org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper;
import
org.apache.kafka.clients.consumer.internals.events.ConsumerRebalanceListenerCallbackCompletedEvent;
+import
org.apache.kafka.clients.consumer.internals.events.CreateFetchRequestsEvent;
import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.clients.consumer.internals.events.EventProcessor;
import
org.apache.kafka.clients.consumer.internals.events.FetchCommittedOffsetsEvent;
@@ -108,6 +109,7 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -506,6 +508,129 @@ public class AsyncKafkaConsumerTest {
assertTrue(elapsed < 500, "Wakeup should interrupt promptly, took " +
elapsed + "ms");
}
+ /**
+ * When a single {@link AsyncKafkaConsumer#poll(Duration)} call runs
multiple internal iterations (because
+ * fetches keep coming back empty), the consumer must keep a fetch request
pending on the broker: if the
+ * poll event from one iteration completes without returning any records,
the next iteration must submit a
+ * fresh {@link AsyncPollEvent} (which drives a new fetch) rather than
idling. This test drives two such
+ * iterations over an empty fetch buffer and verifies a new poll event is
submitted on each one. See
+ * KAFKA-20780.
+ */
+ @Test
+ public void testInflightPollResubmittedAfterCompletionWithEmptyBuffer() {
+ FetchBuffer fetchBuffer = mock(FetchBuffer.class);
+ SubscriptionState subscriptions = new SubscriptionState(new
LogContext(), AutoOffsetResetStrategy.EARLIEST);
+ consumer = newConsumer(fetchBuffer, mock(ConsumerInterceptors.class),
+ mock(ConsumerRebalanceListenerInvoker.class), subscriptions);
+
+ final TopicPartition tp = new TopicPartition("topic1", 0);
+ subscriptions.assignFromUser(singleton(tp));
+ subscriptions.seek(tp, 0);
+
+ // Capture poll events without completing them here. They are
completed later from awaitWakeup() instead.
+ final List<AsyncPollEvent> submittedEvents = new ArrayList<>();
+ doAnswer(invocation -> {
+ submittedEvents.add(invocation.getArgument(0));
+ return null;
+ }).when(applicationEventHandler).add(isA(AsyncPollEvent.class));
+
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+ // The buffer is empty on every pass (the fetch responses are empty),
so a completed poll must be
+ // replaced to keep a fetch pending.
+ doReturn(true).when(fetchBuffer).isEmpty();
+
+ // Model the application thread being woken by an empty fetch
response: the inflight poll completes, and
+ // the clock advances so the poll loop runs exactly two passes (150ms
per pass, 200ms timeout). The poll
+ // timer is refreshed here because a real FetchBuffer would.
+ doAnswer(invocation -> {
+ submittedEvents.forEach(event -> {
+ if (!event.isComplete()) {
+ event.markValidatePositionsComplete();
+ event.completeSuccessfully();
+ }
+ });
+ time.sleep(150);
+ Timer pollTimer = invocation.getArgument(0);
+ pollTimer.update();
+ return null;
+ }).when(fetchBuffer).awaitWakeup(any());
+
+ consumer.poll(Duration.ofMillis(200));
+
+ // A fresh poll event on each of the two passes; the bug submits only
one (the second pass is starved).
+ verify(applicationEventHandler,
times(2)).add(isA(AsyncPollEvent.class));
+ }
+
+ /**
+ * When the inflight poll (its fetch) completes with an error, {@link
AsyncKafkaConsumer#poll(Duration)}
+ * surfaces the error and clears the event, so a subsequent poll submits a
fresh event and resumes keeping a
+ * fetch request pending on the broker.
+ */
+ @Test
+ public void testPollSurfacesInflightPollErrorAndResumes() {
+ consumer = newConsumer();
+ final TopicPartition tp = new TopicPartition("topic", 0);
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+ completeAssignmentChangeEventSuccessfully();
+ consumer.assign(singleton(tp));
+
+ // The inflight poll completes with an error (e.g. a failed fetch);
poll() must surface it and clear it.
+ final KafkaException fetchError = new KafkaException("fetch failed");
+ doAnswer(invocation -> {
+ AsyncPollEvent event = invocation.getArgument(0);
+ event.completeExceptionally(fetchError);
+ return null;
+ }).when(applicationEventHandler).add(isA(AsyncPollEvent.class));
+ final KafkaException thrown = assertThrows(KafkaException.class, () ->
consumer.poll(Duration.ZERO));
+ assertEquals("fetch failed", thrown.getMessage());
+
+ // The errored event was cleared: the next (successful) poll submits a
fresh event rather than re-throwing.
+ completeAsyncPollEventSuccessfully();
+ assertDoesNotThrow(() -> consumer.poll(Duration.ZERO));
+ verify(applicationEventHandler,
times(2)).add(isA(AsyncPollEvent.class));
+ }
+
+ /**
+ * When an inflight poll completes with records already in the fetch
buffer, the next poll must return those
+ * records <em>without</em> submitting a new poll event: a fresh event
would re-run the validate-positions
+ * stage and starve the buffered records. Guards the buffer-guarded clear
in
+ * {@link AsyncKafkaConsumer#checkInflightPoll(Timer, boolean)}
(KAFKA-20780).
+ */
+ @Test
+ public void testBufferedRecordsReturnedWithoutResubmittingPollEvent() {
+ FetchBuffer fetchBuffer = mock(FetchBuffer.class);
+ SubscriptionState subscriptions = new SubscriptionState(new
LogContext(), AutoOffsetResetStrategy.EARLIEST);
+ consumer = newConsumer(fetchBuffer, new
ConsumerInterceptors<>(Collections.emptyList(), metrics),
+ mock(ConsumerRebalanceListenerInvoker.class), subscriptions);
+ final TopicPartition tp = new TopicPartition("topic1", 0);
+ subscriptions.assignFromUser(singleton(tp));
+ subscriptions.seek(tp, 0);
+
+ // First poll submits a poll event; leave it in flight (incomplete) so
it carries over to the next poll().
+ final List<AsyncPollEvent> submittedEvents = new ArrayList<>();
+ doAnswer(invocation -> {
+ submittedEvents.add(invocation.getArgument(0));
+ return null;
+ }).when(applicationEventHandler).add(isA(AsyncPollEvent.class));
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+ consumer.poll(Duration.ZERO);
+ assertEquals(1, submittedEvents.size());
+
+ // That poll event now completes and its fetch has filled the buffer
with records.
+ final List<ConsumerRecord<String, String>> records = asList(
+ new ConsumerRecord<>("topic1", 0, 2, "key", "value"));
+ submittedEvents.get(0).markValidatePositionsComplete();
+ submittedEvents.get(0).completeSuccessfully();
+ doReturn(false).when(fetchBuffer).isEmpty();
+ doReturn(Fetch.forPartition(tp, records, true, new
OffsetAndMetadata(3, Optional.of(0), "")))
+ .when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
+ // The next poll returns the buffered records and submits no new poll
event (only the original one exists).
+ final ConsumerRecords<String, String> polled =
consumer.poll(Duration.ZERO);
+ assertEquals(1, polled.count());
+ verify(applicationEventHandler,
times(1)).add(isA(AsyncPollEvent.class));
+ }
+
@Test
public void testCommitInRebalanceCallback() {
consumer = newConsumer();
@@ -560,6 +685,10 @@ public class AsyncKafkaConsumerTest {
consumer.poll(Duration.ZERO);
assertDoesNotThrow(() -> consumer.poll(Duration.ZERO));
+
+ // When poll() returns records, the next fetch is pipelined so a fetch
request stays pending on the
+ // broker while the application processes the returned records.
+ verify(applicationEventHandler,
atLeastOnce()).add(isA(CreateFetchRequestsEvent.class));
}
/**
@@ -2501,12 +2630,6 @@ public class AsyncKafkaConsumerTest {
}).when(applicationEventHandler).add(ArgumentMatchers.isA(FetchCommittedOffsetsEvent.class));
}
- private void
completeFetchedCommittedOffsetApplicationEventExceptionally(Exception ex) {
- doThrow(ex)
- .when(applicationEventHandler)
- .addAndGet(any(FetchCommittedOffsetsEvent.class));
- }
-
private void completeUnsubscribeApplicationEventSuccessfully() {
doAnswer(invocation -> {
UnsubscribeEvent event = invocation.getArgument(0);
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
index 6c8e3570781..f32caebc909 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java
@@ -281,6 +281,45 @@ public class FetchRequestManagerTest {
}
}
+ /**
+ * A fetch response that carries no records must still wake up a thread
blocked on the fetch buffer.
+ */
+ @Test
+ public void testEmptyFetchResponseWakesUpBuffer() throws
InterruptedException {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ // Establish an incremental fetch session with a non-empty response
first, then consume the records
+ // and the resulting wakeup so the buffer below starts empty.
+ LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData>
partitions = new LinkedHashMap<>();
+ partitions.put(tidp0, new FetchResponseData.PartitionData()
+ .setPartitionIndex(tp0.partition())
+ .setHighWatermark(100)
+ .setRecords(records));
+ client.prepareResponse(FetchResponse.of(Errors.NONE, 0, 123,
partitions, List.of()));
+ assertEquals(1, sendFetches());
+ networkClientDelegate.poll(time.timer(0));
+ fetchRecords();
+ fetcher.fetchBuffer.awaitWakeup(time.timer(0));
+
+ // A consumer thread blocked waiting for data on the empty buffer.
+ Thread blockedOnBuffer = new Thread(() ->
fetcher.fetchBuffer.awaitWakeup(time.timer(3_600_000L)));
+ blockedOnBuffer.setDaemon(true);
+ blockedOnBuffer.start();
+
+ // An empty incremental fetch response (same session, no partition
data) must wake the thread blocked on the buffer.
+ client.prepareResponse(FetchResponse.of(Errors.NONE, 0, 123, new
LinkedHashMap<>(), List.of()));
+ assertEquals(1, sendFetches());
+ networkClientDelegate.poll(time.timer(0));
+
+ // On a successful run the blocked thread gets unblocked with the
response above so this join completes promptly.
+ // This timeout only caps how long we wait before declaring the wakeup
missing (failure).
+ blockedOnBuffer.join(2_000);
+ assertFalse(blockedOnBuffer.isAlive(), "Empty fetch response did not
wake the thread blocked on the fetch buffer");
+ }
+
@Test
public void testInflightFetchOnPendingPartitions() {
buildFetcher();
