You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/07/19 08:08:51 UTC
(kafka) branch trunk updated: KAFKA-16984: Complete consumer leave on response to leave request (#16569)
This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 931bb62a234 KAFKA-16984: Complete consumer leave on response to leave request (#16569)
931bb62a234 is described below
commit 931bb62a2340a563e598e522b196c07b870c285c
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Fri Jul 19 04:08:40 2024 -0400
KAFKA-16984: Complete consumer leave on response to leave request (#16569)
Improvement to ensure that the consumer unsubscribe operation waits for a response to the leave group request before moving on to close the consumer. This makes it consistent with the behaviour of the legacy consumer.
This will avoid undesired interactions on close, that triggers a leave group, and shuts down the network thread when it completes (which before this PR could led to responses to disconnected clients).
Note that this PR does not change the transitions of the state machine on leave group, only the completion of the leave group future.
Reviewers: Lucas Brutschy <lb...@confluent.io>, Kirk True <kt...@confluent.io>
---
.../internals/HeartbeatRequestManager.java | 5 +-
.../consumer/internals/MembershipManager.java | 8 +-
.../consumer/internals/MembershipManagerImpl.java | 75 +++++--
.../internals/HeartbeatRequestManagerTest.java | 21 +-
.../internals/MembershipManagerImplTest.java | 221 ++++++++++++---------
5 files changed, 203 insertions(+), 127 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
index 3dacbec8ae9..8a62051b2f4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java
@@ -275,7 +275,7 @@ public class HeartbeatRequestManager implements RequestManager {
final boolean ignoreResponse) {
NetworkClientDelegate.UnsentRequest request = makeHeartbeatRequest(ignoreResponse);
heartbeatRequestState.onSendAttempt(currentTimeMs);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
metricsManager.recordHeartbeatSentMs(currentTimeMs);
heartbeatRequestState.resetTimer();
return request;
@@ -318,6 +318,7 @@ public class HeartbeatRequestManager implements RequestManager {
private void onFailure(final Throwable exception, final long responseTimeMs) {
this.heartbeatRequestState.onFailedAttempt(responseTimeMs);
this.heartbeatState.reset();
+ membershipManager.onHeartbeatFailure(exception instanceof RetriableException);
if (exception instanceof RetriableException) {
String message = String.format("GroupHeartbeatRequest failed because of the retriable exception. " +
"Will retry in %s ms: %s",
@@ -348,7 +349,7 @@ public class HeartbeatRequestManager implements RequestManager {
this.heartbeatState.reset();
this.heartbeatRequestState.onFailedAttempt(currentTimeMs);
- membershipManager.onHeartbeatFailure();
+ membershipManager.onHeartbeatFailure(false);
switch (error) {
case NOT_COORDINATOR:
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
index c33ac396d23..7e2d757042b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManager.java
@@ -77,15 +77,17 @@ public interface MembershipManager extends RequestManager {
/**
* Notify the member that an error heartbeat response was received.
+ *
+ * @param retriable True if the request failed with a retriable error.
*/
- void onHeartbeatFailure();
+ void onHeartbeatFailure(boolean retriable);
/**
- * Update state when a heartbeat is sent out. This will transition out of the states that end
+ * Update state when a heartbeat is generated. This will transition out of the states that end
* when a heartbeat request is sent, without waiting for a response (ex.
* {@link MemberState#ACKNOWLEDGING} and {@link MemberState#LEAVING}).
*/
- void onHeartbeatRequestSent();
+ void onHeartbeatRequestGenerated();
/**
* Transition out of the {@link MemberState#LEAVING} state even if the heartbeat was not sent
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 702460e3493..a0e3a06d008 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -426,6 +426,11 @@ public class MembershipManagerImpl implements MembershipManager {
"already leaving the group.", memberId, memberEpoch);
return;
}
+ if (state == MemberState.UNSUBSCRIBED && maybeCompleteLeaveInProgress()) {
+ log.debug("Member {} with epoch {} received a successful response to the heartbeat " +
+ "to leave the group and completed the leave operation. ", memberId, memberEpoch);
+ return;
+ }
if (isNotInGroup()) {
log.debug("Ignoring heartbeat response received from broker. Member {} is in {} state" +
" so it's not a member of the group. ", memberId, state);
@@ -459,8 +464,29 @@ public class MembershipManagerImpl implements MembershipManager {
}
@Override
- public void onHeartbeatFailure() {
- metricsManager.maybeRecordRebalanceFailed();
+ public void onHeartbeatFailure(boolean retriable) {
+ if (!retriable) {
+ metricsManager.maybeRecordRebalanceFailed();
+ }
+ // The leave group request is sent out once (not retried), so we should complete the leave
+ // operation once the request completes, regardless of the response.
+ if (state == MemberState.UNSUBSCRIBED && maybeCompleteLeaveInProgress()) {
+ log.warn("Member {} with epoch {} received a failed response to the heartbeat to " +
+ "leave the group and completed the leave operation. ", memberId, memberEpoch);
+ }
+ }
+
+ /**
+ * Complete the leave in progress (if any). This is expected to be used to complete the leave
+ * in progress when a member receives the response to the leave heartbeat.
+ */
+ private boolean maybeCompleteLeaveInProgress() {
+ if (leaveGroupInProgress.isPresent()) {
+ leaveGroupInProgress.get().complete(null);
+ leaveGroupInProgress = Optional.empty();
+ return true;
+ }
+ return false;
}
/**
@@ -523,8 +549,8 @@ public class MembershipManagerImpl implements MembershipManager {
public void transitionToFenced() {
if (state == MemberState.PREPARE_LEAVING) {
log.info("Member {} with epoch {} got fenced but it is already preparing to leave " +
- "the group, so it will stop sending heartbeat and won't attempt to rejoin.",
- memberId, memberEpoch);
+ "the group, so it will stop sending heartbeat and won't attempt to send the " +
+ "leave request or rejoin.", memberId, memberEpoch);
// Briefly transition to LEAVING to ensure all required actions are applied even
// though there is no need to send a leave group heartbeat (ex. clear epoch and
// notify epoch listeners). Then transition to UNSUBSCRIBED, ensuring that the member
@@ -532,12 +558,15 @@ public class MembershipManagerImpl implements MembershipManager {
// sending heartbeats while it completes the ongoing leaving operation.
transitionToSendingLeaveGroup(false);
transitionTo(MemberState.UNSUBSCRIBED);
+ maybeCompleteLeaveInProgress();
return;
}
if (state == MemberState.LEAVING) {
- log.debug("Member {} with epoch {} got fenced but it is already leaving the group " +
- "with state {}, so it won't attempt to rejoin.", memberId, memberEpoch, state);
+ log.debug("Member {} with epoch {} got fenced before sending leave group heartbeat. " +
+ "It will not send the leave request and won't attempt to rejoin.", memberId, memberEpoch);
+ transitionTo(MemberState.UNSUBSCRIBED);
+ maybeCompleteLeaveInProgress();
return;
}
if (state == MemberState.UNSUBSCRIBED) {
@@ -587,6 +616,14 @@ public class MembershipManagerImpl implements MembershipManager {
return;
}
+ if (previousState == MemberState.LEAVING || previousState == MemberState.PREPARE_LEAVING) {
+ log.info("Member {} with epoch {} was leaving the group with state {} when it got a " +
+ "fatal error from the broker. It will discard the ongoing leave and remain in " +
+ "fatal state.", memberId, memberEpoch, previousState);
+ maybeCompleteLeaveInProgress();
+ return;
+ }
+
// Release assignment
CompletableFuture<Void> callbackResult = invokeOnPartitionsLostCallback(subscriptions.assignedPartitions());
callbackResult.whenComplete((result, error) -> {
@@ -799,7 +836,7 @@ public class MembershipManagerImpl implements MembershipManager {
* {@inheritDoc}
*/
@Override
- public void onHeartbeatRequestSent() {
+ public void onHeartbeatRequestGenerated() {
MemberState state = state();
if (state == MemberState.ACKNOWLEDGING) {
if (targetAssignmentReconciled()) {
@@ -812,13 +849,13 @@ public class MembershipManagerImpl implements MembershipManager {
}
} else if (state == MemberState.LEAVING) {
if (isPollTimerExpired) {
- log.debug("Member {} sent heartbeat to leave due to expired poll timer. It will " +
+ log.debug("Member {} with epoch {} generated the heartbeat to leave due to expired poll timer. It will " +
"remain stale (no heartbeat) until it rejoins the group on the next consumer " +
- "poll.", memberId);
+ "poll.", memberId, memberEpoch);
transitionToStale();
} else {
- log.debug("Member {} sent heartbeat to leave group.", memberId);
- transitionToUnsubscribed();
+ log.debug("Member {} with epoch {} generated the heartbeat to leave the group.", memberId, memberEpoch);
+ transitionTo(MemberState.UNSUBSCRIBED);
}
}
}
@@ -829,18 +866,14 @@ public class MembershipManagerImpl implements MembershipManager {
@Override
public void onHeartbeatRequestSkipped() {
if (state == MemberState.LEAVING) {
- log.debug("Heartbeat for leaving group could not be sent. Member {} with epoch {} will transition to {}.",
- memberId, memberEpoch, MemberState.UNSUBSCRIBED);
- transitionToUnsubscribed();
+ log.warn("Heartbeat to leave group cannot be sent (most probably due to coordinator " +
+ "not known/available). Member {} with epoch {} will transition to {}.",
+ memberId, memberEpoch, MemberState.UNSUBSCRIBED);
+ transitionTo(MemberState.UNSUBSCRIBED);
+ maybeCompleteLeaveInProgress();
}
}
- private void transitionToUnsubscribed() {
- transitionTo(MemberState.UNSUBSCRIBED);
- leaveGroupInProgress.get().complete(null);
- leaveGroupInProgress = Optional.empty();
- }
-
/**
* @return True if there are no assignments waiting to be resolved from metadata or reconciled.
*/
@@ -1128,7 +1161,7 @@ public class MembershipManagerImpl implements MembershipManager {
* <li>Try to find topic names in the metadata cache</li>
* <li>For topics not found in metadata, try to find names in the local topic names cache
* (contains topic id and names currently assigned and resolved)</li>
- * <li>If there are topics that are not in metadata cache or in the local cached
+ * <li>If there are topics that are not in metadata cache or in the local cache
* of topic names assigned to this member, request a metadata update, and continue
* resolving names as the cache is updated.
* </li>
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
index bb2680b2e0b..77b7495b492 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
@@ -306,7 +306,7 @@ public class HeartbeatRequestManagerTest {
assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS, result.timeUntilNextPollMs);
assertEquals(DEFAULT_HEARTBEAT_INTERVAL_MS, heartbeatRequestManager.maximumTimeToWait(time.milliseconds()));
// Membership manager updated (to transition out of the heartbeating state)
- verify(membershipManager).onHeartbeatRequestSent();
+ verify(membershipManager).onHeartbeatRequestGenerated();
}
@Test
@@ -319,6 +319,8 @@ public class HeartbeatRequestManagerTest {
assertEquals(1, result.unsentRequests.size());
// Mimic network timeout
result.unsentRequests.get(0).handler().onFailure(time.milliseconds(), new TimeoutException("timeout"));
+ verify(membershipManager).onHeartbeatFailure(true);
+ verify(backgroundEventHandler, never()).add(any());
// Assure the manager will backoff on timeout
time.sleep(DEFAULT_RETRY_BACKOFF_MS - 1);
@@ -340,6 +342,7 @@ public class HeartbeatRequestManagerTest {
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(1, result.unsentRequests.size());
result.unsentRequests.get(0).handler().onFailure(time.milliseconds(), new KafkaException("fatal"));
+ verify(membershipManager).onHeartbeatFailure(false);
verify(membershipManager).transitionToFatal();
verify(backgroundEventHandler).add(any());
}
@@ -494,6 +497,10 @@ public class HeartbeatRequestManagerTest {
break;
}
+ if (error != Errors.NONE) {
+ verify(membershipManager).onHeartbeatFailure(false);
+ }
+
if (!isFatal) {
// Make sure a next heartbeat is sent for all non-fatal errors (to retry or rejoin)
time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
@@ -524,7 +531,7 @@ public class HeartbeatRequestManagerTest {
assertEquals(Collections.emptyList(), data.subscribedTopicNames());
assertEquals(ConsumerTestBuilder.DEFAULT_REMOTE_ASSIGNOR, data.serverAssignor());
assertEquals(Collections.emptyList(), data.topicPartitions());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
// Mock a response from the group coordinator, that supplies the member ID and a new epoch
@@ -538,7 +545,7 @@ public class HeartbeatRequestManagerTest {
assertNull(data.subscribedTopicNames());
assertNull(data.serverAssignor());
assertEquals(data.topicPartitions(), Collections.emptyList());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
// Join the group and subscribe to a topic, but the response has not yet been received
@@ -555,7 +562,7 @@ public class HeartbeatRequestManagerTest {
assertEquals(Collections.singletonList(topic), data.subscribedTopicNames());
assertEquals(ConsumerTestBuilder.DEFAULT_REMOTE_ASSIGNOR, data.serverAssignor());
assertEquals(Collections.emptyList(), data.topicPartitions());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.JOINING, membershipManager.state());
membershipManager.transitionToFenced();
@@ -568,7 +575,7 @@ public class HeartbeatRequestManagerTest {
assertEquals(Collections.singletonList(topic), data.subscribedTopicNames());
assertEquals(ConsumerTestBuilder.DEFAULT_REMOTE_ASSIGNOR, data.serverAssignor());
assertEquals(Collections.emptyList(), data.topicPartitions());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.JOINING, membershipManager.state());
// Mock the response from the group coordinator which returns an assignment
@@ -622,7 +629,7 @@ public class HeartbeatRequestManagerTest {
verify(membershipManager).transitionToSendingLeaveGroup(true);
verify(heartbeatState).reset();
verify(heartbeatRequestState).reset();
- verify(membershipManager).onHeartbeatRequestSent();
+ verify(membershipManager).onHeartbeatRequestGenerated();
when(membershipManager.state()).thenReturn(MemberState.STALE);
when(membershipManager.shouldSkipHeartbeat()).thenReturn(true);
@@ -798,7 +805,7 @@ public class HeartbeatRequestManagerTest {
);
membershipManager.onHeartbeatSuccess(rs1.data());
membershipManager.poll(time.milliseconds());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
index 5f828bcd3dd..7c6d486f81a 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
@@ -44,6 +44,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -166,34 +167,13 @@ public class MembershipManagerImplTest {
createMembershipManagerJoiningGroup();
}
- @Test
- public void testMembershipManagerRegistersForClusterMetadataUpdatesOnFirstJoin() {
- // First join should register to get metadata updates
- MembershipManagerImpl manager = new MembershipManagerImpl(
- GROUP_ID, Optional.empty(), REBALANCE_TIMEOUT, Optional.empty(),
- subscriptionState, commitRequestManager, metadata, LOG_CONTEXT, Optional.empty(),
- backgroundEventHandler, time, rebalanceMetricsManager);
- manager.transitionToJoining();
- clearInvocations(metadata);
-
- // Following joins should not register again.
- receiveEmptyAssignment(manager);
- mockLeaveGroup();
- manager.leaveGroup();
- verify(subscriptionState).unsubscribe();
- assertEquals(MemberState.LEAVING, manager.state());
- manager.onHeartbeatRequestSent();
- assertEquals(MemberState.UNSUBSCRIBED, manager.state());
- manager.transitionToJoining();
- }
-
@Test
public void testReconcilingWhenReceivingAssignmentFoundInMetadata() {
MembershipManager membershipManager = mockJoinAndReceiveAssignment(true);
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
// When the ack is sent the member should go back to STABLE
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
}
@@ -322,7 +302,7 @@ public class MembershipManagerImplTest {
when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
membershipManager.onHeartbeatSuccess(heartbeatResponse.data());
membershipManager.poll(time.milliseconds());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
assertEquals(MEMBER_ID, membershipManager.memberId());
assertEquals(MEMBER_EPOCH, membershipManager.memberEpoch());
@@ -369,6 +349,22 @@ public class MembershipManagerImplTest {
assertTrue(membershipManager.shouldSkipHeartbeat());
}
+ @Test
+ public void testFencingWhenStateIsPrepareLeavingCompletesTheLeaveOperation() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+ mockPrepareLeaving(new TopicPartition("topic1", 0));
+ CompletableFuture<Void> leaveOperation = membershipManager.leaveGroup();
+ assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
+ assertFalse(leaveOperation.isDone());
+ clearInvocations(subscriptionState);
+
+ membershipManager.transitionToFenced();
+
+ testFenceIsNoOp(membershipManager);
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ assertTrue(leaveOperation.isDone(), "Fenced member should complete the ongoing leave operation");
+ }
+
@Test
public void testNewAssignmentIgnoredWhenStateIsPrepareLeaving() {
MembershipManagerImpl membershipManager = createMemberInStableState();
@@ -398,7 +394,7 @@ public class MembershipManagerImplTest {
// Start leaving group.
mockLeaveGroup();
- membershipManager.leaveGroup();
+ CompletableFuture<Void> leaveOperation = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
assertEquals(MemberState.LEAVING, membershipManager.state());
@@ -407,17 +403,21 @@ public class MembershipManagerImplTest {
clearInvocations(subscriptionState);
membershipManager.transitionToFenced();
testFenceIsNoOp(membershipManager);
- assertEquals(MemberState.LEAVING, membershipManager.state());
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state(), "Member should " +
+ "transition from LEAVING to UNSUBSCRIBED when getting fenced (it does not need to " +
+ "send leave request if fenced");
+ assertTrue(leaveOperation.isDone(), "Fenced member should complete the ongoing leave operation");
+ }
- membershipManager.onHeartbeatRequestSent();
+ private void assertTransitionToUnsubscribeOnHBSentAndWaitForResponseToCompleteLeave(MembershipManager membershipManager, CompletableFuture<Void> sendLeave) {
+ assertEquals(MemberState.LEAVING, membershipManager.state());
+ membershipManager.onHeartbeatRequestGenerated();
+ assertFalse(sendLeave.isDone(), "Send leave operation should not complete until a response is received");
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
- // Last heartbeat sent is expected to fail, leading to a call to transitionToFenced. That
- // should be no-op because the member already left.
- clearInvocations(subscriptionState);
- membershipManager.transitionToFenced();
- testFenceIsNoOp(membershipManager);
- assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ membershipManager.onHeartbeatSuccess(createConsumerGroupHeartbeatResponse(new Assignment()).data());
+
+ assertSendLeaveCompleted(membershipManager, sendLeave);
}
@Test
@@ -622,7 +622,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.RECONCILING, membershipManager.state());
membershipManager.poll(time.milliseconds());
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
assertEquals(toTopicIdPartitionMap(assignment1), membershipManager.currentAssignment().partitions);
}
@@ -660,7 +660,7 @@ public class MembershipManagerImplTest {
verifyReconciliationTriggeredAndCompleted(membershipManager,
Collections.singletonList(new TopicIdPartition(topic1, new TopicPartition("topic1", 0)))
);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
clearInvocations(membershipManager);
@@ -700,7 +700,7 @@ public class MembershipManagerImplTest {
String topic1 = "topic1";
MembershipManagerImpl membershipManager =
mockMemberSuccessfullyReceivesAndAcksAssignment(topicId1, topic1, Collections.singletonList(0));
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
clearInvocations(membershipManager, subscriptionState);
when(subscriptionState.assignedPartitions()).thenReturn(Collections.singleton(new TopicPartition(topic1, 0)));
@@ -739,7 +739,7 @@ public class MembershipManagerImplTest {
// After acknowledging the assignment, we should be back to RECONCILING, because we have not
// yet reached the target assignment.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertEquals(topic2Assignment, membershipManager.topicPartitionsAwaitingReconciliation());
@@ -807,7 +807,7 @@ public class MembershipManagerImplTest {
// After acknowledging the assignment, we should be back to RECONCILING, because we have not
// yet reached the target assignment.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
clearInvocations(membershipManager, commitRequestManager);
@@ -835,7 +835,7 @@ public class MembershipManagerImplTest {
MembershipManagerImpl membershipManager =
mockMemberSuccessfullyReceivesAndAcksAssignment(topicId1, topic1, Collections.singletonList(0));
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
when(subscriptionState.assignedPartitions()).thenReturn(getTopicPartitions(Collections.singleton(topicId1Partition0)));
@@ -857,7 +857,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
assertTrue(membershipManager.shouldHeartbeatNow());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
verifyReconciliationNotTriggered(membershipManager);
assertEquals(MemberState.RECONCILING, membershipManager.state());
@@ -886,21 +886,51 @@ public class MembershipManagerImplTest {
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
}
@Test
- public void testIgnoreHeartbeatWhenLeavingGroup() {
+ public void testHeartbeatSuccessfulResponseWhenLeavingGroupCompletesLeave() {
MembershipManager membershipManager = createMemberInStableState();
mockLeaveGroup();
CompletableFuture<Void> leaveResult = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
+ assertEquals(MemberState.LEAVING, membershipManager.state());
+ assertFalse(leaveResult.isDone());
+
+ membershipManager.onHeartbeatRequestGenerated();
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ assertFalse(leaveResult.isDone());
membershipManager.onHeartbeatSuccess(createConsumerGroupHeartbeatResponse(createAssignment(true)).data());
+ assertSendLeaveCompleted(membershipManager, leaveResult);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testHeartbeatFailedResponseWhenLeavingGroupCompletesLeave(boolean retriableResponseError) {
+ MembershipManager membershipManager = createMemberInStableState();
+ mockLeaveGroup();
+ CompletableFuture<Void> leaveResult = membershipManager.leaveGroup();
assertEquals(MemberState.LEAVING, membershipManager.state());
+ assertFalse(leaveResult.isDone());
+
+ membershipManager.onHeartbeatRequestGenerated();
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ assertFalse(leaveResult.isDone());
+
+ membershipManager.onHeartbeatFailure(retriableResponseError);
+ assertSendLeaveCompleted(membershipManager, leaveResult);
+ }
+
+ private void assertSendLeaveCompleted(MembershipManager membershipManager,
+ CompletableFuture<Void> sendLeave) {
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state(), "Member should " +
+ "remain UNSUBSCRIBED after receiving the response to the HB to leave");
assertEquals(-1, membershipManager.memberEpoch());
assertEquals(MEMBER_ID, membershipManager.memberId());
assertTrue(membershipManager.currentAssignment().isNone());
- assertFalse(leaveResult.isDone(), "Leave group result should not complete until the " +
- "heartbeat request to leave is sent out.");
+ assertTrue(sendLeave.isDone(), "Leave group result should complete when the response to" +
+ " the heartbeat request to leave is received.");
+ assertFalse(sendLeave.isCompletedExceptionally());
}
@ParameterizedTest
@@ -987,9 +1017,7 @@ public class MembershipManagerImplTest {
assertFalse(leaveResult2.isDone());
// Complete first leave group operation. Should also complete the second leave group.
- membershipManager.onHeartbeatRequestSent();
- assertTrue(leaveResult1.isDone());
- assertFalse(leaveResult1.isCompletedExceptionally());
+ assertTransitionToUnsubscribeOnHBSentAndWaitForResponseToCompleteLeave(membershipManager, leaveResult1);
assertTrue(leaveResult2.isDone());
assertFalse(leaveResult2.isCompletedExceptionally());
@@ -1006,10 +1034,7 @@ public class MembershipManagerImplTest {
CompletableFuture<Void> leaveResult1 = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
assertEquals(MemberState.LEAVING, membershipManager.state());
- membershipManager.onHeartbeatRequestSent();
- assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
- assertTrue(leaveResult1.isDone());
- assertFalse(leaveResult1.isCompletedExceptionally());
+ assertTransitionToUnsubscribeOnHBSentAndWaitForResponseToCompleteLeave(membershipManager, leaveResult1);
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
clearInvocations(subscriptionState);
@@ -1033,9 +1058,10 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.FENCED, membershipManager.state());
mockLeaveGroup();
- membershipManager.leaveGroup();
+ CompletableFuture<Void> leaveOperation = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ assertTrue(leaveOperation.isDone());
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
completeCallback(callbackEvent, membershipManager);
@@ -1090,9 +1116,10 @@ public class MembershipManagerImplTest {
// Start leaving group.
mockLeaveGroup();
- membershipManager.leaveGroup();
+ CompletableFuture<Void> leaveOperation = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
assertEquals(MemberState.LEAVING, membershipManager.state());
+ assertFalse(leaveOperation.isDone());
// Get fatal failure while waiting to send the heartbeat to leave. Member should
// transition to FATAL, so the last heartbeat for leaving won't be sent because the member
@@ -1100,7 +1127,8 @@ public class MembershipManagerImplTest {
testStateUpdateOnFatalFailure(membershipManager);
assertEquals(MemberState.FATAL, membershipManager.state());
- membershipManager.onHeartbeatRequestSent();
+ assertTrue(leaveOperation.isDone(), "Member should complete the ongoing leave operation when transitioning to FATAL state");
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.FATAL, membershipManager.state());
}
@@ -1110,22 +1138,27 @@ public class MembershipManagerImplTest {
// Start leaving group.
mockLeaveGroup();
- membershipManager.leaveGroup();
+ CompletableFuture<Void> sendLeave = membershipManager.leaveGroup();
verify(subscriptionState).unsubscribe();
assertEquals(MemberState.LEAVING, membershipManager.state());
// Last heartbeat sent.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
// Fatal error received in response for the last heartbeat. Member should remain in FATAL
// state but no callbacks should be triggered because the member already left the group.
MockRebalanceListener rebalanceListener = new MockRebalanceListener();
when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(rebalanceListener));
+ membershipManager.onHeartbeatFailure(false);
membershipManager.transitionToFatal();
assertEquals(0, rebalanceListener.lostCount);
assertEquals(MemberState.FATAL, membershipManager.state());
+
+ assertTrue(sendLeave.isDone());
+ assertFalse(sendLeave.isCompletedExceptionally(), "Send leave should complete when a " +
+ "response to the leave group is received, even if it contains errors.");
}
@Test
@@ -1154,7 +1187,7 @@ public class MembershipManagerImplTest {
// When the ack is sent nothing should change. Member still has nothing to reconcile,
// only topics waiting for metadata.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertFalse(membershipManager.topicsAwaitingReconciliation().isEmpty());
@@ -1176,7 +1209,7 @@ public class MembershipManagerImplTest {
// When ack for the reconciled assignment is sent, member should go back to STABLE
// because the first assignment that was not resolved should have been discarded
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
assertTrue(membershipManager.topicsAwaitingReconciliation().isEmpty());
}
@@ -1195,7 +1228,7 @@ public class MembershipManagerImplTest {
// When the ack is sent nothing should change. Member still has nothing to reconcile,
// only topics waiting for metadata.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertFalse(membershipManager.topicsAwaitingReconciliation().isEmpty());
@@ -1214,7 +1247,7 @@ public class MembershipManagerImplTest {
verifyReconciliationTriggeredAndCompleted(membershipManager, Collections.emptyList());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
}
@@ -1235,7 +1268,7 @@ public class MembershipManagerImplTest {
verifyReconciliationNotTriggered(membershipManager);
membershipManager.poll(time.milliseconds());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertFalse(membershipManager.topicsAwaitingReconciliation().isEmpty());
@@ -1302,7 +1335,7 @@ public class MembershipManagerImplTest {
// When the ack is sent the member should go back to RECONCILING because it still has
// unresolved assignment to be reconciled.
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Target assignment received again with the same unresolved topic. Client should keep it
@@ -1370,7 +1403,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
clearInvocations(subscriptionState, membershipManager);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
// Receive same assignment again - should not trigger reconciliation
@@ -1490,7 +1523,7 @@ public class MembershipManagerImplTest {
.setTopicId(topicId)
.setPartitions(Arrays.asList(0, 1))));
MembershipManagerImpl membershipManager = mockJoinAndReceiveAssignment(true, targetAssignment);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Should not trigger reconciliation, and request a metadata update.
@@ -1520,7 +1553,7 @@ public class MembershipManagerImplTest {
.setTopicId(topicId)
.setPartitions(Arrays.asList(0, 1))));
MembershipManagerImpl membershipManager = mockJoinAndReceiveAssignment(true, targetAssignment);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Should not trigger reconciliation, and request a metadata update.
@@ -1632,7 +1665,7 @@ public class MembershipManagerImplTest {
assertFalse(membershipManager.reconciliationInProgress());
// Step 4: Send ack and make sure we're done and our listener was called appropriately
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
assertEquals(topicIdPartitionsMap(topicId, 0, 1), membershipManager.currentAssignment().partitions);
@@ -1669,7 +1702,7 @@ public class MembershipManagerImplTest {
assertFalse(membershipManager.reconciliationInProgress());
// Step 8: Send ack and make sure we're done and our listener was called appropriately
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
assertFalse(membershipManager.reconciliationInProgress());
@@ -1724,7 +1757,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Step 4: Send ack and make sure we're done and our listener was called appropriately
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertEquals(1, listener.revokedCount());
@@ -1788,7 +1821,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.RECONCILING, membershipManager.state());
// Step 5: Send ack and make sure we're done and our listener was called appropriately
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.RECONCILING, membershipManager.state());
assertEquals(1, listener.revokedCount());
@@ -1868,7 +1901,7 @@ public class MembershipManagerImplTest {
private void assertLeaveGroupDueToExpiredPollAndTransitionToStale(MembershipManagerImpl membershipManager) {
assertDoesNotThrow(() -> membershipManager.transitionToSendingLeaveGroup(true));
assertEquals(LEAVE_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertStaleMemberLeavesGroupAndClearsAssignment(membershipManager);
}
@@ -1911,7 +1944,7 @@ public class MembershipManagerImplTest {
MembershipManagerImpl membershipManager = createMemberInStableState();
doNothing().when(subscriptionState).assignFromSubscribed(any());
membershipManager.transitionToSendingLeaveGroup(true);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STALE, membershipManager.state());
assertTrue(membershipManager.shouldSkipHeartbeat(), "Stale member should not send heartbeats");
// Check that a transition to joining is allowed, which is what is expected to happen
@@ -1942,7 +1975,7 @@ public class MembershipManagerImplTest {
when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
membershipManager.transitionToSendingLeaveGroup(true);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STALE, membershipManager.state());
assertFalse(backgroundEventQueue.isEmpty());
@@ -1975,7 +2008,7 @@ public class MembershipManagerImplTest {
doNothing().when(subscriptionState).assignFromSubscribed(any());
when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
membershipManager.transitionToSendingLeaveGroup(true);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
return membershipManager;
}
private void mockPartitionOwnedAndNewPartitionAdded(String topicName,
@@ -2032,7 +2065,7 @@ public class MembershipManagerImplTest {
assertTrue(membershipManager.currentAssignment().isNone());
// Step 4: Receive ack and make sure we're done and our listener was called appropriately
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.JOINING, membershipManager.state());
assertEquals(0, listener.revokedCount());
@@ -2151,7 +2184,7 @@ public class MembershipManagerImplTest {
verifyReconciliationNotTriggered(membershipManager);
membershipManager.poll(time.milliseconds());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
}
@@ -2189,7 +2222,7 @@ public class MembershipManagerImplTest {
@Test
public void testMetricsWhenHeartbeatFailed() {
MembershipManagerImpl membershipManager = createMemberInStableState();
- membershipManager.onHeartbeatFailure();
+ membershipManager.onHeartbeatFailure(false);
// Not expecting rebalance failures with only the empty assignment being reconciled.
assertEquals(1.0d, getMetricValue(metrics, rebalanceMetricsManager.rebalanceTotal));
@@ -2256,7 +2289,7 @@ public class MembershipManagerImplTest {
listener.reset();
// ack
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
// revoke all
when(subscriptionState.assignedPartitions()).thenReturn(topicPartitions(topicName, 0, 1));
@@ -2280,7 +2313,7 @@ public class MembershipManagerImplTest {
topicPartitions(topicName, 2),
true
);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
long secondRebalanceMs = listener.sleepMs;
long total = firstRebalanaceTimesMs + secondRebalanceMs;
@@ -2311,7 +2344,7 @@ public class MembershipManagerImplTest {
time.sleep(2300);
assertTrue(rebalanceMetricsManager.rebalanceStarted());
- membershipManager.onHeartbeatFailure();
+ membershipManager.onHeartbeatFailure(false);
assertEquals((double) 0, getMetricValue(metrics, rebalanceMetricsManager.rebalanceLatencyTotal));
assertEquals(0d, getMetricValue(metrics, rebalanceMetricsManager.rebalanceTotal));
@@ -2459,7 +2492,7 @@ public class MembershipManagerImplTest {
}
private void mockAckSent(MembershipManagerImpl membershipManager) {
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
}
private void mockTopicNameInMetadataCache(Map<Uuid, String> topicNames, boolean isPresent) {
@@ -2582,7 +2615,7 @@ public class MembershipManagerImplTest {
assertEquals(MemberState.RECONCILING, membershipManager.state());
membershipManager.poll(time.milliseconds());
assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertEquals(MemberState.STABLE, membershipManager.state());
clearInvocations(subscriptionState, membershipManager, commitRequestManager);
@@ -2689,11 +2722,7 @@ public class MembershipManagerImplTest {
assertFalse(leaveResult.isDone(), "Leave group result should not complete until the " +
"heartbeat request to leave is sent out.");
- membershipManager.onHeartbeatRequestSent();
-
- assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
- assertTrue(leaveResult.isDone());
- assertFalse(leaveResult.isCompletedExceptionally());
+ assertTransitionToUnsubscribeOnHBSentAndWaitForResponseToCompleteLeave(membershipManager, leaveResult);
assertEquals(MEMBER_ID, membershipManager.memberId());
assertEquals(-1, membershipManager.memberEpoch());
assertTrue(membershipManager.currentAssignment().isNone());
@@ -2710,14 +2739,7 @@ public class MembershipManagerImplTest {
ConsumerRebalanceListenerInvoker invoker) {
String topicName = "topic1";
TopicPartition ownedPartition = new TopicPartition(topicName, 0);
-
- // Start leaving group, blocked waiting for callback to complete.
- CounterConsumerRebalanceListener listener = new CounterConsumerRebalanceListener();
- when(subscriptionState.assignedPartitions()).thenReturn(Collections.singleton(ownedPartition));
- when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
- when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
- doNothing().when(subscriptionState).markPendingRevocation(anySet());
- when(commitRequestManager.autoCommitEnabled()).thenReturn(false);
+ mockPrepareLeaving(ownedPartition);
membershipManager.leaveGroup();
return performCallback(
membershipManager,
@@ -2728,6 +2750,16 @@ public class MembershipManagerImplTest {
);
}
+ private void mockPrepareLeaving(TopicPartition ownedPartition) {
+ // Start leaving group, blocked waiting for callback to complete.
+ CounterConsumerRebalanceListener listener = new CounterConsumerRebalanceListener();
+ when(subscriptionState.assignedPartitions()).thenReturn(Collections.singleton(ownedPartition));
+ when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
+ when(subscriptionState.rebalanceListener()).thenReturn(Optional.of(listener));
+ doNothing().when(subscriptionState).markPendingRevocation(anySet());
+ when(commitRequestManager.autoCommitEnabled()).thenReturn(false);
+ }
+
private ConsumerRebalanceListenerCallbackCompletedEvent mockFencedMemberStuckOnUserCallback(
MembershipManagerImpl membershipManager,
ConsumerRebalanceListenerInvoker invoker) {
@@ -2754,6 +2786,7 @@ public class MembershipManagerImplTest {
String memberId = membershipManager.memberId();
int lastEpoch = membershipManager.memberEpoch();
when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(true);
+ membershipManager.onHeartbeatFailure(false);
membershipManager.transitionToFatal();
assertEquals(MemberState.FATAL, membershipManager.state());
// Should keep its last member id and epoch.
@@ -2814,10 +2847,10 @@ public class MembershipManagerImplTest {
private MembershipManagerImpl memberJoinWithAssignment() {
Uuid topicId = Uuid.randomUuid();
MembershipManagerImpl membershipManager = mockJoinAndReceiveAssignment(true);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
when(metadata.topicNames()).thenReturn(Collections.singletonMap(topicId, "topic"));
receiveAssignment(topicId, Collections.singletonList(0), membershipManager);
- membershipManager.onHeartbeatRequestSent();
+ membershipManager.onHeartbeatRequestGenerated();
assertFalse(membershipManager.currentAssignment().isNone());
return membershipManager;
}