You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2024/12/05 11:23:54 UTC
(kafka) branch kip1071 updated: Use LeaveOnClose event with Streams membership manager (#18054)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch kip1071
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/kip1071 by this push:
new 5cc622a650f Use LeaveOnClose event with Streams membership manager (#18054)
5cc622a650f is described below
commit 5cc622a650fb77fa92dab699a86b10cc5811cf0d
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Thu Dec 5 12:23:48 2024 +0100
Use LeaveOnClose event with Streams membership manager (#18054)
The Streams membership manager did not use the LeaveOnClose event
to faster leaving the group on closing. With the event the Streams
membership manager leaves the group without invoking any callback
for releasing the active tasks.
With this commit the Streams membership member uses the LeaveOnClose
event.
---
checkstyle/suppressions.xml | 2 +-
.../StreamsGroupHeartbeatRequestManager.java | 2 +-
.../internals/StreamsMembershipManager.java | 54 +++++++++++++----
.../events/ApplicationEventProcessor.java | 15 +++--
.../internals/StreamsMembershipManagerTest.java | 67 ++++++++++++++++++----
.../events/ApplicationEventProcessorTest.java | 66 +++++++++++++++++++++
6 files changed, 178 insertions(+), 28 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index c17c9882f00..113555b93e0 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -119,7 +119,7 @@
files="(Sender|Fetcher|FetchRequestManager|OffsetFetcher|KafkaConsumer|Metrics|RequestResponse|TransactionManager|KafkaAdminClient|Message|KafkaProducer)Test.java"/>
<suppress checks="ClassFanOutComplexity"
- files="(ConsumerCoordinator|KafkaConsumer|RequestResponse|Fetcher|FetchRequestManager|KafkaAdminClient|Message|KafkaProducer|NetworkClient)Test.java"/>
+ files="(ConsumerCoordinator|KafkaConsumer|RequestResponse|Fetcher|FetchRequestManager|KafkaAdminClient|Message|KafkaProducer|NetworkClient|ApplicationEventProcessor)Test.java"/>
<suppress checks="ClassFanOutComplexity"
files="MockAdminClient.java"/>
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
index c99fea8a01a..3d235222f36 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
@@ -103,7 +103,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
@Override
public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
- if (!coordinatorRequestManager.coordinator().isPresent() || membershipManager.shouldSkipHeartbeat()) {
+ if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
membershipManager.transitionToUnsubscribeIfLeaving();
return NetworkClientDelegate.PollResult.EMPTY;
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java
index a3f55c92d3a..72378c0583f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java
@@ -243,7 +243,7 @@ public class StreamsMembershipManager implements RequestManager {
}
resetEpoch();
transitionTo(MemberState.JOINING);
- clearPendingTaskAssignment();
+ clearCurrentTaskAssignment();
}
private void transitionToSendingLeaveGroup(boolean dueToExpiredPollTimer) {
@@ -271,7 +271,7 @@ public class StreamsMembershipManager implements RequestManager {
private void finalizeLeaving() {
updateMemberEpoch(StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH);
- clearPendingTaskAssignment();
+ clearCurrentTaskAssignment();
}
private void transitionToStale() {
@@ -373,7 +373,7 @@ public class StreamsMembershipManager implements RequestManager {
}
}
- private void clearPendingTaskAssignment() {
+ private void clearCurrentTaskAssignment() {
currentAssignment = LocalAssignment.NONE;
}
@@ -582,6 +582,27 @@ public class StreamsMembershipManager implements RequestManager {
.collect(Collectors.toMap(StreamsGroupHeartbeatResponseData.TaskIds::subtopologyId, taskId -> new TreeSet<>(taskId.partitions())));
}
+ /**
+ * Leaves the group when the member closes.
+ *
+ * <p>
+ * This method does the following:
+ * <ol>
+ * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
+ * <li>Skips the invocation of the revocation callback or lost callback.</li>
+ * <li>Clears the current and target assignment, unsubscribes from all topics and
+ * transitions the member state to {@link MemberState#LEAVING}.</li>
+ * </ol>
+ * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
+ * to send a leave group heartbeat.
+ * </p>
+ *
+ * @return future that will complete when the heartbeat to leave the group has been sent out.
+ */
+ public CompletableFuture<Void> leaveGroupOnClose() {
+ return leaveGroup(true);
+ }
+
/**
* Leaves the group.
*
@@ -589,8 +610,8 @@ public class StreamsMembershipManager implements RequestManager {
* This method does the following:
* <ol>
* <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Requests the invocation of the revocation callback.</li>
- * <li>Once the revocation callback completes, it clears the current and target assignment, unsubscribes from
+ * <li>Requests the invocation of the revocation callback or lost callback.</li>
+ * <li>Once the callback completes, it clears the current and target assignment, unsubscribes from
* all topics and transitions the member state to {@link MemberState#LEAVING}.</li>
* </ol>
* States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
@@ -601,6 +622,10 @@ public class StreamsMembershipManager implements RequestManager {
* to leave the group has been sent out.
*/
public CompletableFuture<Void> leaveGroup() {
+ return leaveGroup(false);
+ }
+
+ private CompletableFuture<Void> leaveGroup(final boolean isOnClose) {
if (isNotInGroup()) {
if (state == MemberState.FENCED) {
clearTaskAndPartitionAssignment();
@@ -615,16 +640,21 @@ public class StreamsMembershipManager implements RequestManager {
return leaveGroupInProgress.get();
}
+ transitionTo(MemberState.PREPARE_LEAVING);
CompletableFuture<Void> onGroupLeft = new CompletableFuture<>();
leaveGroupInProgress = Optional.of(onGroupLeft);
- CompletableFuture<Void> onAllTasksRevokedCallbackExecuted = prepareLeaving();
- onAllTasksRevokedCallbackExecuted.whenComplete((__, callbackError) -> leaving(callbackError));
+ if (!isOnClose) {
+ CompletableFuture<Void> onAllActiveTasksReleasedCallbackExecuted = releaseActiveTasks();
+ onAllActiveTasksReleasedCallbackExecuted
+ .whenComplete((__, callbackError) -> leavingAfterReleasingActiveTasks(callbackError));
+ } else {
+ leaving();
+ }
return onGroupLeft;
}
- private CompletableFuture<Void> prepareLeaving() {
- transitionTo(MemberState.PREPARE_LEAVING);
+ private CompletableFuture<Void> releaseActiveTasks() {
if (memberEpoch > 0) {
return revokeActiveTasks(toTaskIdSet(currentAssignment.activeTasks));
} else {
@@ -632,7 +662,7 @@ public class StreamsMembershipManager implements RequestManager {
}
}
- private void leaving(Throwable callbackError) {
+ private void leavingAfterReleasingActiveTasks(Throwable callbackError) {
if (callbackError != null) {
log.error("Member {} callback to revoke task assignment failed. It will proceed " +
"to clear its assignment and send a leave group heartbeat",
@@ -642,6 +672,10 @@ public class StreamsMembershipManager implements RequestManager {
"to clear its assignment and send a leave group heartbeat",
memberIdInfoForLog());
}
+ leaving();
+ }
+
+ private void leaving() {
clearTaskAndPartitionAssignment();
subscriptionState.unsubscribe();
transitionToSendingLeaveGroup(false);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
index e2cf2526d38..6c8333a5f04 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
@@ -455,12 +455,15 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
}
private void process(final LeaveGroupOnCloseEvent event) {
- if (requestManagers.consumerMembershipManager.isEmpty())
- return;
-
- log.debug("Signal the ConsumerMembershipManager to leave the consumer group since the consumer is closing");
- CompletableFuture<Void> future = requestManagers.consumerMembershipManager.get().leaveGroupOnClose();
- future.whenComplete(complete(event.future()));
+ if (requestManagers.consumerMembershipManager.isPresent()) {
+ CompletableFuture<Void> future = requestManagers.consumerMembershipManager.get().leaveGroupOnClose();
+ future.whenComplete(complete(event.future()));
+ log.debug("Signal the ConsumerMembershipManager to leave the consumer group since the consumer is closing");
+ } else if (requestManagers.streamsMembershipManager.isPresent()) {
+ CompletableFuture<Void> future = requestManagers.streamsMembershipManager.get().leaveGroupOnClose();
+ future.whenComplete(complete(event.future()));
+ log.debug("Signal the StreamsMembershipManager to leave the Streams group since the member is closing");
+ }
}
/**
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java
index 1ee0975a8cb..ea0b3fa9d9e 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java
@@ -17,7 +17,6 @@
package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent;
-import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
import org.apache.kafka.clients.consumer.internals.events.StreamsOnAllTasksLostCallbackCompletedEvent;
import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksAssignedCallbackCompletedEvent;
import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksAssignedCallbackNeededEvent;
@@ -35,8 +34,6 @@ import org.apache.kafka.common.utils.Time;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -49,6 +46,7 @@ import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.function.Supplier;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -89,11 +87,7 @@ public class StreamsMembershipManagerTest {
@Mock
private StreamsAssignmentInterface streamsAssignmentInterface;
- @Captor
- private ArgumentCaptor<StreamsOnTasksAssignedCallbackNeededEvent> onAssignmentCallbackNeededEventCaptor;
-
private Queue<BackgroundEvent> backgroundEventQueue = new LinkedList<>();
- private BackgroundEventHandler backgroundEventHandler = new BackgroundEventHandler(backgroundEventQueue);
@BeforeEach
public void setup() {
@@ -355,7 +349,16 @@ public class StreamsMembershipManagerTest {
@Test
public void testLeaveGroupWhenNotInGroup() {
- final CompletableFuture<Void> future = membershipManager.leaveGroup();
+ testLeaveGroupWhenNotInGroup(membershipManager::leaveGroup);
+ }
+
+ @Test
+ public void testLeaveGroupOnCloseWhenNotInGroup() {
+ testLeaveGroupWhenNotInGroup(membershipManager::leaveGroupOnClose);
+ }
+
+ private void testLeaveGroupWhenNotInGroup(final Supplier<CompletableFuture<Void>> leaveGroup) {
+ final CompletableFuture<Void> future = leaveGroup.get();
assertFalse(membershipManager.isLeavingGroup());
assertTrue(future.isDone());
@@ -367,12 +370,21 @@ public class StreamsMembershipManagerTest {
@Test
public void testLeaveGroupWhenNotInGroupAndFenced() {
+ testLeaveGroupOnCloseWhenNotInGroupAndFenced(membershipManager::leaveGroup);
+ }
+
+ @Test
+ public void testLeaveGroupOnCloseWhenNotInGroupAndFenced() {
+ testLeaveGroupOnCloseWhenNotInGroupAndFenced(membershipManager::leaveGroupOnClose);
+ }
+
+ private void testLeaveGroupOnCloseWhenNotInGroupAndFenced(final Supplier<CompletableFuture<Void>> leaveGroup) {
final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
.thenReturn(onAllTasksLostCallbackExecuted);
joining();
fenced();
- final CompletableFuture<Void> future = membershipManager.leaveGroup();
+ final CompletableFuture<Void> future = leaveGroup.get();
assertFalse(membershipManager.isLeavingGroup());
assertTrue(future.isDone());
@@ -405,14 +417,49 @@ public class StreamsMembershipManagerTest {
verifyInStatePrepareLeaving(membershipManager);
final CompletableFuture<Void> onGroupLeftBeforeRevocationCallback = membershipManager.leaveGroup();
assertEquals(onGroupLeft, onGroupLeftBeforeRevocationCallback);
+ final CompletableFuture<Void> onGroupLeftOnCloseBeforeRevocationCallback = membershipManager.leaveGroupOnClose();
+ assertEquals(onGroupLeft, onGroupLeftOnCloseBeforeRevocationCallback);
onTasksRevokedCallbackExecuted.complete(null);
verify(subscriptionState).unsubscribe();
assertFalse(onGroupLeft.isDone());
verifyInStateLeaving(membershipManager);
final CompletableFuture<Void> onGroupLeftAfterRevocationCallback = membershipManager.leaveGroup();
assertEquals(onGroupLeft, onGroupLeftAfterRevocationCallback);
- membershipManager.transitionToUnsubscribeIfLeaving();
+ membershipManager.onHeartbeatRequestGenerated();
+ verifyInStateUnsubscribed(membershipManager);
+ membershipManager.onHeartbeatSuccess(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
+ assertTrue(onGroupLeft.isDone());
+ assertFalse(onGroupLeft.isCompletedExceptionally());
+ }
+
+ @Test
+ public void testLeaveGroupOnCloseWhenInGroupWithAssignment() {
+ setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final Set<StreamsAssignmentInterface.TaskId> activeTasks =
+ Set.of(new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ joining();
+ reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
+
+ final CompletableFuture<Void> onGroupLeft = membershipManager.leaveGroupOnClose();
+
+ assertFalse(onGroupLeft.isDone());
+ verifyInStateLeaving(membershipManager);
+ verify(subscriptionState).unsubscribe();
+ verify(streamsAssignmentInterface, never()).requestOnTasksRevokedCallbackInvocation(any());
+ final CompletableFuture<Void> onGroupLeftBeforeHeartbeatRequestGenerated = membershipManager.leaveGroup();
+ assertEquals(onGroupLeft, onGroupLeftBeforeHeartbeatRequestGenerated);
+ final CompletableFuture<Void> onGroupLeftOnCloseBeforeHeartbeatRequestGenerated = membershipManager.leaveGroupOnClose();
+ assertEquals(onGroupLeft, onGroupLeftOnCloseBeforeHeartbeatRequestGenerated);
+ assertFalse(onGroupLeft.isDone());
+ membershipManager.onHeartbeatRequestGenerated();
verifyInStateUnsubscribed(membershipManager);
+ membershipManager.onHeartbeatSuccess(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
+ assertTrue(onGroupLeft.isDone());
+ assertFalse(onGroupLeft.isCompletedExceptionally());
}
@Test
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index 511e602b9bf..7e808935951 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -25,6 +25,7 @@ import org.apache.kafka.clients.consumer.internals.CommitRequestManager;
import org.apache.kafka.clients.consumer.internals.ConsumerHeartbeatRequestManager;
import org.apache.kafka.clients.consumer.internals.ConsumerMembershipManager;
import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
+import org.apache.kafka.clients.consumer.internals.ConsumerUtils;
import org.apache.kafka.clients.consumer.internals.CoordinatorRequestManager;
import org.apache.kafka.clients.consumer.internals.FetchRequestManager;
import org.apache.kafka.clients.consumer.internals.MockRebalanceListener;
@@ -63,8 +64,10 @@ import static org.apache.kafka.clients.consumer.internals.events.CompletableEven
import static org.apache.kafka.test.TestUtils.assertFutureThrows;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
@@ -586,6 +589,69 @@ public class ApplicationEventProcessorTest {
verify(streamsMembershipManager).onAllTasksLostCallbackCompleted(event);
}
+
+ @Test
+ public void testLeaveOnCloseWithStreamsMembershipManager() {
+ LeaveGroupOnCloseEvent event = new LeaveGroupOnCloseEvent(ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ when(streamsMembershipManager.leaveGroupOnClose()).thenReturn(future);
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verifyLeaveOnClose(event, future);
+ }
+
+ @Test
+ public void testLeaveOnCloseCompletesExceptionallyWithStreamsMembershipManager() {
+ LeaveGroupOnCloseEvent event = new LeaveGroupOnCloseEvent(ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ when(streamsMembershipManager.leaveGroupOnClose()).thenReturn(future);
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verifyLeaveOnCloseCompletesExceptionally(event, future);
+ }
+
+ @Test
+ public void testLeaveOnCloseWithConsumerMembershipManager() {
+ LeaveGroupOnCloseEvent event = new LeaveGroupOnCloseEvent(ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ when(membershipManager.leaveGroupOnClose()).thenReturn(future);
+
+ setupProcessor(true);
+ processor.process(event);
+ verifyLeaveOnClose(event, future);
+ }
+
+ @Test
+ public void testLeaveOnCloseCompletesExceptionallyWithConsumerMembershipManager() {
+ LeaveGroupOnCloseEvent event = new LeaveGroupOnCloseEvent(ConsumerUtils.DEFAULT_CLOSE_TIMEOUT_MS);
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ when(membershipManager.leaveGroupOnClose()).thenReturn(future);
+
+ setupProcessor(true);
+ processor.process(event);
+ verifyLeaveOnCloseCompletesExceptionally(event, future);
+ }
+
+ private void verifyLeaveOnClose(LeaveGroupOnCloseEvent event, CompletableFuture<Void> future) {
+ assertFalse(event.future().isDone());
+ future.complete(null);
+ assertTrue(event.future().isDone());
+ assertFalse(event.future().isCompletedExceptionally());
+ }
+
+ private void verifyLeaveOnCloseCompletesExceptionally(LeaveGroupOnCloseEvent event, CompletableFuture<Void> future) {
+ assertFalse(event.future().isDone());
+ RuntimeException exception = new RuntimeException("Nobody expects the Spanish Inquisition!");
+ future.completeExceptionally(exception);
+ assertTrue(event.future().isDone());
+ assertTrue(event.future().isCompletedExceptionally());
+ ExecutionException thrown = assertThrows(ExecutionException.class, () -> event.future().get());
+ assertInstanceOf(RuntimeException.class, thrown.getCause());
+ assertEquals(exception.getMessage(), thrown.getCause().getMessage());
+ }
+
private List<NetworkClientDelegate.UnsentRequest> mockCommitResults() {
return Collections.singletonList(mock(NetworkClientDelegate.UnsentRequest.class));
}