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/11/28 16:59:19 UTC
(kafka) 40/42: KIP1071 trunk rebase 11_25 add streams membership manager stream thread integration (#17968)
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
commit b3eeed7a093e252b4057804dd1e4f25ed516a44d
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Thu Nov 28 07:39:16 2024 -0500
KIP1071 trunk rebase 11_25 add streams membership manager stream thread integration (#17968)
Rebase on trunk
---
.../consumer/internals/AsyncKafkaConsumer.java | 11 +-
.../consumer/internals/CommitRequestManager.java | 2 +-
.../internals/ConsumerDelegateCreator.java | 3 +-
.../consumer/internals/RequestManagers.java | 70 ++-
.../internals/StreamsAssignmentInterface.java | 179 +++++++-
.../StreamsGroupHeartbeatRequestManager.java | 103 +----
.../internals/StreamsMembershipManager.java | 238 +++++++---
.../internals/events/ApplicationEvent.java | 4 +-
.../events/ApplicationEventProcessor.java | 76 +++-
.../consumer/internals/events/BackgroundEvent.java | 3 +-
...reamsOnAllTasksLostCallbackCompletedEvent.java} | 8 +-
... StreamsOnAllTasksLostCallbackNeededEvent.java} | 21 +-
...eamsOnTasksAssignedCallbackCompletedEvent.java} | 8 +-
...StreamsOnTasksAssignedCallbackNeededEvent.java} | 6 +-
...reamsOnTasksRevokedCallbackCompletedEvent.java} | 8 +-
... StreamsOnTasksRevokedCallbackNeededEvent.java} | 18 +-
.../StreamsGroupHeartbeatRequestManagerTest.java | 61 ++-
.../internals/StreamsMembershipManagerTest.java | 501 ++++++++++++---------
.../events/ApplicationEventProcessorTest.java | 85 ++++
.../coordinator/group/GroupCoordinatorShard.java | 21 +-
.../streams/CoordinatorStreamsRecordHelpers.java | 28 +-
.../message/ConsumerGroupRegularExpressionKey.json | 4 +-
.../StreamsGroupCurrentMemberAssignmentKey.json | 6 +-
.../message/StreamsGroupMemberMetadataKey.json | 6 +-
.../common/message/StreamsGroupMetadataKey.json | 4 +-
.../message/StreamsGroupPartitionMetadataKey.json | 4 +-
.../StreamsGroupTargetAssignmentMemberKey.json | 6 +-
.../StreamsGroupTargetAssignmentMetadataKey.json | 4 +-
.../common/message/StreamsGroupTopologyKey.json | 4 +-
.../CoordinatorStreamsRecordHelpersTest.java | 2 +-
.../streams/processor/internals/StreamThread.java | 70 ++-
.../internals/StreamsRebalanceListener.java | 2 -
32 files changed, 1014 insertions(+), 552 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 459d26c6010..b74dcce64fe 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -387,7 +387,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
applicationEventProcessorSupplier,
networkClientDelegateSupplier,
requestManagersSupplier);
-
+ streamsAssignmentInterface.ifPresent(
+ sai -> sai.setApplicationEventHandler(applicationEventHandler)
+ );
this.rebalanceListenerInvoker = new ConsumerRebalanceListenerInvoker(
logContext,
subscriptions,
@@ -477,8 +479,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
Deserializer<V> valueDeserializer,
KafkaClient client,
SubscriptionState subscriptions,
- ConsumerMetadata metadata,
- Optional<StreamsAssignmentInterface> streamsInstanceMetadata) {
+ ConsumerMetadata metadata) {
this.log = logContext.logger(getClass());
this.subscriptions = subscriptions;
this.clientId = config.getString(ConsumerConfig.CLIENT_ID_CONFIG);
@@ -546,8 +547,8 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
clientTelemetryReporter,
metrics,
offsetCommitCallbackInvoker,
- memberStateListener,
- streamsInstanceMetadata
+ this::updateGroupMetadata,
+ Optional.empty()
);
Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(
logContext,
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 41522889768..c6e08ad377c 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
@@ -96,7 +96,7 @@ public class CommitRequestManager implements RequestManager, MemberStateListener
private Optional<Integer> lastEpochSentOnCommit;
/**
- * The member ID and latest member epoch received via the {@link MemberStateListener#onMemberEpochUpdated(Optional, String)},
+ * The member ID and latest member epoch received via the {@link MemberStateListener#onMemberEpochUpdated(Optional, Optional)},
* to be included in the OffsetFetch and OffsetCommit requests. This will have
* the latest memberEpoch received from the broker.
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
index 3f5926080fc..d4ded4377b2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
@@ -92,8 +92,7 @@ public class ConsumerDelegateCreator {
valueDeserializer,
client,
subscriptions,
- metadata,
- Optional.empty()
+ metadata
);
else
return new ClassicKafkaConsumer<>(
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index 4058bd72725..919615774cc 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -54,6 +54,7 @@ public class RequestManagers implements Closeable {
public final Optional<ShareHeartbeatRequestManager> shareHeartbeatRequestManager;
public final Optional<ConsumerMembershipManager> consumerMembershipManager;
public final Optional<ShareMembershipManager> shareMembershipManager;
+ public final Optional<StreamsMembershipManager> streamsMembershipManager;
public final OffsetsRequestManager offsetsRequestManager;
public final TopicMetadataRequestManager topicMetadataRequestManager;
public final FetchRequestManager fetchRequestManager;
@@ -70,7 +71,8 @@ public class RequestManagers implements Closeable {
Optional<CommitRequestManager> commitRequestManager,
Optional<ConsumerHeartbeatRequestManager> heartbeatRequestManager,
Optional<ConsumerMembershipManager> membershipManager,
- Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager) {
+ Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager,
+ Optional<StreamsMembershipManager> streamsMembershipManager) {
this.log = logContext.logger(RequestManagers.class);
this.offsetsRequestManager = requireNonNull(offsetsRequestManager, "OffsetsRequestManager cannot be null");
this.coordinatorRequestManager = coordinatorRequestManager;
@@ -83,6 +85,7 @@ public class RequestManagers implements Closeable {
this.consumerMembershipManager = membershipManager;
this.shareMembershipManager = Optional.empty();
this.streamsGroupHeartbeatRequestManager = streamsGroupHeartbeatRequestManager;
+ this.streamsMembershipManager = streamsMembershipManager;
List<Optional<? extends RequestManager>> list = new ArrayList<>();
list.add(coordinatorRequestManager);
@@ -93,6 +96,7 @@ public class RequestManagers implements Closeable {
list.add(Optional.of(topicMetadataRequestManager));
list.add(Optional.of(fetchRequestManager));
list.add(streamsGroupHeartbeatRequestManager);
+ list.add(streamsMembershipManager);
entries = Collections.unmodifiableList(list);
}
@@ -110,6 +114,7 @@ public class RequestManagers implements Closeable {
this.consumerMembershipManager = Optional.empty();
this.shareMembershipManager = shareMembershipManager;
this.streamsGroupHeartbeatRequestManager = Optional.empty();
+ this.streamsMembershipManager = Optional.empty();
this.offsetsRequestManager = null;
this.topicMetadataRequestManager = null;
this.fetchRequestManager = null;
@@ -194,6 +199,7 @@ public class RequestManagers implements Closeable {
CoordinatorRequestManager coordinator = null;
CommitRequestManager commitRequestManager = null;
StreamsGroupHeartbeatRequestManager streamsGroupHeartbeatRequestManager = null;
+ StreamsMembershipManager streamsMembershipManager = null;
if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
Optional<String> serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
@@ -214,43 +220,57 @@ public class RequestManagers implements Closeable {
groupRebalanceConfig.groupInstanceId,
metrics,
metadata);
- membershipManager = new ConsumerMembershipManager(
+
+ if (streamsInstanceMetadata.isPresent()) {
+ streamsMembershipManager = new StreamsMembershipManager(
groupRebalanceConfig.groupId,
- groupRebalanceConfig.groupInstanceId,
- groupRebalanceConfig.rebalanceTimeoutMs,
- serverAssignor,
+ streamsInstanceMetadata.get(),
subscriptions,
- commitRequestManager,
- metadata,
logContext,
- backgroundEventHandler,
time,
metrics);
+ streamsMembershipManager.registerStateListener(commitRequestManager);
+ streamsMembershipManager.registerStateListener(applicationThreadMemberStateListener);
- // Update the group member ID label in the client telemetry reporter.
- // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
- // Therefore, we can update the group member ID during initialization.
- if (clientTelemetryReporter.isPresent()) {
- clientTelemetryReporter.get()
- .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, membershipManager.memberId()));
- }
-
- membershipManager.registerStateListener(commitRequestManager);
- membershipManager.registerStateListener(applicationThreadMemberStateListener);
+ if (clientTelemetryReporter.isPresent()) {
+ clientTelemetryReporter.get()
+ .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, streamsMembershipManager.memberId()));
+ }
- if (streamsInstanceMetadata.isPresent()) {
streamsGroupHeartbeatRequestManager = new StreamsGroupHeartbeatRequestManager(
logContext,
time,
config,
coordinator,
- membershipManager,
+ streamsMembershipManager,
backgroundEventHandler,
metrics,
- streamsInstanceMetadata.get(),
- metadata
- );
+ streamsInstanceMetadata.get()
+ );
} else {
+ membershipManager = new ConsumerMembershipManager(
+ groupRebalanceConfig.groupId,
+ groupRebalanceConfig.groupInstanceId,
+ groupRebalanceConfig.rebalanceTimeoutMs,
+ serverAssignor,
+ subscriptions,
+ commitRequestManager,
+ metadata,
+ logContext,
+ backgroundEventHandler,
+ time,
+ metrics);
+
+ // Update the group member ID label in the client telemetry reporter.
+ // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
+ // Therefore, we can update the group member ID during initialization.
+ if (clientTelemetryReporter.isPresent()) {
+ clientTelemetryReporter.get()
+ .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, membershipManager.memberId()));
+ }
+
+ membershipManager.registerStateListener(commitRequestManager);
+ membershipManager.registerStateListener(applicationThreadMemberStateListener);
heartbeatRequestManager = new ConsumerHeartbeatRequestManager(
logContext,
time,
@@ -260,6 +280,7 @@ public class RequestManagers implements Closeable {
membershipManager,
backgroundEventHandler,
metrics);
+
}
}
@@ -284,7 +305,8 @@ public class RequestManagers implements Closeable {
Optional.ofNullable(commitRequestManager),
Optional.ofNullable(heartbeatRequestManager),
Optional.ofNullable(membershipManager),
- Optional.ofNullable(streamsGroupHeartbeatRequestManager)
+ Optional.ofNullable(streamsGroupHeartbeatRequestManager),
+ Optional.ofNullable(streamsMembershipManager)
);
}
};
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
index de6cb9e2098..8bafa619627 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
@@ -16,20 +16,37 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent;
+import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
+import org.apache.kafka.clients.consumer.internals.events.EventProcessor;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnAllTasksLostCallbackCompletedEvent;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnAllTasksLostCallbackNeededEvent;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksAssignedCallbackCompletedEvent;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksAssignedCallbackNeededEvent;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksRevokedCallbackCompletedEvent;
+import org.apache.kafka.clients.consumer.internals.events.StreamsOnTasksRevokedCallbackNeededEvent;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.function.Supplier;
/**
* Defines a self-contained object to exchange assignment-related metadata with the Kafka Streams instance.
@@ -128,6 +145,8 @@ public class StreamsAssignmentInterface {
public static class Assignment {
+ public static final Assignment EMPTY = new Assignment();
+
public final Set<TaskId> activeTasks = new HashSet<>();
public final Set<TaskId> standbyTasks = new HashSet<>();
@@ -281,17 +300,173 @@ public class StreamsAssignmentInterface {
}
}
+ private final BlockingQueue<BackgroundEvent> onCallbackRequests = new LinkedBlockingQueue<>();
+
+ private ApplicationEventHandler applicationEventHandler = null;
+
+ private Optional<Function<Set<StreamsAssignmentInterface.TaskId>, Optional<Exception>>> onTasksRevokedCallback = null;
+ private Optional<Function<Assignment, Optional<Exception>>> onTasksAssignedCallback = null;
+ private Optional<Supplier<Optional<Exception>>> onAllTasksLostCallback = null;
+
+ private final StreamsRebalanceEventProcessor streamsRebalanceEventProcessor;
+
+ private class StreamsRebalanceEventProcessor implements EventProcessor<BackgroundEvent> {
+
+ @Override
+ public void process(final BackgroundEvent event) {
+ switch (event.type()) {
+ case ERROR:
+ process((ErrorEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ process((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ process((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ process((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
+ default:
+ throw new IllegalArgumentException("Background event type " + event.type() + " was not expected");
+
+ }
+ }
+
+ private void process(final ErrorEvent event) {
+ throw event.error();
+ }
+
+ private void process(final StreamsOnTasksRevokedCallbackNeededEvent event) {
+ StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent = invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackNeededEvent event) {
+ StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent = invokeOnTasksAssignedCallback(event.assignment(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackNeededEvent event) {
+ StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent = invokeOnAllTasksLostCallback(event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke,
+ final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback = onTasksRevokedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks assignment callback set!")).apply(activeTasksToRevoke);
+
+ return exceptionFromCallback
+ .map(exception ->
+ new StreamsOnTasksRevokedCallbackCompletedEvent(
+ future,
+ Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exception, "Task revocation callback throws an error"))
+ ))
+ .orElseGet(() -> new StreamsOnTasksRevokedCallbackCompletedEvent(future, Optional.empty()));
+ }
+
+ private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallback(final StreamsAssignmentInterface.Assignment assignment,
+ final CompletableFuture<Void> future) {
+ Optional<KafkaException> error = Optional.empty();
+ // ToDo: Can we avoid the following check?
+ if (!assignment.equals(reconciledAssignment.get())) {
+
+ final Optional<Exception> exceptionFromCallback = onTasksAssignedCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks assignment callback set!")).apply(assignment);
+
+ if (exceptionFromCallback.isPresent()) {
+ error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
+ } else {
+ reconciledAssignment.set(assignment);
+ }
+ }
+ return new StreamsOnTasksAssignedCallbackCompletedEvent(future, error);
+ }
+
+ private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
+ final Optional<Exception> exceptionFromCallback = onAllTasksLostCallback
+ .orElseThrow(() -> new IllegalStateException("No tasks assignment callback set!")).get();
+
+ final Optional<KafkaException> error;
+
+ if (exceptionFromCallback.isPresent()) {
+ error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
+ } else {
+ error = Optional.empty();
+ reconciledAssignment.set(Assignment.EMPTY);
+ }
+
+ return new StreamsOnAllTasksLostCallbackCompletedEvent(future, error);
+ }
+ }
+
public StreamsAssignmentInterface(UUID processId,
Optional<HostInfo> endpoint,
Map<String, Subtopology> subtopologyMap,
- Map<String, String> clientTags
- ) {
+ Map<String, String> clientTags) {
this.processId = processId;
this.endpoint = endpoint;
this.subtopologyMap = subtopologyMap;
this.taskLags = new HashMap<>();
this.shutdownRequested = new AtomicBoolean(false);
this.clientTags = clientTags;
+ this.streamsRebalanceEventProcessor = new StreamsRebalanceEventProcessor();
+ }
+
+ public void setOnTasksRevokedCallback(final Function<Set<StreamsAssignmentInterface.TaskId>, Optional<Exception>> onTasksRevokedCallback) {
+ this.onTasksRevokedCallback = Optional.ofNullable(onTasksRevokedCallback);
+ }
+
+ public void setOnTasksAssignedCallback(final Function<Assignment, Optional<Exception>> onTasksAssignedCallback) {
+ this.onTasksAssignedCallback = Optional.ofNullable(onTasksAssignedCallback);
+ }
+
+ public void setOnAllTasksLostCallback(final Supplier<Optional<Exception>> onAllTasksLostCallback) {
+ this.onAllTasksLostCallback = Optional.ofNullable(onAllTasksLostCallback);
+ }
+
+ public void setApplicationEventHandler(final ApplicationEventHandler applicationEventHandler) {
+ this.applicationEventHandler = applicationEventHandler;
+ }
+
+ public CompletableFuture<Void> requestOnTasksAssignedCallbackInvocation(final Assignment assignment) {
+ final StreamsOnTasksAssignedCallbackNeededEvent onTasksAssignedCallbackNeededEvent = new StreamsOnTasksAssignedCallbackNeededEvent(assignment);
+ onCallbackRequests.add(onTasksAssignedCallbackNeededEvent);
+ return onTasksAssignedCallbackNeededEvent.future();
+ }
+
+ public CompletableFuture<Void> requestOnTasksRevokedCallbackInvocation(final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke) {
+ final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = new StreamsOnTasksRevokedCallbackNeededEvent(activeTasksToRevoke);
+ onCallbackRequests.add(onTasksRevokedCallbackNeededEvent);
+ return onTasksRevokedCallbackNeededEvent.future();
+ }
+
+ public CompletableFuture<Void> requestOnAllTasksLostCallbackInvocation() {
+ final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = new StreamsOnAllTasksLostCallbackNeededEvent();
+ onCallbackRequests.add(onAllTasksLostCallbackNeededEvent);
+ return onAllTasksLostCallbackNeededEvent.future();
+ }
+
+ public void processStreamsRebalanceEvents() {
+ LinkedList<BackgroundEvent> events = new LinkedList<>();
+ onCallbackRequests.drainTo(events);
+ for (BackgroundEvent event : events) {
+ streamsRebalanceEventProcessor.process(event);
+ }
}
@Override
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 7506a497939..f435b1cd21b 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
@@ -23,11 +23,8 @@ import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler
import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.clients.consumer.internals.metrics.HeartbeatMetricsManager;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.RetriableException;
-import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.TopicPartitions;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.CopartitionGroup;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TaskIds;
@@ -35,7 +32,6 @@ import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData.Endpoint;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.requests.ConsumerGroupHeartbeatResponse;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
import org.apache.kafka.common.utils.LogContext;
@@ -51,11 +47,9 @@ import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
-import java.util.stream.Stream;
public class StreamsGroupHeartbeatRequestManager implements RequestManager {
@@ -69,7 +63,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
private final StreamsGroupHeartbeatRequestManager.HeartbeatState heartbeatState;
- private final ConsumerMembershipManager membershipManager;
+ private final StreamsMembershipManager membershipManager;
private final BackgroundEventHandler backgroundEventHandler;
@@ -79,20 +73,15 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
private StreamsAssignmentInterface streamsInterface;
- private final Map<String, Uuid> assignedTopicIdCache;
-
- private final ConsumerMetadata metadata;
-
public StreamsGroupHeartbeatRequestManager(
final LogContext logContext,
final Time time,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
- final ConsumerMembershipManager membershipManager,
+ final StreamsMembershipManager membershipManager,
final BackgroundEventHandler backgroundEventHandler,
final Metrics metrics,
- final StreamsAssignmentInterface streamsAssignmentInterface,
- final ConsumerMetadata metadata
+ final StreamsAssignmentInterface streamsAssignmentInterface
) {
this.coordinatorRequestManager = coordinatorRequestManager;
this.logger = logContext.logger(getClass());
@@ -108,14 +97,12 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
this.pollTimer = time.timer(maxPollIntervalMs);
this.metricsManager = new HeartbeatMetricsManager(metrics);
this.streamsInterface = streamsAssignmentInterface;
- this.assignedTopicIdCache = new HashMap<>();
- this.metadata = metadata;
}
@Override
public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
if (!coordinatorRequestManager.coordinator().isPresent() || membershipManager.shouldSkipHeartbeat()) {
- membershipManager.onHeartbeatRequestSkipped();
+ membershipManager.transitionToUnsubscribeIfLeaving();
return NetworkClientDelegate.PollResult.EMPTY;
}
pollTimer.update(currentTimeMs);
@@ -126,7 +113,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
"messages. You can address this either by increasing max.poll.interval.ms or by " +
"reducing the maximum size of batches returned in poll() with max.poll.records.");
- membershipManager.transitionToSendingLeaveGroup(true);
+ membershipManager.onPollTimerExpired();
NetworkClientDelegate.UnsentRequest leaveHeartbeat = makeHeartbeatRequest(currentTimeMs, true);
// We can ignore the leave response because we can join before or after receiving the response.
@@ -145,7 +132,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs, Collections.singletonList(request));
}
- public ConsumerMembershipManager membershipManager() {
+ public StreamsMembershipManager membershipManager() {
return membershipManager;
}
@@ -252,14 +239,6 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
streamsInterface.partitionsByHost.set(convertHostInfoMap(data));
}
- ConsumerGroupHeartbeatResponseData cgData = new ConsumerGroupHeartbeatResponseData();
- cgData.setMemberId(data.memberId());
- cgData.setMemberEpoch(data.memberEpoch());
- cgData.setErrorCode(data.errorCode());
- cgData.setErrorMessage(data.errorMessage());
- cgData.setThrottleTimeMs(data.throttleTimeMs());
- cgData.setHeartbeatIntervalMs(data.heartbeatIntervalMs());
-
List<StreamsGroupHeartbeatResponseData.Status> statuses = data.status();
if (statuses != null && !statuses.isEmpty()) {
@@ -269,54 +248,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
logger.warn("Membership is in the following statuses: {}.", statusDetails);
}
- if (data.activeTasks() != null && data.standbyTasks() != null && data.warmupTasks() != null) {
-
- setTargetAssignment(data);
- setTargetAssignmentForConsumerGroup(data, cgData);
-
- } else {
- if (data.activeTasks() != null || data.standbyTasks() != null || data.warmupTasks() != null) {
- throw new IllegalStateException("Invalid response data, task collections must be all null or all non-null: " + data);
- }
- }
-
- membershipManager.onHeartbeatSuccess(new ConsumerGroupHeartbeatResponse(cgData));
- }
-
- private void setTargetAssignmentForConsumerGroup(final StreamsGroupHeartbeatResponseData data,
- final ConsumerGroupHeartbeatResponseData cgData) {
- Map<String, TopicPartitions> tps = new HashMap<>();
- data.activeTasks().forEach(taskId -> Stream.concat(
- streamsInterface.subtopologyMap().get(taskId.subtopologyId()).sourceTopics.stream(),
- streamsInterface.subtopologyMap().get(taskId.subtopologyId()).repartitionSourceTopics.keySet().stream()
- )
- .forEach(topic -> {
- final TopicPartitions toInsert = tps.computeIfAbsent(topic, k -> {
- final Optional<Uuid> uuid = findTopicIdInGlobalOrLocalCache(topic);
- if (uuid.isPresent()) {
- TopicPartitions t =
- new TopicPartitions();
- t.setTopicId(uuid.get());
- return t;
- } else {
- return null;
- }
- });
- if (toInsert != null) {
- toInsert.partitions().addAll(taskId.partitions());
- }
- }));
- ConsumerGroupHeartbeatResponseData.Assignment cgAssignment = new ConsumerGroupHeartbeatResponseData.Assignment();
- cgAssignment.setTopicPartitions(new ArrayList<>(tps.values()));
- cgData.setAssignment(cgAssignment);
- }
-
- private void setTargetAssignment(final StreamsGroupHeartbeatResponseData data) {
- Assignment targetAssignment = new Assignment();
- updateTaskIdCollection(data.activeTasks(), targetAssignment.activeTasks);
- updateTaskIdCollection(data.standbyTasks(), targetAssignment.standbyTasks);
- updateTaskIdCollection(data.warmupTasks(), targetAssignment.warmupTasks);
- streamsInterface.targetAssignment.set(targetAssignment);
+ membershipManager.onHeartbeatSuccess(response);
}
private static Map<StreamsAssignmentInterface.HostInfo, List<TopicPartition>> convertHostInfoMap(
@@ -410,7 +342,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
message = String.format("StreamsGroupHeartbeatRequest failed for member %s because epoch %s is fenced.",
membershipManager.memberId(), membershipManager.memberEpoch());
logInfo(message, response, currentTimeMs);
- membershipManager.transitionToFenced();
+ membershipManager.onFenced();
// Skip backoff so that a next HB to rejoin is sent as soon as the fenced member releases its assignment
heartbeatRequestState.reset();
break;
@@ -419,7 +351,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
message = String.format("StreamsGroupHeartbeatRequest failed because member %s is unknown.",
membershipManager.memberId());
logInfo(message, response, currentTimeMs);
- membershipManager.transitionToFenced();
+ membershipManager.onFenced();
// Skip backoff so that a next HB to rejoin is sent as soon as the fenced member releases its assignment
heartbeatRequestState.reset();
break;
@@ -507,22 +439,9 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
}
}
- private Optional<Uuid> findTopicIdInGlobalOrLocalCache(String topicName) {
- Uuid idFromMetadataCache = metadata.topicIds().getOrDefault(topicName, null);
- if (idFromMetadataCache != null) {
- // Add topic name to local cache, so it can be reused if included in a next target
- // assignment if metadata cache not available.
- assignedTopicIdCache.put(topicName, idFromMetadataCache);
- return Optional.of(idFromMetadataCache);
- } else {
- Uuid idFromLocalCache = assignedTopicIdCache.getOrDefault(topicName, null);
- return Optional.ofNullable(idFromLocalCache);
- }
- }
-
static class HeartbeatState {
- private final ConsumerMembershipManager membershipManager;
+ private final StreamsMembershipManager membershipManager;
private final int rebalanceTimeoutMs;
private final StreamsGroupHeartbeatRequestManager.HeartbeatState.SentFields sentFields;
@@ -533,7 +452,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
public HeartbeatState(
final StreamsAssignmentInterface streamsInterface,
- final ConsumerMembershipManager membershipManager,
+ final StreamsMembershipManager membershipManager,
final int rebalanceTimeoutMs) {
this.membershipManager = membershipManager;
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 fb34f50847b..a3f55c92d3a 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
@@ -16,9 +16,9 @@
*/
package org.apache.kafka.clients.consumer.internals;
-import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
-import org.apache.kafka.clients.consumer.internals.events.StreamsOnAssignmentCallbackCompletedEvent;
-import org.apache.kafka.clients.consumer.internals.events.StreamsOnAssignmentCallbackNeededEvent;
+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.StreamsOnTasksRevokedCallbackCompletedEvent;
import org.apache.kafka.clients.consumer.internals.metrics.ConsumerRebalanceMetricsManager;
import org.apache.kafka.clients.consumer.internals.metrics.RebalanceMetricsManager;
import org.apache.kafka.common.KafkaException;
@@ -28,13 +28,12 @@ import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
-import org.apache.kafka.common.telemetry.internals.ClientTelemetryProvider;
-import org.apache.kafka.common.telemetry.internals.ClientTelemetryReporter;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.slf4j.Logger;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -145,8 +144,6 @@ public class StreamsMembershipManager implements RequestManager {
private final SubscriptionState subscriptionState;
- private final BackgroundEventHandler backgroundEventHandler;
-
private MemberState state;
private final String groupId;
@@ -165,7 +162,7 @@ public class StreamsMembershipManager implements RequestManager {
private boolean rejoinedWhileReconciliationInProgress;
- private final Optional<ClientTelemetryReporter> clientTelemetryReporter;
+ private final List<MemberStateListener> stateUpdatesListeners = new ArrayList<>();
private LocalAssignment targetAssignment = LocalAssignment.NONE;
@@ -183,8 +180,6 @@ public class StreamsMembershipManager implements RequestManager {
final StreamsAssignmentInterface streamsAssignmentInterface,
final SubscriptionState subscriptionState,
final LogContext logContext,
- final Optional<ClientTelemetryReporter> clientTelemetryReporter,
- final BackgroundEventHandler backgroundEventHandler,
final Time time,
final Metrics metrics) {
log = logContext.logger(StreamsMembershipManager.class);
@@ -192,8 +187,6 @@ public class StreamsMembershipManager implements RequestManager {
this.groupId = groupId;
this.streamsAssignmentInterface = streamsAssignmentInterface;
this.subscriptionState = subscriptionState;
- this.clientTelemetryReporter = clientTelemetryReporter;
- this.backgroundEventHandler = backgroundEventHandler;
metricsManager = new ConsumerRebalanceMetricsManager(metrics);
this.time = time;
}
@@ -231,6 +224,14 @@ public class StreamsMembershipManager implements RequestManager {
state == MemberState.STALE;
}
+ public void registerStateListener(MemberStateListener listener) {
+ stateUpdatesListeners.add(Objects.requireNonNull(listener, "State updates listener cannot be null"));
+ }
+
+ void notifyEpochChange(Optional<Integer> epoch) {
+ stateUpdatesListeners.forEach(stateListener -> stateListener.onMemberEpochUpdated(epoch, memberId));
+ }
+
private void transitionToJoining() {
if (state == MemberState.FATAL) {
log.warn("No action taken to join the group with the updated subscription because " +
@@ -276,8 +277,9 @@ public class StreamsMembershipManager implements RequestManager {
private void transitionToStale() {
transitionTo(MemberState.STALE);
- CompletableFuture<Void> onAllTasksRevokedDone = invokeRevokingAllTasksCallback();
- staleMemberAssignmentRelease = onAllTasksRevokedDone.whenComplete((result, error) -> {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecution =
+ streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation();
+ staleMemberAssignmentRelease = onAllTasksLostCallbackExecution.whenComplete((result, error) -> {
if (error != null) {
log.error("Task revocation callback invocation failed " +
"after member left group due to expired poll timer.", error);
@@ -293,6 +295,7 @@ public class StreamsMembershipManager implements RequestManager {
MemberState previousState = state;
transitionTo(MemberState.FATAL);
log.error("Member {} with epoch {} transitioned to fatal state", memberIdInfoForLog(), memberEpoch);
+ notifyEpochChange(Optional.empty());
if (previousState == MemberState.UNSUBSCRIBED) {
log.debug("Member {} with epoch {} got fatal error from the broker but it already " +
@@ -308,8 +311,8 @@ public class StreamsMembershipManager implements RequestManager {
return;
}
- CompletableFuture<Void> callbackResult = invokeRevokingAllTasksCallback();
- callbackResult.whenComplete((result, error) -> {
+ CompletableFuture<Void> onAllTasksLostCallbackExecuted = streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation();
+ onAllTasksLostCallbackExecuted.whenComplete((result, error) -> {
if (error != null) {
log.error("onTaskAssignment callback invocation failed while releasing assignment" +
"after member failed with fatal error.", error);
@@ -359,7 +362,15 @@ public class StreamsMembershipManager implements RequestManager {
}
private void updateMemberEpoch(int newEpoch) {
- memberEpoch = newEpoch;
+ boolean newEpochReceived = this.memberEpoch != newEpoch;
+ this.memberEpoch = newEpoch;
+ if (newEpochReceived) {
+ if (memberEpoch > 0) {
+ notifyEpochChange(Optional.of(memberEpoch));
+ } else {
+ notifyEpochChange(Optional.empty());
+ }
+ }
}
private void clearPendingTaskAssignment() {
@@ -434,15 +445,6 @@ public class StreamsMembershipManager implements RequestManager {
return;
}
- // Update the group member id label in the client telemetry reporter if the member id has
- // changed. Initially the member id is empty, and it is updated when the member joins the
- // group. This is done here to avoid updating the label on every heartbeat response. Also
- // check if the member id is null, as the schema defines it as nullable.
- if (responseData.memberId() != null && !responseData.memberId().equals(memberId)) {
- clientTelemetryReporter.ifPresent(reporter -> reporter.updateMetricsLabels(
- Collections.singletonMap(ClientTelemetryProvider.GROUP_MEMBER_ID, responseData.memberId())));
- }
-
memberId = responseData.memberId();
updateMemberEpoch(responseData.memberEpoch());
@@ -519,17 +521,17 @@ public class StreamsMembershipManager implements RequestManager {
log.debug("Member {} with epoch {} transitioned to {} state. It will release its " +
"assignment and rejoin the group.", memberIdInfoForLog(), memberEpoch, MemberState.FENCED);
- CompletableFuture<Void> callbackResult = invokeRevokingAllTasksCallback();
+ CompletableFuture<Void> callbackResult = streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation();
callbackResult.whenComplete((result, error) -> {
if (error != null) {
- log.error("onTaskAssignment callback invocation failed while releasing assignment" +
+ log.error("onAllTasksLost callback invocation failed while releasing assignment" +
" after member got fenced. Member will rejoin the group anyways.", error);
}
clearTaskAndPartitionAssignment();
if (state == MemberState.FENCED) {
transitionToJoining();
} else {
- log.debug("Fenced member onTaskAssignment callback completed but the state has " +
+ log.debug("Fenced member onAllTasksLost callback completed but the state has " +
"already changed to {}, so the member won't rejoin the group", state);
}
});
@@ -615,15 +617,19 @@ public class StreamsMembershipManager implements RequestManager {
CompletableFuture<Void> onGroupLeft = new CompletableFuture<>();
leaveGroupInProgress = Optional.of(onGroupLeft);
- CompletableFuture<Void> onAllTasksRevokedDone = prepareLeaving();
- onAllTasksRevokedDone.whenComplete((__, callbackError) -> leaving(callbackError));
+ CompletableFuture<Void> onAllTasksRevokedCallbackExecuted = prepareLeaving();
+ onAllTasksRevokedCallbackExecuted.whenComplete((__, callbackError) -> leaving(callbackError));
return onGroupLeft;
}
private CompletableFuture<Void> prepareLeaving() {
transitionTo(MemberState.PREPARE_LEAVING);
- return invokeRevokingAllTasksCallback();
+ if (memberEpoch > 0) {
+ return revokeActiveTasks(toTaskIdSet(currentAssignment.activeTasks));
+ } else {
+ return releaseLostActiveTasks();
+ }
}
private void leaving(Throwable callbackError) {
@@ -636,8 +642,8 @@ public class StreamsMembershipManager implements RequestManager {
"to clear its assignment and send a leave group heartbeat",
memberIdInfoForLog());
}
- subscriptionState.unsubscribe();
clearTaskAndPartitionAssignment();
+ subscriptionState.unsubscribe();
transitionToSendingLeaveGroup(false);
}
@@ -709,41 +715,54 @@ public class StreamsMembershipManager implements RequestManager {
// ToDo: add standby and warmup tasks
SortedSet<StreamsAssignmentInterface.TaskId> assignedActiveTasks = toTaskIdSet(targetAssignment.activeTasks);
+ SortedSet<StreamsAssignmentInterface.TaskId> ownedActiveTasks = toTaskIdSet(currentAssignment.activeTasks);
+ SortedSet<StreamsAssignmentInterface.TaskId> activeTasksToRevoke = new TreeSet<>(ownedActiveTasks);
+ activeTasksToRevoke.removeAll(assignedActiveTasks);
log.info("Assigned tasks with local epoch {}\n" +
"\tMember: {}\n" +
- "\tActive tasks: {}\n",
+ "\tAssigned active tasks: {}\n" +
+ "\tOwned active tasks: {}\n" +
+ "\tActive tasks to revoke: {}\n",
targetAssignment.localEpoch,
memberIdInfoForLog(),
- assignedActiveTasks
+ assignedActiveTasks,
+ ownedActiveTasks,
+ activeTasksToRevoke
);
- SortedSet<TopicPartition> ownedTopicPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
- ownedTopicPartitions.addAll(subscriptionState.assignedPartitions());
+ SortedSet<TopicPartition> ownedTopicPartitionsFromSubscriptionState = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ ownedTopicPartitionsFromSubscriptionState.addAll(subscriptionState.assignedPartitions());
+ SortedSet<TopicPartition> ownedTopicPartitionsFromAssignedTasks =
+ topicPartitionsForActiveTasks(currentAssignment.activeTasks);
+ if (!ownedTopicPartitionsFromAssignedTasks.equals(ownedTopicPartitionsFromSubscriptionState)) {
+ throw new IllegalStateException("Owned partitions from subscription state and owned partitions from " +
+ "assigned active tasks are not equal. " +
+ "Owned partitions from subscription state: " + ownedTopicPartitionsFromSubscriptionState + ", " +
+ "Owned partitions from assigned active tasks: " + ownedTopicPartitionsFromAssignedTasks);
+ }
SortedSet<TopicPartition> assignedTopicPartitions = topicPartitionsForActiveTasks(targetAssignment.activeTasks);
SortedSet<TopicPartition> assignedTopicPartitionsNotPreviouslyOwned =
- assignedTopicPartitionsNotPreviouslyOwned(assignedTopicPartitions, ownedTopicPartitions);
+ partitionsToAssignNotPreviouslyOwned(assignedTopicPartitions, ownedTopicPartitionsFromSubscriptionState);
+ SortedSet<TopicPartition> partitionsToRevoke = new TreeSet<>(ownedTopicPartitionsFromSubscriptionState);
+ partitionsToRevoke.removeAll(assignedTopicPartitions);
- subscriptionState.assignFromSubscribedAwaitingCallback(
- assignedTopicPartitions,
- assignedTopicPartitionsNotPreviouslyOwned
- );
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted = revokeActiveTasks(activeTasksToRevoke);
+
+ final CompletableFuture<Void> onTasksRevokedAndAssignedCallbacksExecuted = onTasksRevokedCallbackExecuted.thenCompose(__ -> {
+ if (!maybeAbortReconciliation()) {
+ return assignActiveTasks(assignedActiveTasks, ownedActiveTasks);
+ }
+ return CompletableFuture.completedFuture(null);
+ });
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- new StreamsOnAssignmentCallbackNeededEvent(new StreamsAssignmentInterface.Assignment(
- assignedActiveTasks,
- Collections.emptySet(),
- Collections.emptySet()
- ));
- CompletableFuture<Void> onTasksAssignmentDone = onAssignmentCallbackNeededEvent.future();
- backgroundEventHandler.add(onAssignmentCallbackNeededEvent);
// The current target assignment is captured to ensure that acknowledging the current assignment is done with
// the same target assignment that was used when this reconciliation was initiated.
LocalAssignment currentTargetAssignment = targetAssignment;
- onTasksAssignmentDone.whenComplete((__, callbackError) -> {
+ onTasksRevokedAndAssignedCallbacksExecuted.whenComplete((__, callbackError) -> {
if (callbackError != null) {
- log.error("Reconciliation failed: onTasksAssignment callback invocation failed for tasks {}",
- targetAssignment, callbackError);
+ log.error("Reconciliation failed: callback invocation failed for tasks {}",
+ currentTargetAssignment, callbackError);
markReconciliationCompleted();
} else {
if (reconciliationInProgress && !maybeAbortReconciliation()) {
@@ -756,8 +775,61 @@ public class StreamsMembershipManager implements RequestManager {
});
}
- private SortedSet<TopicPartition> assignedTopicPartitionsNotPreviouslyOwned(final SortedSet<TopicPartition> assignedTopicPartitions,
- final SortedSet<TopicPartition> ownedTopicPartitions) {
+ private CompletableFuture<Void> revokeActiveTasks(final SortedSet<StreamsAssignmentInterface.TaskId> activeTasksToRevoke) {
+ if (activeTasksToRevoke.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ log.info("Revoking previously assigned active tasks {}", activeTasksToRevoke.stream()
+ .map(StreamsAssignmentInterface.TaskId::toString)
+ .collect(Collectors.joining(", ")));
+
+ final SortedSet<TopicPartition> partitionsToRevoke = topicPartitionsForActiveTasks(activeTasksToRevoke);
+ log.debug("Marking partitions pending for revocation: {}", partitionsToRevoke);
+ subscriptionState.markPendingRevocation(partitionsToRevoke);
+
+ return streamsAssignmentInterface.requestOnTasksRevokedCallbackInvocation(activeTasksToRevoke);
+ }
+
+ private CompletableFuture<Void> assignActiveTasks(final SortedSet<StreamsAssignmentInterface.TaskId> activeTasksToAssign,
+ final SortedSet<StreamsAssignmentInterface.TaskId> ownedActiveTasks) {
+ log.info("Assigning active tasks {}", activeTasksToAssign.stream()
+ .map(StreamsAssignmentInterface.TaskId::toString)
+ .collect(Collectors.joining(", ")));
+
+ final SortedSet<TopicPartition> partitionsToAssign = topicPartitionsForActiveTasks(activeTasksToAssign);
+ final SortedSet<TopicPartition> partitionsToAssigneNotPreviouslyOwned =
+ partitionsToAssignNotPreviouslyOwned(partitionsToAssign, topicPartitionsForActiveTasks(ownedActiveTasks));
+
+ subscriptionState.assignFromSubscribedAwaitingCallback(
+ partitionsToAssign,
+ partitionsToAssigneNotPreviouslyOwned
+ );
+
+ return streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(
+ new StreamsAssignmentInterface.Assignment(
+ activeTasksToAssign,
+ Collections.emptySet(),
+ Collections.emptySet()
+ )
+ );
+ }
+
+ private CompletableFuture<Void> releaseLostActiveTasks() {
+ final SortedSet<StreamsAssignmentInterface.TaskId> activeTasksToRelease = toTaskIdSet(currentAssignment.activeTasks);
+ log.info("Revoking previously assigned and now lost active tasks {}", activeTasksToRelease.stream()
+ .map(StreamsAssignmentInterface.TaskId::toString)
+ .collect(Collectors.joining(", ")));
+
+ final SortedSet<TopicPartition> partitionsToRelease = topicPartitionsForActiveTasks(activeTasksToRelease);
+ log.debug("Marking lost partitions pending for revocation: {}", partitionsToRelease);
+ subscriptionState.markPendingRevocation(partitionsToRelease);
+
+ return streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation();
+ }
+
+ private SortedSet<TopicPartition> partitionsToAssignNotPreviouslyOwned(final SortedSet<TopicPartition> assignedTopicPartitions,
+ final SortedSet<TopicPartition> ownedTopicPartitions) {
SortedSet<TopicPartition> assignedPartitionsNotPreviouslyOwned = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
assignedPartitionsNotPreviouslyOwned.addAll(assignedTopicPartitions);
assignedPartitionsNotPreviouslyOwned.removeAll(ownedTopicPartitions);
@@ -779,6 +851,19 @@ public class StreamsMembershipManager implements RequestManager {
return topicPartitions;
}
+ private SortedSet<TopicPartition> topicPartitionsForActiveTasks(final SortedSet<StreamsAssignmentInterface.TaskId> activeTasks) {
+ final SortedSet<TopicPartition> topicPartitions = new TreeSet<>(TOPIC_PARTITION_COMPARATOR);
+ activeTasks.forEach(task ->
+ Stream.concat(
+ streamsAssignmentInterface.subtopologyMap().get(task.subtopologyId()).sourceTopics.stream(),
+ streamsAssignmentInterface.subtopologyMap().get(task.subtopologyId()).repartitionSourceTopics.keySet().stream()
+ ).forEach(topic -> {
+ topicPartitions.add(new TopicPartition(topic, task.partitionId()));
+ })
+ );
+ return topicPartitions;
+ }
+
private void markReconciliationCompleted() {
reconciliationInProgress = false;
rejoinedWhileReconciliationInProgress = false;
@@ -801,28 +886,47 @@ public class StreamsMembershipManager implements RequestManager {
rejoinedWhileReconciliationInProgress = false;
}
- private CompletableFuture<Void> invokeRevokingAllTasksCallback() {
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- new StreamsOnAssignmentCallbackNeededEvent(new StreamsAssignmentInterface.Assignment(
- Collections.emptySet(),
- Collections.emptySet(),
- Collections.emptySet()
- ));
- backgroundEventHandler.add(onAssignmentCallbackNeededEvent);
- return onAssignmentCallbackNeededEvent.future();
+ public void onTasksRevokedCallbackCompleted(final StreamsOnTasksRevokedCallbackCompletedEvent event) {
+ Optional<KafkaException> error = event.error();
+ CompletableFuture<Void> future = event.future();
+
+ if (error.isPresent()) {
+ Exception e = error.get();
+ log.warn("The onTasksRevoked callback completed with an error ({}); " +
+ "signaling to continue to the next phase of rebalance", e.getMessage());
+ future.completeExceptionally(e);
+ } else {
+ log.debug("The onTasksRevoked callback completed successfully; signaling to continue to the next phase of rebalance");
+ future.complete(null);
+ }
+ }
+
+ public void onTasksAssignedCallbackCompleted(final StreamsOnTasksAssignedCallbackCompletedEvent event) {
+ Optional<KafkaException> error = event.error();
+ CompletableFuture<Void> future = event.future();
+
+ if (error.isPresent()) {
+ Exception e = error.get();
+ log.warn("The onTasksAssigned callback completed with an error ({}); " +
+ "signaling to continue to the next phase of rebalance", e.getMessage());
+ future.completeExceptionally(e);
+ } else {
+ log.debug("The onTasksAssigned callback completed successfully; signaling to continue to the next phase of rebalance");
+ future.complete(null);
+ }
}
- public void onTaskAssignmentCallbackCompleted(StreamsOnAssignmentCallbackCompletedEvent event) {
+ public void onAllTasksLostCallbackCompleted(final StreamsOnAllTasksLostCallbackCompletedEvent event) {
Optional<KafkaException> error = event.error();
CompletableFuture<Void> future = event.future();
if (error.isPresent()) {
Exception e = error.get();
- log.warn("The onTaskAssignment callback completed with an error ({}); " +
+ log.warn("The onAllTasksLost callback completed with an error ({}); " +
"signaling to continue to the next phase of rebalance", e.getMessage());
future.completeExceptionally(e);
} else {
- log.debug("The onTaskAssignment callback completed successfully; signaling to continue to the next phase of rebalance");
+ log.debug("The onAllTasksLost callback completed successfully; signaling to continue to the next phase of rebalance");
future.complete(null);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java
index dc0c393e279..b23584b053c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java
@@ -40,7 +40,9 @@ public abstract class ApplicationEvent {
SHARE_ACKNOWLEDGE_ON_CLOSE,
SHARE_ACKNOWLEDGEMENT_COMMIT_CALLBACK_REGISTRATION,
SEEK_UNVALIDATED,
- STREAMS_ON_ASSIGNMENT_CALLBACK_COMPLETED,
+ STREAMS_ON_TASKS_ASSIGNED_CALLBACK_COMPLETED,
+ STREAMS_ON_TASKS_REVOKED_CALLBACK_COMPLETED,
+ STREAMS_ON_ALL_TASKS_LOST_CALLBACK_COMPLETED,
}
private final Type type;
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 135857a06b6..f36852c091a 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
@@ -180,6 +180,18 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
process((SeekUnvalidatedEvent) event);
return;
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_COMPLETED:
+ process((StreamsOnTasksRevokedCallbackCompletedEvent) event);
+ return;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_COMPLETED:
+ process((StreamsOnTasksAssignedCallbackCompletedEvent) event);
+ return;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_COMPLETED:
+ process((StreamsOnAllTasksLostCallbackCompletedEvent) event);
+ return;
+
default:
log.warn("Application event type {} was not expected", event.type());
}
@@ -289,21 +301,29 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
* it is already a member on the next poll.
*/
private void process(final TopicSubscriptionChangeEvent event) {
- if (requestManagers.consumerHeartbeatRequestManager.isEmpty()) {
+ if (requestManagers.consumerMembershipManager.isPresent()) {
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener())) {
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ }
+ requestManagers.consumerMembershipManager.get().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ } else if (requestManagers.streamsMembershipManager.isPresent()) {
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener())) {
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ }
+ requestManagers.streamsMembershipManager.get().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ } else {
log.warn("Group membership manager not present when processing a subscribe event");
event.future().complete(null);
- return;
- }
-
- try {
- if (subscriptions.subscribe(event.topics(), event.listener()))
- this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
-
- // Join the group if not already part of it, or just send the new subscription to the broker on the next poll.
- requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
- event.future().complete(null);
- } catch (Exception e) {
- event.future().completeExceptionally(e);
}
}
@@ -374,6 +394,9 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
if (requestManagers.consumerMembershipManager.isPresent()) {
CompletableFuture<Void> future = requestManagers.consumerMembershipManager.get().leaveGroup();
future.whenComplete(complete(event.future()));
+ } else if (requestManagers.streamsMembershipManager.isPresent()) {
+ CompletableFuture<Void> future = requestManagers.streamsMembershipManager.get().leaveGroup();
+ future.whenComplete(complete(event.future()));
} else {
// If the consumer is not using the group management capabilities, we still need to clear all assignments it may have.
subscriptions.unsubscribe();
@@ -548,6 +571,33 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
manager.setAcknowledgementCommitCallbackRegistered(event.isCallbackRegistered());
}
+ private void process(final StreamsOnTasksRevokedCallbackCompletedEvent event) {
+ if (!requestManagers.streamsMembershipManager.isPresent()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onTasksRevoked callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onTasksRevokedCallbackCompleted(event);
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackCompletedEvent event) {
+ if (!requestManagers.streamsMembershipManager.isPresent()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onTasksAssigned callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onTasksAssignedCallbackCompleted(event);
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackCompletedEvent event) {
+ if (!requestManagers.streamsMembershipManager.isPresent()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onAllTasksLost callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onAllTasksLostCallbackCompleted(event);
+ }
+
private <T> BiConsumer<? super T, ? super Throwable> complete(final CompletableFuture<T> b) {
return (value, exception) -> {
if (exception != null)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java
index cffcca42f48..b18611895f9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java
@@ -28,7 +28,8 @@ public abstract class BackgroundEvent {
public enum Type {
ERROR, CONSUMER_REBALANCE_LISTENER_CALLBACK_NEEDED, SHARE_ACKNOWLEDGEMENT_COMMIT_CALLBACK,
- STREAMS_ON_ASSIGNMENT_CALLBACK_NEEDED
+ STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED, STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED,
+ STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED
}
private final Type type;
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackCompletedEvent.java
similarity index 82%
copy from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
copy to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackCompletedEvent.java
index 5b702efb62c..b84e9d0c138 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackCompletedEvent.java
@@ -22,14 +22,14 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class StreamsOnAssignmentCallbackCompletedEvent extends ApplicationEvent {
+public class StreamsOnAllTasksLostCallbackCompletedEvent extends ApplicationEvent {
private final CompletableFuture<Void> future;
private final Optional<KafkaException> error;
- public StreamsOnAssignmentCallbackCompletedEvent(final CompletableFuture<Void> future,
- final Optional<KafkaException> error) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_COMPLETED);
+ public StreamsOnAllTasksLostCallbackCompletedEvent(final CompletableFuture<Void> future,
+ final Optional<KafkaException> error) {
+ super(Type.STREAMS_ON_ALL_TASKS_LOST_CALLBACK_COMPLETED);
this.future = Objects.requireNonNull(future);
this.error = Objects.requireNonNull(error);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackNeededEvent.java
similarity index 57%
copy from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
copy to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackNeededEvent.java
index 5ef6cd9c462..294cf3e2fde 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAllTasksLostCallbackNeededEvent.java
@@ -16,26 +16,15 @@
*/
package org.apache.kafka.clients.consumer.internals.events;
-import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface;
+public class StreamsOnAllTasksLostCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
-import java.util.Objects;
-
-public class StreamsOnAssignmentCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
-
- private final StreamsAssignmentInterface.Assignment assignment;
-
- public StreamsOnAssignmentCallbackNeededEvent(StreamsAssignmentInterface.Assignment assignment) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_NEEDED, Long.MAX_VALUE);
- this.assignment = Objects.requireNonNull(assignment);
- }
-
- public StreamsAssignmentInterface.Assignment assignment() {
- return assignment;
+ public StreamsOnAllTasksLostCallbackNeededEvent() {
+ super(Type.STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED, Long.MAX_VALUE);
}
@Override
protected String toStringBase() {
- return super.toStringBase() +
- ", assignment=" + assignment;
+ return super.toStringBase();
}
}
+
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackCompletedEvent.java
similarity index 81%
copy from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
copy to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackCompletedEvent.java
index 5b702efb62c..96c2519bb2d 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackCompletedEvent.java
@@ -22,14 +22,14 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class StreamsOnAssignmentCallbackCompletedEvent extends ApplicationEvent {
+public class StreamsOnTasksAssignedCallbackCompletedEvent extends ApplicationEvent {
private final CompletableFuture<Void> future;
private final Optional<KafkaException> error;
- public StreamsOnAssignmentCallbackCompletedEvent(final CompletableFuture<Void> future,
- final Optional<KafkaException> error) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_COMPLETED);
+ public StreamsOnTasksAssignedCallbackCompletedEvent(final CompletableFuture<Void> future,
+ final Optional<KafkaException> error) {
+ super(Type.STREAMS_ON_TASKS_ASSIGNED_CALLBACK_COMPLETED);
this.future = Objects.requireNonNull(future);
this.error = Objects.requireNonNull(error);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackNeededEvent.java
similarity index 82%
copy from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
copy to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackNeededEvent.java
index 5ef6cd9c462..276eba3ca4a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksAssignedCallbackNeededEvent.java
@@ -20,12 +20,12 @@ import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface;
import java.util.Objects;
-public class StreamsOnAssignmentCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
+public class StreamsOnTasksAssignedCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
private final StreamsAssignmentInterface.Assignment assignment;
- public StreamsOnAssignmentCallbackNeededEvent(StreamsAssignmentInterface.Assignment assignment) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_NEEDED, Long.MAX_VALUE);
+ public StreamsOnTasksAssignedCallbackNeededEvent(StreamsAssignmentInterface.Assignment assignment) {
+ super(Type.STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED, Long.MAX_VALUE);
this.assignment = Objects.requireNonNull(assignment);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackCompletedEvent.java
similarity index 82%
rename from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
rename to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackCompletedEvent.java
index 5b702efb62c..5717012ac45 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackCompletedEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackCompletedEvent.java
@@ -22,14 +22,14 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-public class StreamsOnAssignmentCallbackCompletedEvent extends ApplicationEvent {
+public class StreamsOnTasksRevokedCallbackCompletedEvent extends ApplicationEvent {
private final CompletableFuture<Void> future;
private final Optional<KafkaException> error;
- public StreamsOnAssignmentCallbackCompletedEvent(final CompletableFuture<Void> future,
- final Optional<KafkaException> error) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_COMPLETED);
+ public StreamsOnTasksRevokedCallbackCompletedEvent(final CompletableFuture<Void> future,
+ final Optional<KafkaException> error) {
+ super(Type.STREAMS_ON_TASKS_REVOKED_CALLBACK_COMPLETED);
this.future = Objects.requireNonNull(future);
this.error = Objects.requireNonNull(error);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackNeededEvent.java
similarity index 62%
rename from clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
rename to clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackNeededEvent.java
index 5ef6cd9c462..be7ae13b446 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnAssignmentCallbackNeededEvent.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/StreamsOnTasksRevokedCallbackNeededEvent.java
@@ -19,23 +19,25 @@ package org.apache.kafka.clients.consumer.internals.events;
import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface;
import java.util.Objects;
+import java.util.Set;
-public class StreamsOnAssignmentCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
+public class StreamsOnTasksRevokedCallbackNeededEvent extends CompletableBackgroundEvent<Void> {
- private final StreamsAssignmentInterface.Assignment assignment;
+ private final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke;
- public StreamsOnAssignmentCallbackNeededEvent(StreamsAssignmentInterface.Assignment assignment) {
- super(Type.STREAMS_ON_ASSIGNMENT_CALLBACK_NEEDED, Long.MAX_VALUE);
- this.assignment = Objects.requireNonNull(assignment);
+ public StreamsOnTasksRevokedCallbackNeededEvent(final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke) {
+ super(Type.STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED, Long.MAX_VALUE);
+ this.activeTasksToRevoke = Objects.requireNonNull(activeTasksToRevoke);
}
- public StreamsAssignmentInterface.Assignment assignment() {
- return assignment;
+ public Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke() {
+ return activeTasksToRevoke;
}
@Override
protected String toStringBase() {
return super.toStringBase() +
- ", assignment=" + assignment;
+ ", active tasks to revoke=" + activeTasksToRevoke;
}
}
+
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
index 9e88e2c7569..a9e34cf2067 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
@@ -19,22 +19,17 @@ package org.apache.kafka.clients.consumer.internals;
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest;
-import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.Assignment;
import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.Subtopology;
-import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.TaskId;
import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.TopicInfo;
import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.TopicPartitions;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.requests.ConsumerGroupHeartbeatResponse;
import org.apache.kafka.common.requests.RequestHeader;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
@@ -59,7 +54,6 @@ import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
-import java.util.stream.Collectors;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -93,7 +87,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
private CoordinatorRequestManager coordinatorRequestManager;
@Mock
- private ConsumerMembershipManager membershipManager;
+ private StreamsMembershipManager membershipManager;
@Mock
private BackgroundEventHandler backgroundEventHandler;
@@ -141,8 +135,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
membershipManager,
backgroundEventHandler,
metrics,
- streamsAssignmentInterface,
- metadata
+ streamsAssignmentInterface
);
when(membershipManager.groupId()).thenReturn(TEST_GROUP_ID);
@@ -160,7 +153,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, result.unsentRequests.size());
- verify(membershipManager).onHeartbeatRequestSkipped();
+ verify(membershipManager).transitionToUnsubscribeIfLeaving();
}
@Test
@@ -171,7 +164,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
assertEquals(0, result.unsentRequests.size());
- verify(membershipManager).onHeartbeatRequestSkipped();
+ verify(membershipManager).transitionToUnsubscribeIfLeaving();
}
@Test
@@ -347,35 +340,35 @@ class StreamsGroupHeartbeatRequestManagerTest {
mockResponse(data);
- ArgumentCaptor<ConsumerGroupHeartbeatResponse> captor = ArgumentCaptor.forClass(ConsumerGroupHeartbeatResponse.class);
+ ArgumentCaptor<StreamsGroupHeartbeatResponse> captor = ArgumentCaptor.forClass(StreamsGroupHeartbeatResponse.class);
verify(membershipManager, times(1)).onHeartbeatSuccess(captor.capture());
- ConsumerGroupHeartbeatResponseData response = captor.getValue().data();
+ StreamsGroupHeartbeatResponseData response = captor.getValue().data();
assertEquals(Errors.NONE.code(), response.errorCode());
assertEquals(TEST_MEMBER_ID, response.memberId());
assertEquals(TEST_MEMBER_EPOCH, response.memberEpoch());
assertEquals(TEST_THROTTLE_TIME_MS, response.throttleTimeMs());
assertEquals(1000, response.heartbeatIntervalMs());
- final List<TopicPartitions> tps = response.assignment().topicPartitions();
- assertEquals(2, tps.size());
- assertEquals(Set.of(uuid0, uuid1), tps.stream().map(TopicPartitions::topicId).collect(Collectors.toSet()));
- assertEquals(Collections.singletonList(0), tps.get(0).partitions());
- assertEquals(Collections.singletonList(0), tps.get(1).partitions());
-
- final Assignment targetAssignment = streamsAssignmentInterface.targetAssignment.get();
- assertEquals(1, targetAssignment.activeTasks.size());
- final TaskId activeTaskId = targetAssignment.activeTasks.stream().findFirst().get();
- assertEquals(activeTaskId.subtopologyId(), "0");
- assertEquals(activeTaskId.partitionId(), 0);
-
- assertEquals(1, targetAssignment.standbyTasks.size());
- final TaskId standbyTaskId = targetAssignment.standbyTasks.stream().findFirst().get();
- assertEquals(standbyTaskId.subtopologyId(), "1");
- assertEquals(standbyTaskId.partitionId(), 1);
-
- assertEquals(1, targetAssignment.warmupTasks.size());
- final TaskId warmupTaskId = targetAssignment.warmupTasks.stream().findFirst().get();
- assertEquals(warmupTaskId.subtopologyId(), "2");
- assertEquals(warmupTaskId.partitionId(), 2);
+// final List<TopicPartitions> tps = response.assign.topicPartitions();
+// assertEquals(2, tps.size());
+// assertEquals(Set.of(uuid0, uuid1), tps.stream().map(TopicPartitions::topicId).collect(Collectors.toSet()));
+// assertEquals(Collections.singletonList(0), tps.get(0).partitions());
+// assertEquals(Collections.singletonList(0), tps.get(1).partitions());
+
+// final Assignment targetAssignment = streamsAssignmentInterface.targetAssignment.get();
+// assertEquals(1, targetAssignment.activeTasks.size());
+// final TaskId activeTaskId = targetAssignment.activeTasks.stream().findFirst().get();
+// assertEquals(activeTaskId.subtopologyId(), "0");
+// assertEquals(activeTaskId.partitionId(), 0);
+//
+// assertEquals(1, targetAssignment.standbyTasks.size());
+// final TaskId standbyTaskId = targetAssignment.standbyTasks.stream().findFirst().get();
+// assertEquals(standbyTaskId.subtopologyId(), "1");
+// assertEquals(standbyTaskId.partitionId(), 1);
+//
+// assertEquals(1, targetAssignment.warmupTasks.size());
+// final TaskId warmupTaskId = targetAssignment.warmupTasks.stream().findFirst().get();
+// assertEquals(warmupTaskId.subtopologyId(), "2");
+// assertEquals(warmupTaskId.partitionId(), 2);
}
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 7d954477da9..1ee0975a8cb 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
@@ -18,8 +18,9 @@ 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.StreamsOnAssignmentCallbackCompletedEvent;
-import org.apache.kafka.clients.consumer.internals.events.StreamsOnAssignmentCallbackNeededEvent;
+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;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
@@ -34,6 +35,8 @@ 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;
@@ -54,6 +57,7 @@ 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.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -85,6 +89,9 @@ public class StreamsMembershipManagerTest {
@Mock
private StreamsAssignmentInterface streamsAssignmentInterface;
+ @Captor
+ private ArgumentCaptor<StreamsOnTasksAssignedCallbackNeededEvent> onAssignmentCallbackNeededEventCaptor;
+
private Queue<BackgroundEvent> backgroundEventQueue = new LinkedList<>();
private BackgroundEventHandler backgroundEventHandler = new BackgroundEventHandler(backgroundEventQueue);
@@ -95,8 +102,6 @@ public class StreamsMembershipManagerTest {
streamsAssignmentInterface,
subscriptionState,
new LogContext("test"),
- Optional.empty(),
- backgroundEventHandler,
time,
metrics
);
@@ -151,62 +156,87 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingEmptyToSingleActiveTask() {
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> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedPartitionAssignment);
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final Set<StreamsAssignmentInterface.TaskId> activeTasks =
- Set.of(new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0));
- final StreamsAssignmentInterface.Assignment expectedTaskAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedTaskAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ final Collection<TopicPartition> expectedPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
+ verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedPartitionsToAssign);
+ verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ onTasksAssignedCallbackExecuted.complete(null);
+ verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateAcknowledging(membershipManager);
+ verify(streamsAssignmentInterface, never()).requestOnTasksRevokedCallbackInvocation(any());
}
@Test
public void testReconcilingActiveTaskToDifferentActiveTask() {
setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ when(streamsAssignmentInterface.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
+ .thenReturn(onTasksRevokedCallbackExecuted);
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
.thenReturn(Collections.emptySet())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedPartitionAssignment);
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
- );
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
+ final Collection<TopicPartition> expectedPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
+ verify(subscriptionState).markPendingRevocation(expectedPartitionsToRevoke);
+ verify(subscriptionState, never()).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedPartitionsToAssign);
+ onTasksRevokedCallbackExecuted.complete(null);
+ verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedPartitionsToAssign);
+ verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ onTasksAssignedCallbackExecuted.complete(null);
+ verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateAcknowledging(membershipManager);
}
@Test
public void testReconcilingSingleActiveTaskToAdditionalActiveTask() {
setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0),
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
.thenReturn(Collections.emptySet())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
@@ -214,57 +244,58 @@ public class StreamsMembershipManagerTest {
new TopicPartition(TOPIC_0, PARTITION_0),
new TopicPartition(TOPIC_0, PARTITION_1)
);
- final Collection<TopicPartition> expectedAdditionalPartitionAssignment = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(
- expectedPartitionAssignment,
- expectedAdditionalPartitionAssignment
- );
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0),
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
- );
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedAdditionalPartitionAssignment);
+ final Collection<TopicPartition> expectedAdditionalPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
+ verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedAdditionalPartitionsToAssign);
+ verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedAdditionalPartitionAssignment);
+ onTasksAssignedCallbackExecuted.complete(null);
+ verify(subscriptionState).enablePartitionsAwaitingCallback(expectedAdditionalPartitionsToAssign);
verifyInStateAcknowledging(membershipManager);
+ verify(streamsAssignmentInterface, never()).requestOnTasksRevokedCallbackInvocation(any());
}
@Test
public void testReconcilingMultipleActiveTaskToSingleActiveTask() {
setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0),
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
+ );
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ when(streamsAssignmentInterface.requestOnTasksRevokedCallbackInvocation(activeTasksToRevoke))
+ .thenReturn(onTasksRevokedCallbackExecuted);
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
.thenReturn(Collections.emptySet())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0), new TopicPartition(TOPIC_0, PARTITION_1)));
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(
- new TopicPartition(TOPIC_0, PARTITION_1)
- );
- final Collection<TopicPartition> expectedAdditionalPartitionAssignment = Collections.emptySet();
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(
- expectedPartitionAssignment,
- expectedAdditionalPartitionAssignment
- );
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_1)
- );
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedAdditionalPartitionAssignment);
+ final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
+ final Collection<TopicPartition> expectedPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
+ final Collection<TopicPartition> expectedAdditionalPartitionsToAssign = Collections.emptySet();
+ verify(subscriptionState).markPendingRevocation(expectedPartitionsToRevoke);
+ verify(subscriptionState, never()).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedAdditionalPartitionsToAssign);
+ onTasksRevokedCallbackExecuted.complete(null);
+ verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedAdditionalPartitionsToAssign);
+ verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedAdditionalPartitionsToAssign);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedAdditionalPartitionAssignment);
+ onTasksAssignedCallbackExecuted.complete(null);
+ verify(subscriptionState).enablePartitionsAwaitingCallback(expectedAdditionalPartitionsToAssign);
verifyInStateAcknowledging(membershipManager);
}
@@ -274,79 +305,50 @@ public class StreamsMembershipManagerTest {
SUB_TOPOLOGY_ID_0, TOPIC_0,
SUB_TOPOLOGY_ID_1, TOPIC_1
);
- joining();
-
- reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0), SUB_TOPOLOGY_ID_1, List.of(PARTITION_0)));
-
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(
- new TopicPartition(TOPIC_0, PARTITION_0),
- new TopicPartition(TOPIC_1, PARTITION_0)
- );
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedPartitionAssignment);
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0),
new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_1, PARTITION_0)
);
- final StreamsAssignmentInterface.Assignment expectedTaskAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedTaskAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
- verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
- verifyInStateAcknowledging(membershipManager);
- }
-
- @Test
- public void testReconcilingEmptyToMultipleActiveTaskOfConcatenatedSubtopologies() {
- setupStreamsAssignmentInterfaceWithTwoConcatenedSubtopologies(
- SUB_TOPOLOGY_ID_0, TOPIC_0,
- SUB_TOPOLOGY_ID_1, TOPIC_1
- );
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
joining();
- reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0), SUB_TOPOLOGY_ID_1, List.of(PARTITION_0)));
+ reconcile(makeHeartbeatResponse(
+ SUB_TOPOLOGY_ID_0, List.of(PARTITION_0),
+ SUB_TOPOLOGY_ID_1, List.of(PARTITION_0))
+ );
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(
+ final Collection<TopicPartition> expectedPartitionsToAssign = Set.of(
new TopicPartition(TOPIC_0, PARTITION_0),
new TopicPartition(TOPIC_1, PARTITION_0)
);
- verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedPartitionAssignment);
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final Set<StreamsAssignmentInterface.TaskId> activeTasks = Set.of(
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0),
- new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_1, PARTITION_0)
- );
- final StreamsAssignmentInterface.Assignment expectedTaskAssignment = makeTaskAssignment(activeTasks);
- assertEquals(expectedTaskAssignment, onAssignmentCallbackNeededEvent.assignment());
- verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionsToAssign, expectedPartitionsToAssign);
+ verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().complete(null);
- verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
+ onTasksAssignedCallbackExecuted.complete(null);
+ verify(subscriptionState).enablePartitionsAwaitingCallback(expectedPartitionsToAssign);
verifyInStateAcknowledging(membershipManager);
+ verify(streamsAssignmentInterface, never()).requestOnTasksRevokedCallbackInvocation(any());
}
@Test
public void testReconcilingAndAssignmentCallbackFails() {
- final String topicName = "test_topic";
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, topicName);
+ setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsAssignmentInterface.TaskId> activeTasks =
Set.of(new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0));
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(activeTasks);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks)))
+ .thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- final Collection<TopicPartition> expectedPartitionAssignment = Set.of(new TopicPartition(topicName, PARTITION_0));
+ final Collection<TopicPartition> expectedPartitionAssignment = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
verify(subscriptionState).assignFromSubscribedAwaitingCallback(expectedPartitionAssignment, expectedPartitionAssignment);
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
verifyInStateReconciling(membershipManager);
- onAssignmentCallbackNeededEvent.future().completeExceptionally(new RuntimeException("KABOOM!"));
+ onTasksAssignedCallbackExecuted.completeExceptionally(new RuntimeException("KABOOM!"));
verifyInStateReconciling(membershipManager);
verify(subscriptionState, never()).enablePartitionsAwaitingCallback(expectedPartitionAssignment);
}
@@ -365,6 +367,9 @@ public class StreamsMembershipManagerTest {
@Test
public void testLeaveGroupWhenNotInGroupAndFenced() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
fenced();
final CompletableFuture<Void> future = membershipManager.leaveGroup();
@@ -380,39 +385,52 @@ public class StreamsMembershipManagerTest {
@Test
public void testLeaveGroupWhenInGroupWithAssignment() {
- final StreamsAssignmentInterface.Assignment emptyStreamsAssignment = makeTaskAssignment(Collections.emptySet());
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic");
+ 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);
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksRevokedCallbackInvocation(activeTasks))
+ .thenReturn(onTasksRevokedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> future = membershipManager.leaveGroup();
+ final CompletableFuture<Void> onGroupLeft = membershipManager.leaveGroup();
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- assertEquals(emptyStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
+ assertFalse(onGroupLeft.isDone());
verify(subscriptionState, never()).unsubscribe();
verifyInStatePrepareLeaving(membershipManager);
- final CompletableFuture<Void> futureBeforeRevocationCallback = membershipManager.leaveGroup();
- assertEquals(future, futureBeforeRevocationCallback);
- onAssignmentCallbackNeededEvent.future().complete(null);
+ final CompletableFuture<Void> onGroupLeftBeforeRevocationCallback = membershipManager.leaveGroup();
+ assertEquals(onGroupLeft, onGroupLeftBeforeRevocationCallback);
+ onTasksRevokedCallbackExecuted.complete(null);
verify(subscriptionState).unsubscribe();
- assertFalse(future.isDone());
+ assertFalse(onGroupLeft.isDone());
verifyInStateLeaving(membershipManager);
- final CompletableFuture<Void> futureAfterRevocationCallback = membershipManager.leaveGroup();
- assertEquals(future, futureAfterRevocationCallback);
+ final CompletableFuture<Void> onGroupLeftAfterRevocationCallback = membershipManager.leaveGroup();
+ assertEquals(onGroupLeft, onGroupLeftAfterRevocationCallback);
membershipManager.transitionToUnsubscribeIfLeaving();
verifyInStateUnsubscribed(membershipManager);
}
@Test
public void testTransitionToUnsubscribeWhenInLeaving() {
- final StreamsGroupHeartbeatResponse response = makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0));
setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic");
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ final CompletableFuture<Void> onAllTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
+ .thenReturn(onAllTasksRevokedCallbackExecuted);
joining();
- reconcile(response);
- acknowledging();
- CompletableFuture<Void> future = leaving();
+ reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
+ CompletableFuture<Void> future = leaving(onAllTasksRevokedCallbackExecuted);
membershipManager.transitionToUnsubscribeIfLeaving();
@@ -424,10 +442,15 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnPollTimerExpired() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic");
+ 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();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
membershipManager.onPollTimerExpired();
@@ -437,10 +460,15 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInAcknowleding() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic");
+ 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();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
membershipManager.onHeartbeatRequestGenerated();
@@ -449,11 +477,16 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInAcknowledgingAndNewTargetAssignment() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ 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)));
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_1)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
membershipManager.onHeartbeatRequestGenerated();
@@ -462,8 +495,11 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInLeaving() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- leaving();
+ leaving(onAllTasksLostCallbackExecuted);
membershipManager.onHeartbeatRequestGenerated();
@@ -472,6 +508,9 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInLeavingAndPollTimerExpired() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
membershipManager.onPollTimerExpired();
@@ -489,51 +528,69 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ 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)));
- backgroundEventQueue.poll();
testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
@Test
public void testOnFencedWhenInAcknowledging() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ 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();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
@Test
public void testOnFencedWhenInStable() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ 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();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
stable();
testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
private void testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
+
membershipManager.onFenced();
verifyInStateFenced(membershipManager);
assertEquals(StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(Collections.emptySet());
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
- onAssignmentCallbackNeededEvent.future().complete(null);
+ verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
+ onAllTasksLostCallbackExecuted.complete(null);
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
verifyInStateJoining(membershipManager);
}
@Test
public void testOnFencedWhenInPrepareLeaving() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
testOnFencedWhenInPrepareLeavingOrLeaving(prepareLeaving());
@@ -541,9 +598,12 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInLeaving() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testOnFencedWhenInPrepareLeavingOrLeaving(leaving());
+ testOnFencedWhenInPrepareLeavingOrLeaving(leaving(onAllTasksLostCallbackExecuted));
}
private void testOnFencedWhenInPrepareLeavingOrLeaving(final CompletableFuture<Void> onGroupLeft) {
@@ -558,6 +618,9 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInPrepareLeaving() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
testTransitionToFatalWhenInPrepareLeavingOrLeaving(prepareLeaving());
@@ -565,9 +628,12 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInLeaving() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testTransitionToFatalWhenInPrepareLeavingOrLeaving(leaving());
+ testTransitionToFatalWhenInPrepareLeavingOrLeaving(leaving(onAllTasksLostCallbackExecuted));
}
private void testTransitionToFatalWhenInPrepareLeavingOrLeaving(final CompletableFuture<Void> onGroupLeft) {
@@ -581,63 +647,89 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInJoining() {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
}
@Test
public void testTransitionToFatalWhenInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- backgroundEventQueue.poll();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
}
@Test
public void testTransitionToFatalWhenInAcknowledging() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
}
@Test
public void testTransitionToFatalWhenInStable() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, "topic0");
+ setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUB_TOPOLOGY_ID_0, TOPIC_0);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ final Set<StreamsAssignmentInterface.TaskId> activeTasksSetup = Set.of(
+ new StreamsAssignmentInterface.TaskId(SUB_TOPOLOGY_ID_0, PARTITION_0)
+ );
+ when(streamsAssignmentInterface.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup)))
+ .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ when(streamsAssignmentInterface.requestOnAllTasksLostCallbackInvocation())
+ .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponse(SUB_TOPOLOGY_ID_0, List.of(PARTITION_0)));
- acknowledging();
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
stable();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
}
- private void testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable() {
+ private void testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(final CompletableFuture<Void> future) {
membershipManager.transitionToFatal();
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- final StreamsAssignmentInterface.Assignment expectedStreamsAssignment = makeTaskAssignment(Collections.emptySet());
- assertEquals(expectedStreamsAssignment, onAssignmentCallbackNeededEvent.assignment());
- onAssignmentCallbackNeededEvent.future().complete(null);
+ verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
+ future.complete(null);
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
verifyInStateFatal(membershipManager);
}
@Test
- public void testOnTaskAssignmentCallbackCompleted() {
+ public void testOnTasksAssignedCallbackCompleted() {
final CompletableFuture<Void> future = new CompletableFuture<>();
- final StreamsOnAssignmentCallbackCompletedEvent event = new StreamsOnAssignmentCallbackCompletedEvent(
+ final StreamsOnTasksAssignedCallbackCompletedEvent event = new StreamsOnTasksAssignedCallbackCompletedEvent(
future,
Optional.empty()
);
- membershipManager.onTaskAssignmentCallbackCompleted(event);
+ membershipManager.onTasksAssignedCallbackCompleted(event);
assertTrue(future.isDone());
assertFalse(future.isCancelled());
@@ -645,15 +737,30 @@ public class StreamsMembershipManagerTest {
}
@Test
- public void testOnTaskAssignmentCallbackCompletedWhenCallbackFails() {
+ public void testOnAllTasksLostCallbackCompleted() {
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ final StreamsOnAllTasksLostCallbackCompletedEvent event = new StreamsOnAllTasksLostCallbackCompletedEvent(
+ future,
+ Optional.empty()
+ );
+
+ membershipManager.onAllTasksLostCallbackCompleted(event);
+
+ assertTrue(future.isDone());
+ assertFalse(future.isCancelled());
+ assertFalse(future.isCompletedExceptionally());
+ }
+
+ @Test
+ public void testOnTasksAssignedCallbackCompletedWhenCallbackFails() {
final String errorMessage = "KABOOM!";
final CompletableFuture<Void> future = new CompletableFuture<>();
- final StreamsOnAssignmentCallbackCompletedEvent event = new StreamsOnAssignmentCallbackCompletedEvent(
+ final StreamsOnAllTasksLostCallbackCompletedEvent event = new StreamsOnAllTasksLostCallbackCompletedEvent(
future,
Optional.of(new KafkaException(errorMessage))
);
- membershipManager.onTaskAssignmentCallbackCompleted(event);
+ membershipManager.onAllTasksLostCallbackCompleted(event);
assertTrue(future.isDone());
assertFalse(future.isCancelled());
@@ -781,43 +888,6 @@ public class StreamsMembershipManagerTest {
);
}
- private void setupStreamsAssignmentInterfaceWithTwoConcatenedSubtopologies(final String subtopologyId1,
- final String topicName1,
- final String subtopologyId2,
- final String topicName2) {
- when(streamsAssignmentInterface.subtopologyMap()).thenReturn(
- mkMap(
- mkEntry(
- subtopologyId1,
- new StreamsAssignmentInterface.Subtopology(
- Set.of(topicName1),
- Collections.emptySet(),
- Collections.emptyMap(),
- Collections.emptyMap(),
- Collections.emptyList()
- )
- ),
- mkEntry(
- subtopologyId2,
- new StreamsAssignmentInterface.Subtopology(
- Set.of(topicName2),
- Collections.emptySet(),
- mkMap(mkEntry(
- topicName2,
- new StreamsAssignmentInterface.TopicInfo(
- Optional.empty(),
- Optional.empty(),
- Collections.emptyMap()
- )
- )),
- Collections.emptyMap(),
- Collections.emptyList()
- )
- )
- )
- );
- }
-
private StreamsGroupHeartbeatResponse makeHeartbeatResponse(final String subtopologyId,
final List<Integer> partitions) {
return makeHeartbeatResponse(Collections.singletonList(
@@ -874,23 +944,26 @@ public class StreamsMembershipManagerTest {
}
private void acknowledging() {
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
+ final StreamsOnTasksAssignedCallbackNeededEvent onAssignmentCallbackNeededEvent =
+ (StreamsOnTasksAssignedCallbackNeededEvent) backgroundEventQueue.poll();
onAssignmentCallbackNeededEvent.future().complete(null);
verifyInStateAcknowledging(membershipManager);
}
+ private void acknowledging(final CompletableFuture<Void> future) {
+ future.complete(null);
+ verifyInStateAcknowledging(membershipManager);
+ }
+
private CompletableFuture<Void> prepareLeaving() {
final CompletableFuture<Void> onGroupLeft = membershipManager.leaveGroup();
verifyInStatePrepareLeaving(membershipManager);
return onGroupLeft;
}
- private CompletableFuture<Void> leaving() {
+ private CompletableFuture<Void> leaving(final CompletableFuture<Void> onAllTasksRevokedCallbackExecuted) {
final CompletableFuture<Void> future = prepareLeaving();
- final StreamsOnAssignmentCallbackNeededEvent onAssignmentCallbackNeededEvent =
- (StreamsOnAssignmentCallbackNeededEvent) backgroundEventQueue.poll();
- onAssignmentCallbackNeededEvent.future().complete(null);
+ onAllTasksRevokedCallbackExecuted.complete(null);
verifyInStateLeaving(membershipManager);
return future;
}
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 25e41a88257..511e602b9bf 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
@@ -31,6 +31,8 @@ import org.apache.kafka.clients.consumer.internals.MockRebalanceListener;
import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate;
import org.apache.kafka.clients.consumer.internals.OffsetsRequestManager;
import org.apache.kafka.clients.consumer.internals.RequestManagers;
+import org.apache.kafka.clients.consumer.internals.StreamsGroupHeartbeatRequestManager;
+import org.apache.kafka.clients.consumer.internals.StreamsMembershipManager;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.clients.consumer.internals.TopicMetadataRequestManager;
import org.apache.kafka.common.Cluster;
@@ -81,6 +83,8 @@ public class ApplicationEventProcessorTest {
private final CommitRequestManager commitRequestManager = mock(CommitRequestManager.class);
private final ConsumerHeartbeatRequestManager heartbeatRequestManager = mock(ConsumerHeartbeatRequestManager.class);
private final ConsumerMembershipManager membershipManager = mock(ConsumerMembershipManager.class);
+ private final StreamsGroupHeartbeatRequestManager streamsGroupHeartbeatRequestManager = mock(StreamsGroupHeartbeatRequestManager.class);
+ private final StreamsMembershipManager streamsMembershipManager = mock(StreamsMembershipManager.class);
private final OffsetsRequestManager offsetsRequestManager = mock(OffsetsRequestManager.class);
private SubscriptionState subscriptionState = mock(SubscriptionState.class);
private final ConsumerMetadata metadata = mock(ConsumerMetadata.class);
@@ -96,6 +100,7 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(commitRequestManager) : Optional.empty(),
withGroupId ? Optional.of(heartbeatRequestManager) : Optional.empty(),
withGroupId ? Optional.of(membershipManager) : Optional.empty(),
+ Optional.empty(),
Optional.empty()
);
@@ -107,6 +112,28 @@ public class ApplicationEventProcessorTest {
);
}
+ private void setupProcessorWithStreamsMembershipManager() {
+ RequestManagers requestManagers = new RequestManagers(
+ new LogContext(),
+ offsetsRequestManager,
+ mock(TopicMetadataRequestManager.class),
+ mock(FetchRequestManager.class),
+ Optional.of(mock(CoordinatorRequestManager.class)),
+ Optional.of(commitRequestManager),
+ Optional.empty(),
+ Optional.empty(),
+ Optional.of(streamsGroupHeartbeatRequestManager),
+ Optional.of(streamsMembershipManager)
+ );
+
+ processor = new ApplicationEventProcessor(
+ new LogContext(),
+ requestManagers,
+ metadata,
+ subscriptionState
+ );
+ }
+
@Test
public void testPrepClosingCommitEvents() {
setupProcessor(true);
@@ -125,6 +152,15 @@ public class ApplicationEventProcessorTest {
verify(membershipManager).leaveGroup();
}
+ @Test
+ public void testProcessUnsubscribeEventWithStreamsMembershipManager() {
+ setupProcessorWithStreamsMembershipManager();
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ when(streamsMembershipManager.leaveGroup()).thenReturn(CompletableFuture.completedFuture(null));
+ processor.process(new UnsubscribeEvent(0));
+ verify(streamsMembershipManager).leaveGroup();
+ }
+
@Test
public void testProcessUnsubscribeEventWithoutGroupId() {
setupProcessor(false);
@@ -501,6 +537,55 @@ public class ApplicationEventProcessorTest {
);
}
+ @Test
+ public void testSubscriptionChangeEventWithStreamsMembershipManager() {
+ Set<String> topics = Set.of("topic1", "topic2");
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicSubscriptionChangeEvent event = new TopicSubscriptionChangeEvent(topics, listener, 12345);
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verify(streamsMembershipManager).onSubscriptionUpdated();
+ // verify member state doesn't transition to JOINING.
+ verify(streamsMembershipManager, never()).onConsumerPoll();
+ }
+
+ @Test
+ public void testOnTasksRevokedCallbackExecuted() {
+ StreamsOnTasksRevokedCallbackCompletedEvent event = new StreamsOnTasksRevokedCallbackCompletedEvent(
+ new CompletableFuture<>(),
+ Optional.empty()
+ );
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verify(streamsMembershipManager).onTasksRevokedCallbackCompleted(event);
+ }
+
+ @Test
+ public void testOnTasksAssignedCallbackExecuted() {
+ StreamsOnTasksAssignedCallbackCompletedEvent event = new StreamsOnTasksAssignedCallbackCompletedEvent(
+ new CompletableFuture<>(),
+ Optional.empty()
+ );
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verify(streamsMembershipManager).onTasksAssignedCallbackCompleted(event);
+ }
+
+ @Test
+ public void testOnAllTasksLostCallbackExecuted() {
+ StreamsOnAllTasksLostCallbackCompletedEvent event = new StreamsOnAllTasksLostCallbackCompletedEvent(
+ new CompletableFuture<>(),
+ Optional.empty()
+ );
+
+ setupProcessorWithStreamsMembershipManager();
+ processor.process(event);
+ verify(streamsMembershipManager).onAllTasksLostCallbackCompleted(event);
+ }
+
private List<NetworkClientDelegate.UnsentRequest> mockCommitResults() {
return Collections.singletonList(mock(NetworkClientDelegate.UnsentRequest.class));
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index 4270e0d80c9..4af85ad97c9 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -833,7 +833,7 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
* @param record The record to apply to the state machine.
* @throws RuntimeException
*/
- @SuppressWarnings("CyclomaticComplexity")
+ @SuppressWarnings({"CyclomaticComplexity", "MethodLength"})
@Override
public void replay(
long offset,
@@ -954,48 +954,55 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
break;
case 16:
+ groupMetadataManager.replay(
+ (ConsumerGroupRegularExpressionKey) key.message(),
+ (ConsumerGroupRegularExpressionValue) Utils.messageOrNull(value)
+ );
+ break;
+
+ case 17:
groupMetadataManager.replay(
(StreamsGroupMetadataKey) key.message(),
(StreamsGroupMetadataValue) messageOrNull(value)
);
break;
- case 17:
+ case 18:
groupMetadataManager.replay(
(StreamsGroupPartitionMetadataKey) key.message(),
(StreamsGroupPartitionMetadataValue) messageOrNull(value)
);
break;
- case 18:
+ case 19:
groupMetadataManager.replay(
(StreamsGroupMemberMetadataKey) key.message(),
(StreamsGroupMemberMetadataValue) messageOrNull(value)
);
break;
- case 19:
+ case 20:
groupMetadataManager.replay(
(StreamsGroupTargetAssignmentMetadataKey) key.message(),
(StreamsGroupTargetAssignmentMetadataValue) messageOrNull(value)
);
break;
- case 20:
+ case 21:
groupMetadataManager.replay(
(StreamsGroupTargetAssignmentMemberKey) key.message(),
(StreamsGroupTargetAssignmentMemberValue) messageOrNull(value)
);
break;
- case 21:
+ case 22:
groupMetadataManager.replay(
(StreamsGroupCurrentMemberAssignmentKey) key.message(),
(StreamsGroupCurrentMemberAssignmentValue) messageOrNull(value)
);
break;
- case 22:
+ case 23:
groupMetadataManager.replay(
(StreamsGroupTopologyKey) key.message(),
(StreamsGroupTopologyValue) messageOrNull(value)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
index 6e834eb2ba0..9a3236a16a4 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
@@ -58,7 +58,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupMemberMetadataKey()
.setGroupId(groupId)
.setMemberId(member.memberId()),
- (short) 17
+ (short) 19
),
new ApiMessageAndVersion(
new StreamsGroupMemberMetadataValue()
@@ -96,7 +96,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupMemberMetadataKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 17
+ (short) 19
),
null // Tombstone.
);
@@ -137,7 +137,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupPartitionMetadataKey()
.setGroupId(groupId),
- (short) 16
+ (short) 18
),
new ApiMessageAndVersion(
value,
@@ -159,7 +159,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupPartitionMetadataKey()
.setGroupId(groupId),
- (short) 16
+ (short) 18
),
null // Tombstone.
);
@@ -173,7 +173,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupMetadataKey()
.setGroupId(groupId),
- (short) 15
+ (short) 17
),
new ApiMessageAndVersion(
new StreamsGroupMetadataValue()
@@ -196,7 +196,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupMetadataKey()
.setGroupId(groupId),
- (short) 15
+ (short) 17
),
null // Tombstone.
);
@@ -239,7 +239,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupTargetAssignmentMemberKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 19
+ (short) 21
),
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMemberValue()
@@ -267,7 +267,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupTargetAssignmentMemberKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 19
+ (short) 21
),
null // Tombstone.
);
@@ -282,7 +282,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataKey()
.setGroupId(groupId),
- (short) 18
+ (short) 20
),
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataValue()
@@ -305,7 +305,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTargetAssignmentMetadataKey()
.setGroupId(groupId),
- (short) 18
+ (short) 20
),
null // Tombstone.
);
@@ -320,7 +320,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupCurrentMemberAssignmentKey()
.setGroupId(groupId)
.setMemberId(member.memberId()),
- (short) 20
+ (short) 22
),
new ApiMessageAndVersion(
new StreamsGroupCurrentMemberAssignmentValue()
@@ -354,7 +354,7 @@ public class CoordinatorStreamsRecordHelpers {
new StreamsGroupCurrentMemberAssignmentKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 20
+ (short) 22
),
null // Tombstone
);
@@ -396,7 +396,7 @@ public class CoordinatorStreamsRecordHelpers {
return new CoordinatorRecord(new ApiMessageAndVersion(
new StreamsGroupTopologyKey()
.setGroupId(groupId),
- (short) 21),
+ (short) 23),
new ApiMessageAndVersion(value, (short) 0));
}
@@ -466,7 +466,7 @@ public class CoordinatorStreamsRecordHelpers {
new ApiMessageAndVersion(
new StreamsGroupTopologyKey()
.setGroupId(groupId),
- (short) 21
+ (short) 23
),
null // Tombstone
);
diff --git a/group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionKey.json b/group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionKey.json
index 2cb82e4de10..3f761b694e4 100644
--- a/group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionKey.json
+++ b/group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionKey.json
@@ -20,9 +20,9 @@
"validVersions": "16",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "44",
+ { "name": "GroupId", "type": "string", "versions": "16",
"about": "The group id." },
- { "name": "RegularExpression", "type": "string", "versions": "44",
+ { "name": "RegularExpression", "type": "string", "versions": "16",
"about": "The regular expression." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
index a99e32754c6..5f199f7b00a 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupCurrentMemberAssignmentKey",
- "validVersions": "20",
+ "validVersions": "22",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "20",
+ { "name": "GroupId", "type": "string", "versions": "22",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "20",
+ { "name": "MemberId", "type": "string", "versions": "22",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
index 8e0f66deba7..7136d03c149 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupMemberMetadataKey",
- "validVersions": "17",
+ "validVersions": "19",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "17",
+ { "name": "GroupId", "type": "string", "versions": "19",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "17",
+ { "name": "MemberId", "type": "string", "versions": "19",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
index 6c54fb8bd14..d018df39dcf 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupMetadataKey",
- "validVersions": "15",
+ "validVersions": "17",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "15",
+ { "name": "GroupId", "type": "string", "versions": "17",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
index 618fbc78451..4ca6baf7b32 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupPartitionMetadataKey",
- "validVersions": "56",
+ "validVersions": "18",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "56",
+ { "name": "GroupId", "type": "string", "versions": "18",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
index f96f6f89e01..0da4f6684e9 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
@@ -17,12 +17,12 @@
{
"type": "data",
"name": "StreamsGroupTargetAssignmentMemberKey",
- "validVersions": "19",
+ "validVersions": "21",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "19",
+ { "name": "GroupId", "type": "string", "versions": "21",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "19",
+ { "name": "MemberId", "type": "string", "versions": "21",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
index 514885f8ad7..c8e81d84260 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupTargetAssignmentMetadataKey",
- "validVersions": "18",
+ "validVersions": "20",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "18",
+ { "name": "GroupId", "type": "string", "versions": "20",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
index 2e51c117bd7..4cfef5740a0 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
@@ -17,10 +17,10 @@
{
"type": "data",
"name": "StreamsGroupTopologyKey",
- "validVersions": "21",
+ "validVersions": "23",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "21",
+ { "name": "GroupId", "type": "string", "versions": "23",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
index 029c3b8025f..61a77bb5693 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
@@ -118,7 +118,7 @@ class CoordinatorStreamsRecordHelpersTest {
new ApiMessageAndVersion(
new StreamsGroupTopologyKey()
.setGroupId("group-id"),
- (short) 21),
+ (short) 23),
new ApiMessageAndVersion(
new StreamsGroupTopologyValue()
.setTopology(expectedTopology),
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 90e17194b6f..7f45577dca8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -26,7 +26,6 @@ import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface;
-import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.Assignment;
import org.apache.kafka.clients.consumer.internals.StreamsAssignmentInterface.Subtopology;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
@@ -718,6 +717,11 @@ public class StreamThread extends Thread implements ProcessingThread {
this.streamsUncaughtExceptionHandler = streamsUncaughtExceptionHandler;
this.cacheResizer = cacheResizer;
this.streamsAssignmentInterface = streamsAssignmentInterface;
+ if (streamsAssignmentInterface != null) {
+ streamsAssignmentInterface.setOnTasksRevokedCallback(this::onTasksRevoked);
+ streamsAssignmentInterface.setOnTasksAssignedCallback(this::onTasksAssigned);
+ streamsAssignmentInterface.setOnAllTasksLostCallback(this::onAllTasksLost);
+ }
this.streamsMetadataState = streamsMetadataState;
// The following sensors are created here but their references are not stored in this object, since within
@@ -1074,6 +1078,8 @@ public class StreamThread extends Thread implements ProcessingThread {
final long startMs = time.milliseconds();
now = startMs;
+ maybeHandleAssignmentFromStreamsRebalanceProtocol();
+
final long pollLatency;
taskManager.resumePollingForPartitionsWithAvailableSpace();
pollLatency = pollPhase();
@@ -1404,23 +1410,59 @@ public class StreamThread extends Thread implements ProcessingThread {
);
// Process assignment from Streams Rebalance Protocol
- final Assignment newAssignment = streamsAssignmentInterface.targetAssignment.get();
- if (!newAssignment.equals(streamsAssignmentInterface.reconciledAssignment.get())) {
-
- final Map<TaskId, Set<TopicPartition>> activeTasksWithPartitions =
- pairWithTopicPartitions(newAssignment.activeTasks.stream());
- final Map<TaskId, Set<TopicPartition>> standbyTasksWithPartitions =
- pairWithTopicPartitions(Stream.concat(newAssignment.standbyTasks.stream(), newAssignment.warmupTasks.stream()));
+ streamsAssignmentInterface.processStreamsRebalanceEvents();
+ }
+ }
- log.info("Processing new assignment {} from Streams Rebalance Protocol", newAssignment);
+ private Optional<Exception> onTasksRevoked(final Set<StreamsAssignmentInterface.TaskId> activeTasksToRevoke) {
+ try {
+ final Map<TaskId, Set<TopicPartition>> activeTasksToRevokeWithPartitions =
+ pairWithTopicPartitions(activeTasksToRevoke.stream());
+ final Set<TopicPartition> partitionsToRevoke = activeTasksToRevokeWithPartitions.values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
- taskManager.handleAssignment(
- activeTasksWithPartitions,
- standbyTasksWithPartitions
- );
- streamsAssignmentInterface.reconciledAssignment.set(newAssignment);
+ final long start = time.milliseconds();
+ try {
+ log.info("Revoking active tasks {}.", activeTasksToRevoke);
+ taskManager.handleRevocation(partitionsToRevoke);
+ } finally {
+ log.info("partition revocation took {} ms.", time.milliseconds() - start);
}
+ if (state() != State.PENDING_SHUTDOWN) {
+ setState(State.PARTITIONS_REVOKED);
+ }
+ } catch (final Exception exception) {
+ return Optional.of(exception);
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Exception> onTasksAssigned(final StreamsAssignmentInterface.Assignment assignment) {
+ try {
+ final Map<TaskId, Set<TopicPartition>> activeTasksWithPartitions =
+ pairWithTopicPartitions(assignment.activeTasks.stream());
+ final Map<TaskId, Set<TopicPartition>> standbyTasksWithPartitions =
+ pairWithTopicPartitions(Stream.concat(assignment.standbyTasks.stream(), assignment.warmupTasks.stream()));
+
+ log.info("Processing new assignment {} from Streams Rebalance Protocol", assignment);
+
+ taskManager.handleAssignment(activeTasksWithPartitions, standbyTasksWithPartitions);
+ setState(State.PARTITIONS_ASSIGNED);
+ taskManager.handleRebalanceComplete();
+ } catch (final Exception exception) {
+ return Optional.of(exception);
+ }
+ return Optional.empty();
+ }
+
+ private Optional<Exception> onAllTasksLost() {
+ try {
+ taskManager.handleLostAll();
+ } catch (final Exception exception) {
+ return Optional.of(exception);
}
+ return Optional.empty();
}
static Map<TopicPartition, PartitionInfo> getTopicPartitionInfo(final Map<HostInfo, Set<TopicPartition>> partitionsByHost) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java
index 49428724b1a..a2f504cab0f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java
@@ -52,8 +52,6 @@ public class StreamsRebalanceListener implements ConsumerRebalanceListener {
@Override
public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
- streamThread.maybeHandleAssignmentFromStreamsRebalanceProtocol();
-
// NB: all task management is already handled by:
// org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment
if (assignmentErrorCode.get() == AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()) {