KAFKA-18469;KAFKA-18036: AsyncConsumer should request metadata update…

… if ListOffsetRequest encounters a retriable error (#18475) Reviewers: Lianet Magrans <[email protected]>
apache · Jan 13, 2025 · 4fcde45 · 4fcde45
1 parent 15e938e
commit 4fcde45
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 0 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java
@@ -578,6 +578,7 @@ private List<NetworkClientDelegate.UnsentRequest> buildListOffsetsRequests(
                     listOffsetsRequestState.globalResult.complete(listOffsetResult);
                 } else {
                     requestsToRetry.add(listOffsetsRequestState);
+                    metadata.requestUpdate(false);
                 }
             } else {
                 log.debug("ListOffsets request failed with error", error);

diff --git a/.../src/test/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManagerTest.java b/.../src/test/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManagerTest.java
@@ -286,6 +286,8 @@ public void testRequestFailsWithRetriableError_RetrySucceeds(Errors error) throw
         assertFalse(fetchOffsetsFuture.isDone());
         assertEquals(1, requestManager.requestsToRetry());
         assertEquals(0, requestManager.requestsToSend());
+        // A retriable error should be followed by a metadata update request
+        verify(metadata).requestUpdate(false);
 
         // Cluster metadata update. Failed requests should be retried and succeed
         mockSuccessfulRequest(Collections.singletonMap(TEST_PARTITION_1, LEADER_1));
@@ -384,6 +386,8 @@ public void testRequestPartiallyFailsWithRetriableError_RetrySucceeds() throws E
         assertFalse(fetchOffsetsFuture.isDone());
         assertEquals(1, requestManager.requestsToRetry());
         assertEquals(0, requestManager.requestsToSend());
+        // A retriable error should be followed by a metadata update request
+        verify(metadata).requestUpdate(false);
 
         // Cluster metadata update. Failed requests should be retried
         mockSuccessfulRequest(partitionLeaders);