You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2026/08/20 15:53:47 UTC
(kafka) branch 4.4 updated: KAFKA-20854 A more obvious busy loop due to KIP-909 (#23014)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new 642e0a5db02 KAFKA-20854 A more obvious busy loop due to KIP-909 (#23014)
642e0a5db02 is described below
commit 642e0a5db02867f526a1c820a0396935b8341b70
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Thu Aug 20 00:03:04 2026 +0800
KAFKA-20854 A more obvious busy loop due to KIP-909 (#23014)
`FetchRequestManager` currently wakes up the `FetchBuffer`
unconditionally whenever `prepareFetchRequests()` returns no requests
to send. This behavior was intended for one specific case: all
currently fetchable partitions already have buffered data waiting to be
consumed. However, the wakeup also happens for any other case that
produces an empty result, including states where no progress can be
made until some external event occurs:
- No partitions are fetchable at all (no assignment yet, invalid
positions, paused, pending revocation/callback). - The
partition's leader is unknown (`metadata.requestUpdate()` was just
triggered). - The target node is in its reconnect backoff window.
- The target node already has an in-flight fetch request outstanding.
- The target node hosts other partitions that already have buffered
data.
Because this wakeup bypasses the caller's normal `retryBackoffMs` delay,
the application thread and the background `ConsumerNetworkThread` can
enter a tight loop: `poll()` issues a fetch-request event, the
background thread finds nothing to send, wakes the buffer, `poll()`
returns immediately, and the cycle repeats without any meaningful
delay.
This is a pre-existing issue introduced with the non-blocking
`AsyncPollEvent` redesign in KAFKA-18376. However, KIP-909's async
bootstrap DNS resolution extends the "no node available" window from
effectively instantaneous to up to `bootstrapResolveTimeoutMs`, making
the busy loop much more visible.
Reviewers: Andrew Schofield <as...@confluent.io>, Lianet Magrans
<lm...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>
---
.../clients/consumer/internals/AbstractFetch.java | 77 ++++++++--
.../consumer/internals/AsyncKafkaConsumer.java | 48 +++---
.../consumer/internals/FetchRequestManager.java | 35 ++++-
.../kafka/clients/consumer/internals/Fetcher.java | 4 +-
.../consumer/internals/RequestManagers.java | 7 +-
.../consumer/internals/SubscriptionState.java | 11 ++
.../consumer/internals/AsyncKafkaConsumerTest.java | 11 +-
.../internals/FetchRequestManagerTest.java | 161 ++++++++++++++++++++-
8 files changed, 289 insertions(+), 65 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
index 81c7cfe2080..346349bc937 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
@@ -177,8 +177,6 @@ public abstract class AbstractFetch implements Closeable {
final Set<TopicPartition> partitions = new HashSet<>(responseData.keySet());
final FetchMetricsAggregator metricAggregator = new FetchMetricsAggregator(metricsManager, partitions);
- boolean needsWakeup = true;
-
Map<TopicPartition, Metadata.LeaderIdAndEpoch> partitionsWithUpdatedLeaderInfo = new HashMap<>();
for (Map.Entry<TopicPartition, FetchResponseData.PartitionData> entry : responseData.entrySet()) {
TopicPartition partition = entry.getKey();
@@ -225,14 +223,8 @@ public abstract class AbstractFetch implements Closeable {
metricAggregator,
fetchOffset);
fetchBuffer.add(completedFetch);
- needsWakeup = false;
}
- // "Wake" the fetch buffer on any response, even if it's empty, to allow the consumer to not block
- // indefinitely waiting on the fetch buffer to get data.
- if (needsWakeup)
- fetchBuffer.wakeup();
-
if (!partitionsWithUpdatedLeaderInfo.isEmpty()) {
List<Node> leaderNodes = new ArrayList<>();
@@ -298,6 +290,11 @@ public abstract class AbstractFetch implements Closeable {
private void removePendingFetchRequest(Node fetchTarget, int sessionId) {
log.debug("Removing pending request for fetch session: {} for node: {}", sessionId, fetchTarget);
nodesWithPendingFetchRequests.remove(fetchTarget.id());
+
+ // Wake the buffer whenever a node stops having a request in flight, whatever the outcome was: data, an
+ // empty response, a fetch session error, or a failure. This ensures the caller is not left waiting on a
+ // wakeup that only a completed request could have delivered.
+ fetchBuffer.wakeup();
}
/**
@@ -391,7 +388,7 @@ public abstract class AbstractFetch implements Closeable {
}
}
- protected Map<Node, FetchSessionHandler.FetchRequestData> prepareCloseFetchSessionRequests() {
+ protected FetchRequestPreparationResult prepareCloseFetchSessionRequests() {
final Cluster cluster = metadata.fetch();
Map<Node, FetchSessionHandler.Builder> fetchable = new HashMap<>();
@@ -411,14 +408,16 @@ public abstract class AbstractFetch implements Closeable {
fetchable.put(fetchTarget, sessionHandler.newBuilder());
});
- return convert(fetchable);
+ // Closing is a one-shot operation, not part of the steady-state polling loop, so always waking the
+ // buffer here is safe even if this ends up empty.
+ return new FetchRequestPreparationResult(convert(fetchable), true);
}
/**
* Create fetch requests for all nodes for which we have assigned partitions
* that have no existing requests in flight.
*/
- protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
+ protected FetchRequestPreparationResult prepareFetchRequests() {
// Update metrics in case there was an assignment change
metricsManager.maybeUpdateAssignment(subscriptions);
@@ -433,9 +432,14 @@ public abstract class AbstractFetch implements Closeable {
List<TopicPartition> unbuffered = fetchablePartitions(buffered);
if (unbuffered.isEmpty()) {
- // If there are no partitions that don't already have data locally buffered, there's no need to issue
- // any fetch requests at the present time.
- return Collections.emptyMap();
+ // If every currently fetchable partition already has buffered data, there is no need to issue
+ // additional fetch requests. This is a safe point to wake the buffer immediately because progress
+ // can be made by consuming the buffered data. If no partitions are fetchable at all (for example,
+ // no assignment yet, invalid positions, or paused), the state will not change until some external
+ // event occurs, so an immediate wakeup would only busy-loop the caller rather than allowing it
+ // to remain parked until bounded by other mechanisms (such as heartbeat interval or poll timeout).
+ boolean canWakeBufferIfNoFetchRequestsToSend = subscriptions.hasFetchablePartitions(tp -> true);
+ return new FetchRequestPreparationResult(Map.of(), canWakeBufferIfNoFetchRequestsToSend);
}
Set<Integer> bufferedNodes = bufferedNodes(buffered, currentTimeMs);
@@ -484,7 +488,48 @@ public abstract class AbstractFetch implements Closeable {
}
}
- return convert(fetchable);
+ // If every fetchable-but-unbuffered partition was skipped (for example, due to reconnect backoff,
+ // an in-flight request, or its node already hosting buffered partitions), the state will only
+ // change over time. An immediate wakeup would therefore just busy-loop the caller instead of
+ // respecting its normal backoff. This case is only relevant when fetchable partitions exist but
+ // the resulting request map is empty; otherwise the caller ignores this flag.
+ return new FetchRequestPreparationResult(convert(fetchable), false);
+ }
+
+ /**
+ * The result of preparing fetch requests via {@link #prepareFetchRequests()} or
+ * {@link #prepareCloseFetchSessionRequests()}.
+ */
+ protected static final class FetchRequestPreparationResult {
+
+ private final Map<Node, FetchSessionHandler.FetchRequestData> requests;
+ private final boolean canWakeBufferIfNoFetchRequestsToSend;
+
+ FetchRequestPreparationResult(
+ Map<Node, FetchSessionHandler.FetchRequestData> requests,
+ boolean canWakeBufferIfNoFetchRequestsToSend
+ ) {
+ this.requests = requests;
+ this.canWakeBufferIfNoFetchRequestsToSend = canWakeBufferIfNoFetchRequestsToSend;
+ }
+
+ /**
+ * {@link Map} of {@link Node nodes} to the {@link FetchSessionHandler.FetchRequestData} to send them,
+ * empty if there is nothing to fetch right now.
+ */
+ Map<Node, FetchSessionHandler.FetchRequestData> requests() {
+ return requests;
+ }
+
+ /**
+ * Whether, if {@link #requests()} is empty, this is a safe point to wake up the {@link FetchBuffer}
+ * immediately, as opposed to a state that will only change once some other event happens (for example, a
+ * metadata update, reconnect backoff expiration, or an in-flight response arriving). Ignored when
+ * {@link #requests()} is non-empty.
+ */
+ boolean canWakeBufferIfNoFetchRequestsToSend() {
+ return canWakeBufferIfNoFetchRequestsToSend;
+ }
}
/**
@@ -648,4 +693,4 @@ public abstract class AbstractFetch implements Closeable {
*/
void handle(Node target, FetchSessionHandler.FetchRequestData data, T response);
}
-}
\ No newline at end of file
+}
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 253fde03913..496248f7d58 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
@@ -1984,28 +1984,29 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
return fetch;
}
- long pollTimeout = isCommittedOffsetsManagementEnabled()
- ? Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs())
- : timer.remainingMs();
- // With the non-blocking poll design, it's possible that at this point the background thread is
- // concurrently working to update positions. Therefore, a _copy_ of the current assignment is retrieved
- // and iterated looking for any partitions with invalid positions. This is done to avoid being stuck
- // in poll for an unnecessarily long amount of time if we are missing some positions since the offset
- // lookup may be backing off after a failure.
- if (pollTimeout > retryBackoffMs) {
- Set<TopicPartition> partitions = subscriptions.assignedPartitions();
+ long pollTimeout = Math.min(applicationEventHandler.maximumTimeToWait(), timer.remainingMs());
- if (partitions.isEmpty()) {
- // If there aren't any assigned partitions, this could mean that this consumer's group membership
- // has not been established or assignments have been removed and not yet reassigned. In either case,
- // reduce the poll time for the fetch buffer wait.
+ // Bound the wait when background progress may make fetching possible soon.
+ // Use the current application-thread state to avoid relying on stale state from the network thread.
+ if (pollTimeout > retryBackoffMs) {
+ if (subscriptions.numAssignedPartitions() == 0) {
+ // If there are no assigned partitions, reduce the fetch buffer wait time. This may happen when
+ // group membership has not been established yet, assignments have been revoked but not reassigned,
+ // bootstrap DNS resolution is still in progress, or manual assignment has not happened yet.
+ pollTimeout = retryBackoffMs;
+ } else if (!subscriptions.hasAllFetchPositions()) {
+ // If some partitions do not have valid positions, the background thread may still be resolving them,
+ // for example by fetching committed offsets, looking up offsets by timestamp, or backing off after a
+ // failure. Reduce the wait time so the application thread can consume data promptly once positions are
+ // resolved.
pollTimeout = retryBackoffMs;
} else {
- for (TopicPartition tp : partitions) {
- if (!subscriptions.hasValidPosition(tp)) {
- pollTimeout = retryBackoffMs;
- break;
- }
+ Set<TopicPartition> buffered = fetchBuffer.bufferedPartitions();
+ if (subscriptions.hasFetchablePartitions(tp -> !buffered.contains(tp))) {
+ // If any fetchable partition has no buffered data, it may have been skipped due to reconnect
+ // backoff, an in-flight request, or a missing leader. Bound the wait so the application thread
+ // can retry once the condition clears.
+ pollTimeout = retryBackoffMs;
}
}
}
@@ -2102,15 +2103,6 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
return true;
}
- /**
- *
- * Indicates if the consumer is using the Kafka-based offset management strategy,
- * according to config {@link CommonClientConfigs#GROUP_ID_CONFIG}
- */
- private boolean isCommittedOffsetsManagementEnabled() {
- return groupMetadata.get().isPresent();
- }
-
/**
* This method signals the background thread to {@link CreateFetchRequestsEvent create fetch requests} for the
* pre-fetch case, i.e. right before {@link #poll(Duration)} exits. In the pre-fetch case, the application thread
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java
index 5b90fa2b48a..dc28aaf4715 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java
@@ -44,6 +44,7 @@ import java.util.stream.Collectors;
public class FetchRequestManager extends AbstractFetch implements RequestManager {
private final NetworkClientDelegate networkClientDelegate;
+ private final long retryBackoffMs;
private CompletableFuture<Void> pendingFetchRequestFuture;
FetchRequestManager(final LogContext logContext,
@@ -54,9 +55,11 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
final FetchBuffer fetchBuffer,
final FetchMetricsManager metricsManager,
final NetworkClientDelegate networkClientDelegate,
- final ApiVersions apiVersions) {
+ final ApiVersions apiVersions,
+ final long retryBackoffMs) {
super(logContext, metadata, subscriptions, fetchConfig, fetchBuffer, metricsManager, time, apiVersions);
this.networkClientDelegate = networkClientDelegate;
+ this.retryBackoffMs = retryBackoffMs;
}
@Override
@@ -69,6 +72,18 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
networkClientDelegate.maybeThrowAuthFailure(node);
}
+ /**
+ * {@inheritDoc}
+ *
+ * If any request is in flight, its completion will wake the application thread regardless of the outcome, so
+ * no separate bound is needed. Otherwise, the application thread's wait is bounded by {@code retryBackoffMs}
+ * so it can re-evaluate subscription state changes promptly.
+ */
+ @Override
+ public long maximumTimeToWait(long currentTimeMs) {
+ return nodesWithPendingFetchRequests.isEmpty() ? retryBackoffMs : Long.MAX_VALUE;
+ }
+
/**
* Signals the {@link Consumer} wants requests be created for the broker nodes to fetch the next
* batch of records.
@@ -128,8 +143,8 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
* Creates the {@link PollResult poll result} that contains a list of zero or more
* {@link FetchRequest.Builder fetch requests}.
*
- * @param fetchRequestPreparer {@link FetchRequestPreparer} to generate a {@link Map} of {@link Node nodes}
- * to their {@link FetchSessionHandler.FetchRequestData}
+ * @param fetchRequestPreparer {@link FetchRequestPreparer} to generate a {@link FetchRequestPreparationResult}
+ * mapping {@link Node nodes} to their {@link FetchSessionHandler.FetchRequestData}
* @param successHandler {@link ResponseHandler Handler for successful responses}
* @param errorHandler {@link ResponseHandler Handler for failure responses}
* @return {@link PollResult}
@@ -143,12 +158,16 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
}
try {
- Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = fetchRequestPreparer.prepare();
+ FetchRequestPreparationResult result = fetchRequestPreparer.prepare();
+ Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = result.requests();
if (fetchRequests.isEmpty()) {
- // If there's nothing to fetch, wake up the FetchBuffer so it doesn't needlessly wait for a wakeup
- // that won't come until the data in the fetch buffer is consumed.
- fetchBuffer.wakeup();
+ if (result.canWakeBufferIfNoFetchRequestsToSend()) {
+ // If there's nothing to fetch because every fetchable partition already has buffered data,
+ // wake up the FetchBuffer so it doesn't needlessly wait for a wakeup that won't come until
+ // the data in the fetch buffer is consumed.
+ fetchBuffer.wakeup();
+ }
pendingFetchRequestFuture.complete(null);
return PollResult.EMPTY;
}
@@ -186,6 +205,6 @@ public class FetchRequestManager extends AbstractFetch implements RequestManager
@FunctionalInterface
protected interface FetchRequestPreparer {
- Map<Node, FetchSessionHandler.FetchRequestData> prepare();
+ FetchRequestPreparationResult prepare();
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index eaa9a4628bc..f5990457548 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -103,7 +103,7 @@ public class Fetcher<K, V> extends AbstractFetch {
* @return number of fetches sent
*/
public synchronized int sendFetches() {
- final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests();
+ final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests().requests();
sendFetchesInternal(
fetchRequests,
(fetchTarget, data, clientResponse) -> {
@@ -121,7 +121,7 @@ public class Fetcher<K, V> extends AbstractFetch {
protected void maybeCloseFetchSessions(final Timer timer) {
final List<RequestFuture<ClientResponse>> requestFutures = sendFetchesInternal(
- prepareCloseFetchSessionRequests(),
+ prepareCloseFetchSessionRequests().requests(),
this::handleCloseFetchSessionSuccess,
this::handleCloseFetchSessionFailure
);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index e93da12cda6..f8553625f05 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -181,8 +181,8 @@ public class RequestManagers implements Closeable {
protected RequestManagers create() {
final NetworkClientDelegate networkClientDelegate = networkClientDelegateSupplier.get();
final FetchConfig fetchConfig = new FetchConfig(config);
- long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
- long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
+ final long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
+ final long retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
final int requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
final int defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
@@ -194,7 +194,8 @@ public class RequestManagers implements Closeable {
fetchBuffer,
fetchMetricsManager,
networkClientDelegate,
- apiVersions);
+ apiVersions,
+ retryBackoffMs);
final TopicMetadataRequestManager topic = new TopicMetadataRequestManager(
logContext,
time,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
index 56d38b0747a..24bcd56e188 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java
@@ -496,6 +496,17 @@ public class SubscriptionState {
return result;
}
+ public synchronized boolean hasFetchablePartitions(Predicate<TopicPartition> isAvailable) {
+ for (Map.Entry<TopicPartition, TopicPartitionState> entry : assignment.partitionStateMap().entrySet()) {
+ TopicPartition topicPartition = entry.getKey();
+ if ((subscriptionType.equals(SubscriptionType.AUTO_TOPICS_SHARE) || isFetchableAndSubscribed(topicPartition, entry.getValue()))
+ && isAvailable.test(topicPartition)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Check if the partition is fetchable.
* If the consumer has explicitly subscribed to a list of topic names,
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 59d9f345f56..5fa303f5658 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
@@ -259,7 +259,6 @@ public class AsyncKafkaConsumerTest {
ConsumerInterceptors<String, String> interceptors,
ConsumerRebalanceListenerInvoker rebalanceListenerInvoker,
SubscriptionState subscriptions) {
- long retryBackoffMs = 100L;
int requestTimeoutMs = 30000;
int defaultApiTimeoutMs = 1000;
return new AsyncKafkaConsumer<>(
@@ -279,7 +278,7 @@ public class AsyncKafkaConsumerTest {
metrics,
subscriptions,
metadata,
- retryBackoffMs,
+ 100L,
requestTimeoutMs,
defaultApiTimeoutMs,
"group-id",
@@ -2129,7 +2128,7 @@ public class AsyncKafkaConsumerTest {
final TopicPartition tp = new TopicPartition("topic1", 0);
- // Manual assignment with valid position so pollForFetches() does not shrink pollTimeout to retryBackoffMs.
+ // Manual assignment with a valid position, so nothing here overrides the mocked maximumTimeToWait() below.
subscriptions.assignFromUser(singleton(tp));
subscriptions.seek(tp, 0);
@@ -2139,6 +2138,8 @@ public class AsyncKafkaConsumerTest {
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ // The partition is fetchable but already buffered, so pollForFetches should not bound the timeout.
+ doReturn(singleton(tp)).when(fetchBuffer).bufferedPartitions();
// Capture the Timer passed to awaitWakeup so we can assert it was given the full
// poll timeout, i.e. no busy loop. Also advance mock time by the timer's remaining ms so the
@@ -2479,6 +2480,8 @@ public class AsyncKafkaConsumerTest {
client.prepareResponseFrom(result, coordinator);
SubscriptionState subscriptionState = mock(SubscriptionState.class);
+ SubscriptionPattern pattern = new SubscriptionPattern("t*");
+ when(subscriptionState.subscriptionPattern()).thenReturn(pattern);
consumer = new AsyncKafkaConsumer<>(
new LogContext(),
@@ -2492,9 +2495,7 @@ public class AsyncKafkaConsumerTest {
);
completeTopicRe2JPatternSubscriptionChangeEventSuccessfully();
- SubscriptionPattern pattern = new SubscriptionPattern("t*");
consumer.subscribe(pattern);
- when(subscriptionState.subscriptionPattern()).thenReturn(pattern);
TestUtils.waitForCondition(() -> {
try {
// The request is generated in the background thread so allow for that
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 f32caebc909..85b34a3e48d 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
@@ -320,6 +320,159 @@ public class FetchRequestManagerTest {
assertFalse(blockedOnBuffer.isAlive(), "Empty fetch response did not wake the thread blocked on the fetch buffer");
}
+ @Test
+ public void testFailedFetchResponseWakesUpBuffer() throws InterruptedException {
+ // The response body is irrelevant: it is discarded once the response is marked as disconnected.
+ assertRequestCompletionWakesUpBuffer(() -> client.prepareResponse(
+ fullFetchResponse(tidp0, records, Errors.NONE, 100L, 0), true));
+ }
+
+ @Test
+ public void testFetchSessionErrorResponseWakesUpBuffer() throws InterruptedException {
+ assertRequestCompletionWakesUpBuffer(() -> client.prepareResponse(FetchResponse.of(
+ Errors.FETCH_SESSION_ID_NOT_FOUND, 0, INVALID_SESSION_ID, new LinkedHashMap<>(), List.of())));
+ }
+
+ private void assertRequestCompletionWakesUpBuffer(Runnable prepareResponse) throws InterruptedException {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ assertEquals(1, sendFetches());
+
+ // 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();
+
+ prepareResponse.run();
+ networkClientDelegate.poll(time.timer(0));
+
+ blockedOnBuffer.join(2_000);
+ assertFalse(blockedOnBuffer.isAlive(), "Completed fetch request did not wake the thread blocked on the fetch buffer");
+ }
+
+ @Test
+ public void testNoFetchablePartitionsDoesNotWakeUpBuffer() throws InterruptedException {
+ buildFetcher();
+
+ // 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();
+
+ // Simulate a network thread cycle that finds nothing to fetch.
+ assertEquals(0, sendFetches());
+
+ // The thread must still be blocked: an eager wakeup here would busy-loop the caller.
+ blockedOnBuffer.join(500);
+ assertTrue(blockedOnBuffer.isAlive(),
+ "Empty fetch result with no fetchable partitions must not wake the thread blocked on the fetch buffer");
+
+ // Clean up: explicitly wake so the daemon thread can exit instead of leaking as a live thread.
+ fetcher.fetchBuffer.wakeup();
+ blockedOnBuffer.join(2_000);
+ assertFalse(blockedOnBuffer.isAlive());
+ }
+
+ @Test
+ public void testMaximumTimeToWaitUnboundedWhenFetchSent() {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ assertEquals(1, sendFetches());
+ assertEquals(Long.MAX_VALUE, fetcher.maximumTimeToWait(time.milliseconds()));
+ }
+
+ @Test
+ public void testMaximumTimeToWaitBoundedWhenNoInflightRequest() {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ // Fetch data for tp0, but leave it buffered (unconsumed) so the next prepare() finds every fetchable
+ // partition already buffered. With no in-flight request, maximumTimeToWait is bounded.
+ client.prepareResponse(fullFetchResponse(tidp0, records, Errors.NONE, 100L, 0));
+ assertEquals(1, sendFetches());
+ networkClientDelegate.poll(time.timer(0));
+ assertTrue(fetcher.hasCompletedFetches());
+
+ assertEquals(0, sendFetches());
+ assertEquals(retryBackoffMs, fetcher.maximumTimeToWait(time.milliseconds()));
+ }
+
+ @Test
+ public void testMaximumTimeToWaitUnboundedWhenPartitionsSkippedDueToInflight() {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+
+ // A fetch request is sent successfully; maximumTimeToWait remains unbounded.
+ assertEquals(1, sendFetches());
+ assertEquals(Long.MAX_VALUE, fetcher.maximumTimeToWait(time.milliseconds()));
+
+ // The in-flight request blocks the node, so the next prepare() skips the partition. maximumTimeToWait
+ // stays unbounded: the in-flight request's completion will wake the buffer regardless.
+ assertEquals(0, sendFetches());
+ assertEquals(Long.MAX_VALUE, fetcher.maximumTimeToWait(time.milliseconds()));
+
+ // Complete the in-flight request and consume the buffered data.
+ client.prepareResponse(fullFetchResponse(tidp0, records, Errors.NONE, 100L, 0));
+ networkClientDelegate.poll(time.timer(0));
+ fetchRecords();
+
+ // A new fetch request can now be sent; maximumTimeToWait remains unbounded.
+ assertEquals(1, sendFetches());
+ assertEquals(Long.MAX_VALUE, fetcher.maximumTimeToWait(time.milliseconds()));
+ }
+
+ @Test
+ public void testMaximumTimeToWaitBoundedWhenPartitionsSkippedDueToBackoff() {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+ Node node = metadata.fetch().leaderFor(tp0);
+
+ client.backoff(node, 500);
+ assertEquals(0, sendFetches());
+ assertEquals(retryBackoffMs, fetcher.maximumTimeToWait(time.milliseconds()));
+
+ // Once the backoff clears, a fetch request can be sent and maximumTimeToWait reverts to unbounded.
+ time.sleep(500);
+ assertEquals(1, sendFetches());
+ assertEquals(Long.MAX_VALUE, fetcher.maximumTimeToWait(time.milliseconds()));
+ }
+
+ @Test
+ public void testPartitionsSkippedDueToBackoffDoesNotWakeUpBuffer() throws InterruptedException {
+ buildFetcher();
+
+ assignFromUser(singleton(tp0));
+ subscriptions.seek(tp0, 0);
+ Node node = metadata.fetch().leaderFor(tp0);
+ client.backoff(node, 500);
+
+ Thread blockedOnBuffer = new Thread(() -> fetcher.fetchBuffer.awaitWakeup(time.timer(3_600_000L)));
+ blockedOnBuffer.setDaemon(true);
+ blockedOnBuffer.start();
+
+ assertEquals(0, sendFetches());
+
+ blockedOnBuffer.join(500);
+ assertTrue(blockedOnBuffer.isAlive(),
+ "Fetch skipped due to reconnect backoff must not wake the thread blocked on the fetch buffer");
+
+ fetcher.fetchBuffer.wakeup();
+ blockedOnBuffer.join(2_000);
+ assertFalse(blockedOnBuffer.isAlive());
+ }
+
@Test
public void testInflightFetchOnPendingPartitions() {
buildFetcher();
@@ -4180,7 +4333,8 @@ public class FetchRequestManagerTest {
metricsManager,
networkClientDelegate,
fetchCollector,
- apiVersions));
+ apiVersions,
+ retryBackoffMs));
ConsumerNetworkClient consumerNetworkClient = new ConsumerNetworkClient(
logContext,
client,
@@ -4242,8 +4396,9 @@ public class FetchRequestManagerTest {
FetchMetricsManager metricsManager,
NetworkClientDelegate networkClientDelegate,
FetchCollector<K, V> fetchCollector,
- ApiVersions apiVersions) {
- super(logContext, time, metadata, subscriptions, fetchConfig, fetchBuffer, metricsManager, networkClientDelegate, apiVersions);
+ ApiVersions apiVersions,
+ long retryBackoffMs) {
+ super(logContext, time, metadata, subscriptions, fetchConfig, fetchBuffer, metricsManager, networkClientDelegate, apiVersions, retryBackoffMs);
this.fetchCollector = fetchCollector;
}