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 81c7cfe20807b..ba99d4416d614 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 @@ -78,6 +78,12 @@ public abstract class AbstractFetch implements Closeable { protected final BufferSupplier decompressionBufferSupplier; protected final Set nodesWithPendingFetchRequests; + // Whether the most recent empty result from prepareFetchRequests()/prepareCloseFetchSessionRequests() + // is a safe point to wake up the FetchBuffer immediately, rather than a state that can only change + // in response to an external event (for example, a metadata update, reconnect backoff expiration, + // or an in-flight response arriving). + private boolean emptyResultShouldWakeBuffer; + private final Map sessionHandlers; private final ApiVersions apiVersions; @@ -392,6 +398,10 @@ Node selectReadReplica(final TopicPartition partition, final Node leaderReplica, } protected Map prepareCloseFetchSessionRequests() { + // 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. + emptyResultShouldWakeBuffer = true; + final Cluster cluster = metadata.fetch(); Map fetchable = new HashMap<>(); @@ -433,8 +443,13 @@ protected Map prepareFetchRequests() List 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. + // 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, paused, or pending revocation/callback), the state will + // not change until some external event occurs, so an immediate wakeup would only busy-loop the + // caller rather than allowing the normal backoff to apply. + emptyResultShouldWakeBuffer = !subscriptions.fetchablePartitions(tp -> true).isEmpty(); return Collections.emptyMap(); } @@ -484,9 +499,24 @@ protected Map prepareFetchRequests() } } + // 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. + emptyResultShouldWakeBuffer = false; return convert(fetchable); } + /** + * Whether the most recent empty result from {@link #prepareFetchRequests()} or + * {@link #prepareCloseFetchSessionRequests()} represents a safe point to wake up the {@link FetchBuffer} + * immediately, as opposed to a state that will only change once some other event happens. + */ + protected boolean emptyResultShouldWakeBuffer() { + return emptyResultShouldWakeBuffer; + } + /** * This method converts {@link FetchSessionHandler.Builder} instances to * {@link FetchSessionHandler.FetchRequestData} instances. It intentionally forgoes use of the Java Collections 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 5b90fa2b48a8d..ad84d7c886e55 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 @@ -146,9 +146,12 @@ private PollResult pollInternal(FetchRequestPreparer fetchRequestPreparer, Map fetchRequests = fetchRequestPreparer.prepare(); 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 (emptyResultShouldWakeBuffer()) { + // 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; } 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 f32caebc90907..ee2bd6234d2e5 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,29 @@ public void testEmptyFetchResponseWakesUpBuffer() throws InterruptedException { assertFalse(blockedOnBuffer.isAlive(), "Empty fetch response 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 testInflightFetchOnPendingPartitions() { buildFetcher();