KAFKA-20854 A more obvious busy loop due to KIP-909(WIP)

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<Integer> 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<Integer, FetchSessionHandler> sessionHandlers;
 
     private final ApiVersions apiVersions;
@@ -392,6 +398,10 @@ Node selectReadReplica(final TopicPartition partition, final Node leaderReplica,
     }
 
     protected Map<Node, FetchSessionHandler.FetchRequestData> 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<Node, FetchSessionHandler.Builder> fetchable = new HashMap<>();
 
@@ -433,8 +443,13 @@ protected Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests()
         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.
+            // 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<Node, FetchSessionHandler.FetchRequestData> 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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java

@@ -146,9 +146,12 @@ private PollResult pollInternal(FetchRequestPreparer fetchRequestPreparer,
             Map<Node, FetchSessionHandler.FetchRequestData> 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;
             }

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();