You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by li...@apache.org on 2024/09/01 13:59:28 UTC
(kafka) branch trunk updated: KAFKA-17293: New consumer HeartbeatRequestManager should rediscover disconnected coordinator (#16844)
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 6a2789cf706 KAFKA-17293: New consumer HeartbeatRequestManager should rediscover disconnected coordinator (#16844)
6a2789cf706 is described below
commit 6a2789cf70665bc564bfa582fec17f42b93f3616
Author: TengYao Chi <ki...@gmail.com>
AuthorDate: Sun Sep 1 21:59:21 2024 +0800
KAFKA-17293: New consumer HeartbeatRequestManager should rediscover disconnected coordinator (#16844)
Reviewers: Lianet Magrans <li...@gmail.com>, TaiJuWu <tj...@gmail.com>
---
.../internals/AbstractHeartbeatRequestManager.java | 1 +
.../consumer/internals/CommitRequestManager.java | 9 +--------
.../internals/CoordinatorRequestManager.java | 16 ++++++++++++++++
.../internals/CommitRequestManagerTest.java | 12 ++++++++----
.../ConsumerHeartbeatRequestManagerTest.java | 22 ++++++++++++++++++++++
5 files changed, 48 insertions(+), 12 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java
index b310f6de1b9..7e923ca0e46 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java
@@ -307,6 +307,7 @@ public abstract class AbstractHeartbeatRequestManager<R extends AbstractResponse
resetHeartbeatState();
membershipManager().onHeartbeatFailure(exception instanceof RetriableException);
if (exception instanceof RetriableException) {
+ coordinatorRequestManager.handleCoordinatorDisconnect(exception, responseTimeMs);
String message = String.format("%s failed because of the retriable exception. Will retry in %s ms: %s",
heartbeatRequestName(),
heartbeatRequestState.remainingBackoffMs(responseTimeMs),
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
index 814250e0949..939d078ebaf 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
@@ -24,7 +24,6 @@ import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
import org.apache.kafka.clients.consumer.internals.metrics.OffsetCommitMetricsManager;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
@@ -565,12 +564,6 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
return pendingRequests.unsentOffsetFetches;
}
- private void handleCoordinatorDisconnect(Throwable exception, long currentTimeMs) {
- if (exception instanceof DisconnectException) {
- coordinatorRequestManager.markCoordinatorUnknown(exception.getMessage(), currentTimeMs);
- }
- }
-
/**
* Update latest member ID and epoch used by the member.
*
@@ -883,7 +876,7 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
} else {
log.debug("{} completed with error", requestDescription(), error);
onFailedAttempt(requestCompletionTimeMs);
- handleCoordinatorDisconnect(error, requestCompletionTimeMs);
+ coordinatorRequestManager.handleCoordinatorDisconnect(error, requestCompletionTimeMs);
future().completeExceptionally(error);
}
} catch (Throwable t) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
index 5144b41f574..a9e1bf46bed 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler
import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.message.FindCoordinatorRequestData;
@@ -122,6 +123,21 @@ public class CoordinatorRequestManager implements RequestManager {
});
}
+ /**
+ * Handles the disconnection of the current coordinator.
+ * This method checks if the given exception is an instance of {@link DisconnectException}.
+ * If so, it marks the coordinator as unknown, indicating that the client should
+ * attempt to discover a new coordinator. For any other exception type, no action is performed.
+ *
+ * @param exception The exception to handle, which was received as part of a request response.
+ * @param currentTimeMs The current time in milliseconds.
+ */
+ public void handleCoordinatorDisconnect(Throwable exception, long currentTimeMs) {
+ if (exception instanceof DisconnectException) {
+ markCoordinatorUnknown(exception.getMessage(), currentTimeMs);
+ }
+ }
+
/**
* Mark the current coordinator null.
*
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java
index 33d2d6aae1b..523abf4125c 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java
@@ -459,7 +459,7 @@ public class CommitRequestManagerTest {
// Commit should mark the coordinator unknown and fail with RetriableCommitFailedException.
assertTrue(commitResult.isDone());
assertFutureThrows(commitResult, RetriableCommitFailedException.class);
- assertCoordinatorDisconnect();
+ assertCoordinatorDisconnectHandling();
}
@Test
@@ -752,7 +752,7 @@ public class CommitRequestManagerTest {
// Request not completed just yet
assertFalse(result.isDone());
if (shouldRediscoverCoordinator) {
- assertCoordinatorDisconnect();
+ assertCoordinatorDisconnectOnCoordinatorError();
}
// Request should be retried with backoff.
@@ -782,7 +782,7 @@ public class CommitRequestManagerTest {
// Request not completed just yet, but should have marked the coordinator unknown
assertFalse(result.isDone());
- assertCoordinatorDisconnect();
+ assertCoordinatorDisconnectHandling();
time.sleep(retryBackoffMs);
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
@@ -922,7 +922,11 @@ public class CommitRequestManagerTest {
assertRetryBackOff(commitRequestManager, retryBackoffMs);
}
- private void assertCoordinatorDisconnect() {
+ private void assertCoordinatorDisconnectHandling() {
+ verify(coordinatorRequestManager).handleCoordinatorDisconnect(any(), anyLong());
+ }
+
+ private void assertCoordinatorDisconnectOnCoordinatorError() {
verify(coordinatorRequestManager).markCoordinatorUnknown(any(), anyLong());
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java
index f392f87489d..ee5c56296fd 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java
@@ -27,6 +27,7 @@ import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -380,6 +381,27 @@ public class ConsumerHeartbeatRequestManagerTest {
assertEquals(1, result.unsentRequests.size());
}
+ @Test
+ public void testDisconnect() {
+ createHeartbeatRequestStateWithZeroHeartbeatInterval();
+ NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, result.unsentRequests.size());
+ // Mimic disconnect
+ result.unsentRequests.get(0).handler().onFailure(time.milliseconds(), DisconnectException.INSTANCE);
+ verify(membershipManager).onHeartbeatFailure(true);
+ // Ensure that the coordinatorManager rediscovers the coordinator
+ verify(coordinatorRequestManager).handleCoordinatorDisconnect(any(), anyLong());
+ verify(backgroundEventHandler, never()).add(any());
+
+ time.sleep(DEFAULT_RETRY_BACKOFF_MS - 1);
+ result = heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(0, result.unsentRequests.size(), "No request should be generated before the backoff expires");
+
+ time.sleep(1);
+ result = heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, result.unsentRequests.size(), "A new request should be generated after the backoff expires");
+ }
+
@Test
public void testFailureOnFatalException() {
// The initial heartbeatInterval is set to 0