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 2025/04/10 19:06:12 UTC
(kafka) branch trunk updated: KAFKA-19124: Use consumer background event queue for Streams events (#19421)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new c11938c9263 KAFKA-19124: Use consumer background event queue for Streams events (#19421)
c11938c9263 is described below
commit c11938c926317c68df58da80ffd14551d3230271
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Thu Apr 10 21:06:06 2025 +0200
KAFKA-19124: Use consumer background event queue for Streams events (#19421)
In the first version of the integration of the stream thread with the
new Streams rebalance protocol, the consumer used a dedicated event
queue for Streams/specific background events to request the stream
thread to call the rebalance callbacks. That led to an issue where the
consumer times out when unsubscribing.
This commit gets rid of the dedicated queue and incorporates the
Streams-specific background events into event queue used by the
consumer.
Reviewers: Lucas Brutschy <lb...@confluent.io>
---
.../consumer/internals/AsyncKafkaConsumer.java | 134 ++-
.../internals/ConsumerDelegateCreator.java | 2 +-
.../consumer/internals/RequestManagers.java | 7 +-
.../internals/StreamsMembershipManager.java | 63 +-
.../internals/StreamsRebalanceEventsProcessor.java | 206 -----
...allbacks.java => StreamsRebalanceListener.java} | 4 +-
.../consumer/internals/AsyncKafkaConsumerTest.java | 3 -
.../consumer/internals/RequestManagersTest.java | 1 -
.../internals/StreamsMembershipManagerTest.java | 939 +++++++++++----------
.../StreamsRebalanceEventsProcessorTest.java | 286 -------
.../SmokeTestDriverIntegrationTest.java | 2 +
...s.java => DefaultStreamsRebalanceListener.java} | 14 +-
.../streams/processor/internals/StreamThread.java | 58 +-
.../processor/internals/StreamThreadTest.java | 10 -
14 files changed, 675 insertions(+), 1054 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 e421ef93191..249904f87b0 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
@@ -64,6 +64,12 @@ import org.apache.kafka.clients.consumer.internals.events.ResetOffsetEvent;
import org.apache.kafka.clients.consumer.internals.events.ResumePartitionsEvent;
import org.apache.kafka.clients.consumer.internals.events.SeekUnvalidatedEvent;
import org.apache.kafka.clients.consumer.internals.events.StopFindCoordinatorOnCloseEvent;
+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.clients.consumer.internals.events.SyncCommitEvent;
import org.apache.kafka.clients.consumer.internals.events.TopicMetadataEvent;
import org.apache.kafka.clients.consumer.internals.events.TopicPatternSubscriptionChangeEvent;
@@ -180,6 +186,25 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
*/
private class BackgroundEventProcessor implements EventProcessor<BackgroundEvent> {
+ private Optional<StreamsRebalanceListener> streamsGroupRebalanceCallbacks = Optional.empty();
+ private final Optional<StreamsRebalanceData> streamsRebalanceData;
+
+ public BackgroundEventProcessor() {
+ this.streamsRebalanceData = Optional.empty();
+ }
+
+ public BackgroundEventProcessor(final Optional<StreamsRebalanceData> streamsRebalanceData) {
+ this.streamsRebalanceData = streamsRebalanceData;
+ }
+
+ private void setStreamsRebalanceListener(final StreamsRebalanceListener streamsRebalanceListener) {
+ if (streamsRebalanceData.isEmpty()) {
+ throw new IllegalStateException("Background event processor was not created to be used with Streams " +
+ "rebalance protocol events");
+ }
+ this.streamsGroupRebalanceCallbacks = Optional.of(streamsRebalanceListener);
+ }
+
@Override
public void process(final BackgroundEvent event) {
switch (event.type()) {
@@ -191,6 +216,18 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
process((ConsumerRebalanceListenerCallbackNeededEvent) event);
break;
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
+ processStreamsOnTasksRevokedCallbackNeededEvent((StreamsOnTasksRevokedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
+ processStreamsOnTasksAssignedCallbackNeededEvent((StreamsOnTasksAssignedCallbackNeededEvent) event);
+ break;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
+ processStreamsOnAllTasksLostCallbackNeededEvent((StreamsOnAllTasksLostCallbackNeededEvent) event);
+ break;
+
default:
throw new IllegalArgumentException("Background event type " + event.type() + " was not expected");
@@ -213,6 +250,79 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
throw invokedEvent.error().get();
}
}
+
+ private void processStreamsOnTasksRevokedCallbackNeededEvent(final StreamsOnTasksRevokedCallbackNeededEvent event) {
+ StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent = invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void processStreamsOnTasksAssignedCallbackNeededEvent(final StreamsOnTasksAssignedCallbackNeededEvent event) {
+ StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent = invokeOnTasksAssignedCallback(event.assignment(), event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private void processStreamsOnAllTasksLostCallbackNeededEvent(final StreamsOnAllTasksLostCallbackNeededEvent event) {
+ StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent = invokeOnAllTasksLostCallback(event.future());
+ applicationEventHandler.add(invokedEvent);
+ if (invokedEvent.error().isPresent()) {
+ throw invokedEvent.error().get();
+ }
+ }
+
+ private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke,
+ final CompletableFuture<Void> future) {
+ final Optional<KafkaException> error;
+ final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onTasksRevoked(activeTasksToRevoke);
+ if (exceptionFromCallback.isPresent()) {
+ error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task revocation callback throws an error"));
+ } else {
+ error = Optional.empty();
+ }
+ return new StreamsOnTasksRevokedCallbackCompletedEvent(future, error);
+ }
+
+ private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallback(final StreamsRebalanceData.Assignment assignment,
+ final CompletableFuture<Void> future) {
+ final Optional<KafkaException> error;
+ final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onTasksAssigned(assignment);
+ if (exceptionFromCallback.isPresent()) {
+ error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
+ } else {
+ error = Optional.empty();
+ streamsRebalanceData().setReconciledAssignment(assignment);
+ }
+ return new StreamsOnTasksAssignedCallbackCompletedEvent(future, error);
+ }
+
+ private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
+ final Optional<KafkaException> error;
+ final Optional<Exception> exceptionFromCallback = streamsGroupRebalanceCallbacks().onAllTasksLost();
+ if (exceptionFromCallback.isPresent()) {
+ error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "All tasks lost callback throws an error"));
+ } else {
+ error = Optional.empty();
+ streamsRebalanceData().setReconciledAssignment(StreamsRebalanceData.Assignment.EMPTY);
+ }
+ return new StreamsOnAllTasksLostCallbackCompletedEvent(future, error);
+ }
+
+ private StreamsRebalanceData streamsRebalanceData() {
+ return streamsRebalanceData.orElseThrow(
+ () -> new IllegalStateException("Background event processor was not created to be used with Streams " +
+ "rebalance protocol events"));
+ }
+
+ private StreamsRebalanceListener streamsGroupRebalanceCallbacks() {
+ return streamsGroupRebalanceCallbacks.orElseThrow(
+ () -> new IllegalStateException("Background event processor was not created to be used with Streams " +
+ "rebalance protocol events"));
+ }
}
private final ApplicationEventHandler applicationEventHandler;
@@ -284,8 +394,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
public AsyncKafkaConsumer(final ConsumerConfig config,
final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer,
- final Optional<StreamsRebalanceData> streamsRebalanceData,
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor) {
+ final Optional<StreamsRebalanceData> streamsRebalanceData) {
this(
config,
keyDeserializer,
@@ -296,8 +405,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
FetchCollector::new,
ConsumerMetadata::new,
new LinkedBlockingQueue<>(),
- streamsRebalanceData,
- streamsRebalanceEventsProcessor
+ streamsRebalanceData
);
}
@@ -312,8 +420,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
final FetchCollectorFactory<K, V> fetchCollectorFactory,
final ConsumerMetadataFactory metadataFactory,
final LinkedBlockingQueue<BackgroundEvent> backgroundEventQueue,
- final Optional<StreamsRebalanceData> streamsRebalanceData,
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor) {
+ final Optional<StreamsRebalanceData> streamsRebalanceData) {
try {
GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
config,
@@ -390,8 +497,7 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
metrics,
offsetCommitCallbackInvoker,
memberStateListener,
- streamsRebalanceData,
- streamsRebalanceEventsProcessor
+ streamsRebalanceData
);
final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(logContext,
metadata,
@@ -407,17 +513,13 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
requestManagersSupplier,
kafkaConsumerMetrics
);
- streamsRebalanceEventsProcessor.ifPresent(
- processor -> processor.setApplicationEventHandler(applicationEventHandler)
- );
-
this.rebalanceListenerInvoker = new ConsumerRebalanceListenerInvoker(
logContext,
subscriptions,
time,
new RebalanceCallbackMetricsManager(metrics)
);
- this.backgroundEventProcessor = new BackgroundEventProcessor();
+ this.backgroundEventProcessor = new BackgroundEventProcessor(streamsRebalanceData);
this.backgroundEventReaper = backgroundEventReaperFactory.build(logContext);
// The FetchCollector is only used on the application thread.
@@ -581,7 +683,6 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
metrics,
offsetCommitCallbackInvoker,
memberStateListener,
- Optional.empty(),
Optional.empty()
);
Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(
@@ -1857,6 +1958,11 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
subscribeInternal(topics, Optional.of(listener));
}
+ public void subscribe(Collection<String> topics, StreamsRebalanceListener streamsRebalanceListener) {
+ subscribeInternal(topics, Optional.empty());
+ backgroundEventProcessor.setStreamsRebalanceListener(streamsRebalanceListener);
+ }
+
@Override
public void subscribe(Pattern pattern) {
subscribeInternal(pattern, Optional.empty());
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 60d12aec864..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
@@ -61,7 +61,7 @@ public class ConsumerDelegateCreator {
GroupProtocol groupProtocol = GroupProtocol.valueOf(config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));
if (groupProtocol == GroupProtocol.CONSUMER)
- return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty(), Optional.empty());
+ return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty());
else
return new ClassicKafkaConsumer<>(config, keyDeserializer, valueDeserializer);
} catch (KafkaException e) {
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 9f406bd2090..cab7d804cad 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
@@ -169,8 +169,7 @@ public class RequestManagers implements Closeable {
final Metrics metrics,
final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker,
final MemberStateListener applicationThreadMemberStateListener,
- final Optional<StreamsRebalanceData> streamsRebalanceData,
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor
+ final Optional<StreamsRebalanceData> streamsRebalanceData
) {
return new CachedSupplier<>() {
@Override
@@ -220,12 +219,12 @@ public class RequestManagers implements Closeable {
groupRebalanceConfig.groupInstanceId,
metrics,
metadata);
- if (streamsRebalanceEventsProcessor.isPresent()) {
+ if (streamsRebalanceData.isPresent()) {
streamsMembershipManager = new StreamsMembershipManager(
groupRebalanceConfig.groupId,
- streamsRebalanceEventsProcessor.get(),
streamsRebalanceData.get(),
subscriptions,
+ backgroundEventHandler,
logContext,
time,
metrics);
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 a5684bea83a..eb006ac8dda 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,13 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
import org.apache.kafka.clients.consumer.internals.events.StreamsOnAllTasksLostCallbackCompletedEvent;
+import org.apache.kafka.clients.consumer.internals.events.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.clients.consumer.internals.metrics.ConsumerRebalanceMetricsManager;
import org.apache.kafka.clients.consumer.internals.metrics.RebalanceMetricsManager;
import org.apache.kafka.common.KafkaException;
@@ -145,10 +149,10 @@ public class StreamsMembershipManager implements RequestManager {
private final Logger log;
/**
- * The processor that handles events of the Streams rebalance protocol.
+ * The processor that handles events from the background thread (a.k.a. ConsumerNetworkThread).
* For example, requests for invocation of assignment/revocation callbacks.
*/
- private final StreamsRebalanceEventsProcessor streamsRebalanceEventsProcessor;
+ private final BackgroundEventHandler backgroundEventHandler;
/**
* Data needed to participate in the Streams rebalance protocol.
@@ -269,26 +273,25 @@ public class StreamsMembershipManager implements RequestManager {
/**
* Constructs the Streams membership manager.
*
- * @param groupId The ID of the group.
- * @param streamsRebalanceEventsProcessor The processor that handles Streams rebalance events like requests for
- * invocation of assignment/revocation callbacks.
- * @param streamsRebalanceData Data needed to participate in the Streams rebalance protocol.
- * @param subscriptionState The subscription state of the member.
- * @param logContext The log context.
- * @param time The time.
- * @param metrics The metrics.
+ * @param groupId The ID of the group.
+ * @param streamsRebalanceData Data needed to participate in the Streams rebalance protocol.
+ * @param subscriptionState The subscription state of the member.
+ * @param backgroundEventHandler The handler that handles events from the background thread.
+ * @param logContext The log context.
+ * @param time The time.
+ * @param metrics The metrics.
*/
public StreamsMembershipManager(final String groupId,
- final StreamsRebalanceEventsProcessor streamsRebalanceEventsProcessor,
final StreamsRebalanceData streamsRebalanceData,
final SubscriptionState subscriptionState,
+ final BackgroundEventHandler backgroundEventHandler,
final LogContext logContext,
final Time time,
final Metrics metrics) {
log = logContext.logger(StreamsMembershipManager.class);
this.state = MemberState.UNSUBSCRIBED;
this.groupId = groupId;
- this.streamsRebalanceEventsProcessor = streamsRebalanceEventsProcessor;
+ this.backgroundEventHandler = backgroundEventHandler;
this.streamsRebalanceData = streamsRebalanceData;
this.subscriptionState = subscriptionState;
metricsManager = new ConsumerRebalanceMetricsManager(metrics);
@@ -448,9 +451,8 @@ public class StreamsMembershipManager implements RequestManager {
private void transitionToStale() {
transitionTo(MemberState.STALE);
- final CompletableFuture<Void> onAllTasksLostCallbackExecution =
- streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
- staleMemberAssignmentRelease = onAllTasksLostCallbackExecution.whenComplete((result, error) -> {
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted = requestOnAllTasksLostCallbackInvocation();
+ staleMemberAssignmentRelease = onAllTasksLostCallbackExecuted.whenComplete((result, error) -> {
if (error != null) {
log.error("Task revocation callback invocation failed " +
"after member left group due to expired poll timer.", error);
@@ -487,7 +489,7 @@ public class StreamsMembershipManager implements RequestManager {
return;
}
- CompletableFuture<Void> onAllTasksLostCallbackExecuted = streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
+ CompletableFuture<Void> onAllTasksLostCallbackExecuted = requestOnAllTasksLostCallbackInvocation();
onAllTasksLostCallbackExecuted.whenComplete((result, error) -> {
if (error != null) {
log.error("onAllTasksLost callback invocation failed while releasing assignment " +
@@ -763,8 +765,8 @@ public class StreamsMembershipManager implements RequestManager {
log.debug("Member {} with epoch {} transitioned to {} state. It will release its " +
"assignment and rejoin the group.", memberId, memberEpoch, MemberState.FENCED);
- CompletableFuture<Void> callbackResult = streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
- callbackResult.whenComplete((result, error) -> {
+ CompletableFuture<Void> onAllTasksLostCallbackExecuted = requestOnAllTasksLostCallbackInvocation();
+ onAllTasksLostCallbackExecuted.whenComplete((result, error) -> {
if (error != null) {
log.error("onAllTasksLost callback invocation failed while releasing assignment" +
" after member got fenced. Member will rejoin the group anyways.", error);
@@ -1090,8 +1092,7 @@ public class StreamsMembershipManager implements RequestManager {
subscriptionState.markPendingRevocation(partitionsToRevoke);
CompletableFuture<Void> tasksRevoked = new CompletableFuture<>();
- CompletableFuture<Void> onTasksRevokedCallbackExecuted =
- streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksToRevoke);
+ CompletableFuture<Void> onTasksRevokedCallbackExecuted = requestOnTasksRevokedCallbackInvocation(activeTasksToRevoke);
onTasksRevokedCallbackExecuted.whenComplete((__, callbackError) -> {
if (callbackError != null) {
log.error("onTasksRevoked callback invocation failed for tasks {}",
@@ -1131,7 +1132,7 @@ public class StreamsMembershipManager implements RequestManager {
notifyAssignmentChange(partitionsToAssign);
CompletableFuture<Void> onTasksAssignedCallbackExecuted =
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
+ requestOnTasksAssignedCallbackInvocation(
new StreamsRebalanceData.Assignment(
activeTasksToAssign,
standbyTasksToAssign,
@@ -1163,7 +1164,7 @@ public class StreamsMembershipManager implements RequestManager {
log.debug("Marking lost partitions pending for revocation: {}", partitionsToRelease);
subscriptionState.markPendingRevocation(partitionsToRelease);
- return streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
+ return requestOnAllTasksLostCallbackInvocation();
}
private SortedSet<TopicPartition> partitionsToAssignNotPreviouslyOwned(final SortedSet<TopicPartition> assignedTopicPartitions,
@@ -1224,6 +1225,24 @@ public class StreamsMembershipManager implements RequestManager {
rejoinedWhileReconciliationInProgress = false;
}
+ private CompletableFuture<Void> requestOnTasksAssignedCallbackInvocation(final StreamsRebalanceData.Assignment assignment) {
+ final StreamsOnTasksAssignedCallbackNeededEvent onTasksAssignedCallbackNeededEvent = new StreamsOnTasksAssignedCallbackNeededEvent(assignment);
+ backgroundEventHandler.add(onTasksAssignedCallbackNeededEvent);
+ return onTasksAssignedCallbackNeededEvent.future();
+ }
+
+ private CompletableFuture<Void> requestOnAllTasksLostCallbackInvocation() {
+ final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = new StreamsOnAllTasksLostCallbackNeededEvent();
+ backgroundEventHandler.add(onAllTasksLostCallbackNeededEvent);
+ return onAllTasksLostCallbackNeededEvent.future();
+ }
+
+ public CompletableFuture<Void> requestOnTasksRevokedCallbackInvocation(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke) {
+ final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = new StreamsOnTasksRevokedCallbackNeededEvent(activeTasksToRevoke);
+ backgroundEventHandler.add(onTasksRevokedCallbackNeededEvent);
+ return onTasksRevokedCallbackNeededEvent.future();
+ }
+
/**
* Completes the future that marks the completed execution of the onTasksRevoked callback.
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java
deleted file mode 100644
index 30bc7dcfb07..00000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessor.java
+++ /dev/null
@@ -1,206 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-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.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 java.util.LinkedList;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * Processes events from the Streams rebalance protocol.
- * <p>
- * The Streams rebalance processor receives events from the background thread of the async consumer, more precisely
- * from the Streams membership manager and handles them.
- * For example, events are requests for invoking the task assignment and task revocation callbacks.
- * Results of the event handling are passed back to the background thread.
- */
-public class StreamsRebalanceEventsProcessor {
-
- private final BlockingQueue<BackgroundEvent> onCallbackRequests = new LinkedBlockingQueue<>();
- private final StreamsRebalanceData streamsRebalanceData;
- private ApplicationEventHandler applicationEventHandler;
- private StreamsGroupRebalanceCallbacks rebalanceCallbacks;
-
- /**
- * Constructs the Streams rebalance processor.
- *
- * @param streamsRebalanceData
- */
- public StreamsRebalanceEventsProcessor(StreamsRebalanceData streamsRebalanceData) {
- this.streamsRebalanceData = streamsRebalanceData;
- }
-
- /**
- * Requests the invocation of the task assignment callback.
- *
- * @param assignment The tasks to be assigned to the member of the Streams group.
- * @return A future that will be completed when the callback has been invoked.
- */
- public CompletableFuture<Void> requestOnTasksAssignedCallbackInvocation(final StreamsRebalanceData.Assignment assignment) {
- final StreamsOnTasksAssignedCallbackNeededEvent onTasksAssignedCallbackNeededEvent = new StreamsOnTasksAssignedCallbackNeededEvent(assignment);
- onCallbackRequests.add(onTasksAssignedCallbackNeededEvent);
- return onTasksAssignedCallbackNeededEvent.future();
- }
-
- /**
- * Requests the invocation of the task revocation callback.
- *
- * @param activeTasksToRevoke The tasks to revoke from the member of the Streams group
- * @return A future that will be completed when the callback has been invoked.
- */
- public CompletableFuture<Void> requestOnTasksRevokedCallbackInvocation(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke) {
- final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = new StreamsOnTasksRevokedCallbackNeededEvent(activeTasksToRevoke);
- onCallbackRequests.add(onTasksRevokedCallbackNeededEvent);
- return onTasksRevokedCallbackNeededEvent.future();
- }
-
- /**
- * Requests the invocation of the all tasks lost callback.
- *
- * @return A future that will be completed when the callback has been invoked.
- */
- public CompletableFuture<Void> requestOnAllTasksLostCallbackInvocation() {
- final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = new StreamsOnAllTasksLostCallbackNeededEvent();
- onCallbackRequests.add(onAllTasksLostCallbackNeededEvent);
- return onAllTasksLostCallbackNeededEvent.future();
- }
-
- /**
- * Sets the application event handler.
- *
- * The application handler sends the results of the callbacks to the background thread.
- *
- * @param applicationEventHandler The application handler.
- */
- public void setApplicationEventHandler(final ApplicationEventHandler applicationEventHandler) {
- this.applicationEventHandler = applicationEventHandler;
- }
-
- public void setRebalanceCallbacks(final StreamsGroupRebalanceCallbacks rebalanceCallbacks) {
- this.rebalanceCallbacks = rebalanceCallbacks;
- }
-
- private void process(final BackgroundEvent event) {
- switch (event.type()) {
- case ERROR:
- throw ((ErrorEvent) event).error();
-
- case STREAMS_ON_TASKS_REVOKED_CALLBACK_NEEDED:
- processStreamsOnTasksRevokedCallbackNeededEvent((StreamsOnTasksRevokedCallbackNeededEvent) event);
- break;
-
- case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_NEEDED:
- processStreamsOnTasksAssignedCallbackNeededEvent((StreamsOnTasksAssignedCallbackNeededEvent) event);
- break;
-
- case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_NEEDED:
- processStreamsOnAllTasksLostCallbackNeededEvent((StreamsOnAllTasksLostCallbackNeededEvent) event);
- break;
-
- default:
- throw new IllegalArgumentException("Background event type " + event.type() + " was not expected");
-
- }
- }
-
- private void processStreamsOnTasksRevokedCallbackNeededEvent(final StreamsOnTasksRevokedCallbackNeededEvent event) {
- StreamsOnTasksRevokedCallbackCompletedEvent invokedEvent = invokeOnTasksRevokedCallback(event.activeTasksToRevoke(), event.future());
- applicationEventHandler.add(invokedEvent);
- if (invokedEvent.error().isPresent()) {
- throw invokedEvent.error().get();
- }
- }
-
- private void processStreamsOnTasksAssignedCallbackNeededEvent(final StreamsOnTasksAssignedCallbackNeededEvent event) {
- StreamsOnTasksAssignedCallbackCompletedEvent invokedEvent = invokeOnTasksAssignedCallback(event.assignment(), event.future());
- applicationEventHandler.add(invokedEvent);
- if (invokedEvent.error().isPresent()) {
- throw invokedEvent.error().get();
- }
- }
-
- private void processStreamsOnAllTasksLostCallbackNeededEvent(final StreamsOnAllTasksLostCallbackNeededEvent event) {
- StreamsOnAllTasksLostCallbackCompletedEvent invokedEvent = invokeOnAllTasksLostCallback(event.future());
- applicationEventHandler.add(invokedEvent);
- if (invokedEvent.error().isPresent()) {
- throw invokedEvent.error().get();
- }
- }
-
- private StreamsOnTasksRevokedCallbackCompletedEvent invokeOnTasksRevokedCallback(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke,
- final CompletableFuture<Void> future) {
- final Optional<KafkaException> error;
- final Optional<Exception> exceptionFromCallback = rebalanceCallbacks.onTasksRevoked(activeTasksToRevoke);
- if (exceptionFromCallback.isPresent()) {
- error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task revocation callback throws an error"));
- } else {
- error = Optional.empty();
- }
- return new StreamsOnTasksRevokedCallbackCompletedEvent(future, error);
- }
-
- private StreamsOnTasksAssignedCallbackCompletedEvent invokeOnTasksAssignedCallback(final StreamsRebalanceData.Assignment assignment,
- final CompletableFuture<Void> future) {
- final Optional<KafkaException> error;
- final Optional<Exception> exceptionFromCallback = rebalanceCallbacks.onTasksAssigned(assignment);
- if (exceptionFromCallback.isPresent()) {
- error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "Task assignment callback throws an error"));
- } else {
- error = Optional.empty();
- streamsRebalanceData.setReconciledAssignment(assignment);
- }
- return new StreamsOnTasksAssignedCallbackCompletedEvent(future, error);
- }
-
- private StreamsOnAllTasksLostCallbackCompletedEvent invokeOnAllTasksLostCallback(final CompletableFuture<Void> future) {
- final Optional<KafkaException> error;
- final Optional<Exception> exceptionFromCallback = rebalanceCallbacks.onAllTasksLost();
- if (exceptionFromCallback.isPresent()) {
- error = Optional.of(ConsumerUtils.maybeWrapAsKafkaException(exceptionFromCallback.get(), "All tasks lost callback throws an error"));
- } else {
- error = Optional.empty();
- streamsRebalanceData.setReconciledAssignment(StreamsRebalanceData.Assignment.EMPTY);
- }
- return new StreamsOnAllTasksLostCallbackCompletedEvent(future, error);
- }
-
- /**
- * Processes all events received from the background thread so far.
- */
- public void process() {
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- onCallbackRequests.drainTo(events);
- for (BackgroundEvent event : events) {
- process(event);
- }
- }
-
-}
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupRebalanceCallbacks.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
similarity index 93%
rename from clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupRebalanceCallbacks.java
rename to clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
index 4840e79eceb..b8a5db79445 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupRebalanceCallbacks.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceListener.java
@@ -20,9 +20,9 @@ import java.util.Optional;
import java.util.Set;
/**
- * Callbacks for handling Streams group rebalance events in Kafka Streams.
+ * Listener for handling Streams group rebalance events in Kafka Streams.
*/
-public interface StreamsGroupRebalanceCallbacks {
+public interface StreamsRebalanceListener {
/**
* Called when tasks are revoked from a stream thread.
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index df091bfccb6..074fe7ef4a7 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -220,7 +220,6 @@ public class AsyncKafkaConsumerTest {
(a, b, c, d, e, f, g) -> fetchCollector,
(a, b, c, d) -> metadata,
backgroundEventQueue,
- Optional.empty(),
Optional.empty()
);
}
@@ -236,7 +235,6 @@ public class AsyncKafkaConsumerTest {
(a, b, c, d, e, f, g) -> fetchCollector,
(a, b, c, d) -> metadata,
backgroundEventQueue,
- Optional.empty(),
Optional.empty()
);
}
@@ -1320,7 +1318,6 @@ public class AsyncKafkaConsumerTest {
any(),
any(),
applicationThreadMemberStateListener.capture(),
- any(),
any()
));
return applicationThreadMemberStateListener.getValue();
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
index ef3ba79dbf5..499c7f8f39d 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
@@ -63,7 +63,6 @@ public class RequestManagersTest {
new Metrics(),
mock(OffsetCommitCallbackInvoker.class),
listener,
- Optional.empty(),
Optional.empty()
).get();
requestManagers.consumerMembershipManager.ifPresent(
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 896a76ac209..dd8e232fc46 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
@@ -16,9 +16,13 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
import org.apache.kafka.clients.consumer.internals.events.StreamsOnAllTasksLostCallbackCompletedEvent;
+import org.apache.kafka.clients.consumer.internals.events.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.MetricName;
import org.apache.kafka.common.TopicPartition;
@@ -36,6 +40,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -93,7 +99,7 @@ public class StreamsMembershipManagerTest {
private SubscriptionState subscriptionState;
@Mock
- private StreamsRebalanceEventsProcessor streamsRebalanceEventsProcessor;
+ private BackgroundEventHandler backgroundEventHandler;
@Mock
private StreamsRebalanceData streamsRebalanceData;
@@ -101,13 +107,21 @@ public class StreamsMembershipManagerTest {
@Mock
private MemberStateListener memberStateListener;
+ @Captor
+ private ArgumentCaptor<StreamsOnTasksAssignedCallbackNeededEvent> onTasksAssignedCallbackNeededEventCaptor;
+ private int onTasksAssignedCallbackNeededAddCount = 0;
+
+ @Captor
+ private ArgumentCaptor<StreamsOnTasksRevokedCallbackNeededEvent> onTasksRevokedCallbackNeededEventCaptor;
+
+ @Captor
+ private ArgumentCaptor<StreamsOnAllTasksLostCallbackNeededEvent> onAllTasksLostCallbackNeededEventCaptor;
+
@BeforeEach
public void setup() {
membershipManager = new StreamsMembershipManager(
GROUP_ID,
- streamsRebalanceEventsProcessor,
- streamsRebalanceData,
- subscriptionState,
+ streamsRebalanceData, subscriptionState, backgroundEventHandler,
new LogContext("test"),
time,
metrics
@@ -179,18 +193,17 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingEmptyToSingleActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
@@ -199,31 +212,26 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingActiveTaskToDifferentActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
@@ -233,6 +241,8 @@ public class StreamsMembershipManagerTest {
expectedNewPartitionsToAssign
);
onTasksRevokedCallbackExecuted.complete(null);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
@@ -240,9 +250,7 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingSingleActiveTaskToAdditionalActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
@@ -250,19 +258,19 @@ public class StreamsMembershipManagerTest {
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(
new TopicPartition(TOPIC_0, PARTITION_0),
new TopicPartition(TOPIC_0, PARTITION_1)
@@ -276,10 +284,7 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingMultipleActiveTaskToSingleActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
@@ -290,24 +295,22 @@ public class StreamsMembershipManagerTest {
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksToRevoke))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0), new TopicPartition(TOPIC_0, PARTITION_1)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksToRevoke);
final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
- final Set<TopicPartition> expectedNewPartitionsToAssign = Collections.emptySet();
+ final Set<TopicPartition> expectedNewPartitionsToAssign = Set.of();
verifyInStateReconcilingBeforeOnTaskRevokedCallbackExecuted(
expectedPartitionsToRevoke,
expectedFullPartitionsToAssign,
@@ -315,13 +318,15 @@ public class StreamsMembershipManagerTest {
);
onTasksRevokedCallbackExecuted.complete(null);
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
}
@Test
public void testReconcilingEmptyToMultipleActiveTaskOfDifferentSubtopologies() {
- setupStreamsAssignmentInterfaceWithTwoSubtopologies(
+ setupStreamsReabalanceDataWithTwoSubtopologies(
SUBTOPOLOGY_ID_0, TOPIC_0,
SUBTOPOLOGY_ID_1, TOPIC_1
);
@@ -329,12 +334,6 @@ public class StreamsMembershipManagerTest {
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_1, PARTITION_0)
);
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(
@@ -342,6 +341,8 @@ public class StreamsMembershipManagerTest {
SUBTOPOLOGY_ID_1, List.of(PARTITION_0))
);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(
new TopicPartition(TOPIC_0, PARTITION_0),
new TopicPartition(TOPIC_1, PARTITION_0)
@@ -355,38 +356,33 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingActiveTaskToStandbyTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet()))
- ).thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)))
- .thenReturn(Collections.emptySet());
+ .thenReturn(Set.of());
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskRevokedCallbackExecuted(
expectedPartitionsToRevoke,
@@ -395,44 +391,45 @@ public class StreamsMembershipManagerTest {
);
onTasksRevokedCallbackExecuted.complete(null);
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
}
@Test
public void testReconcilingActiveTaskToWarmupTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks))
- ).thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)))
- .thenReturn(Collections.emptySet());
+ .thenReturn(Set.of());
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> expectedPartitionsToRevoke = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskRevokedCallbackExecuted(
expectedPartitionsToRevoke,
@@ -441,6 +438,12 @@ public class StreamsMembershipManagerTest {
);
onTasksRevokedCallbackExecuted.complete(null);
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
}
@@ -449,17 +452,17 @@ public class StreamsMembershipManagerTest {
public void testReconcilingEmptyToSingleStandbyTask() {
final Set<StreamsRebalanceData.TaskId> standbyTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -469,33 +472,33 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingStandbyTaskToDifferentStandbyTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> standbyTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasksSetup, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasksSetup,
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -505,8 +508,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingSingleStandbyTaskToAdditionalStandbyTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> standbyTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
@@ -514,27 +515,29 @@ public class StreamsMembershipManagerTest {
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasksSetup, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasksSetup,
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
verifyThatNoTasksHaveBeenRevoked();
@@ -542,8 +545,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingMultipleStandbyTaskToSingleStandbyTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> standbyTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
@@ -551,27 +552,29 @@ public class StreamsMembershipManagerTest {
final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasksSetup, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasksSetup,
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
verifyThatNoTasksHaveBeenRevoked();
@@ -579,31 +582,25 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingStandbyTaskToActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> standbyTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasksSetup, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasksSetup,
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
@@ -611,6 +608,8 @@ public class StreamsMembershipManagerTest {
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
onTasksAssignedCallbackExecuted.complete(null);
verifyInStateAcknowledgingAfterOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign);
verifyThatNoTasksHaveBeenRevoked();
@@ -618,31 +617,33 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingStandbyTaskToWarmupTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> standbyTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasksSetup, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks))
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasksSetup,
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -654,17 +655,17 @@ public class StreamsMembershipManagerTest {
public void testReconcilingEmptyToSingleWarmupTask() {
final Set<StreamsRebalanceData.TaskId> warmupTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks)
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -674,33 +675,33 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingWarmupTaskToDifferentWarmupTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> warmupTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasksSetup)
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks)
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasksSetup
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -710,8 +711,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingSingleWarmupTaskToAdditionalWarmupTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> warmupTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
@@ -719,25 +718,27 @@ public class StreamsMembershipManagerTest {
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasksSetup)
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks)
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasksSetup
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -747,8 +748,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingMultipleWarmupTaskToSingleWarmupTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> warmupTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0),
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
@@ -756,25 +755,27 @@ public class StreamsMembershipManagerTest {
final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasksSetup)
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasks)
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0, PARTITION_1)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasksSetup
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasks
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -784,35 +785,35 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingWarmupTaskToActiveTask() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> warmupTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasksSetup)
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_1)));
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasksSetup
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
@@ -823,33 +824,33 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingWarmupTaskToStandbyTask() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
final Set<StreamsRebalanceData.TaskId> warmupTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), Collections.emptySet(), warmupTasksSetup)
- )
- ).thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(
- streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(Collections.emptySet(), standbyTasks, Collections.emptySet())
- )
- ).thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithWarmupTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ Set.of(),
+ warmupTasksSetup
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
Mockito.reset(subscriptionState);
Mockito.reset(memberStateListener);
reconcile(makeHeartbeatResponseWithStandbyTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
- final Set<TopicPartition> expectedFullPartitionsToAssign = Collections.emptySet();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ Set.of(),
+ standbyTasks,
+ Set.of()
+ );
+ final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of();
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
onTasksAssignedCallbackExecuted.complete(null);
@@ -859,16 +860,19 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingAndAssignmentCallbackFails() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedNewPartitionsToAssign = expectedFullPartitionsToAssign;
verifyInStateReconcilingBeforeOnTaskAssignedCallbackExecuted(expectedFullPartitionsToAssign, expectedNewPartitionsToAssign);
@@ -881,28 +885,30 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingAndRevocationCallbackFails() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> partitionsToAssignAtSetup = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedPartitionsToRevoke = partitionsToAssignAtSetup;
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
@@ -921,37 +927,34 @@ public class StreamsMembershipManagerTest {
verify(subscriptionState, never())
.enablePartitionsAwaitingCallback(argThat(a -> !a.equals(partitionsToAssignAtSetup)));
verifyInStateReconciling(membershipManager);
- verify(streamsRebalanceEventsProcessor, never()).requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- );
+ verifyTasksNotAssigned(activeTasks, Set.of(), Set.of());
verifyInStateReconciling(membershipManager);
}
@Test
public void testReconcilingWhenReconciliationAbortedBeforeAssignmentDueToRejoin() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> partitionsToAssignAtSetup = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedPartitionsToRevoke = partitionsToAssignAtSetup;
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
@@ -961,8 +964,11 @@ public class StreamsMembershipManagerTest {
expectedFullPartitionsToAssign,
expectedNewPartitionsToAssign
);
+
membershipManager.onPollTimerExpired();
membershipManager.onHeartbeatRequestGenerated();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
onAllTasksLostCallbackExecuted.complete(null);
membershipManager.maybeRejoinStaleMember();
@@ -972,37 +978,34 @@ public class StreamsMembershipManagerTest {
verify(memberStateListener, never()).onGroupAssignmentUpdated(expectedFullPartitionsToAssign);
verify(subscriptionState, never())
.enablePartitionsAwaitingCallback(argThat(a -> !a.equals(partitionsToAssignAtSetup)));
- verify(streamsRebalanceEventsProcessor, never()).requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- );
+ verifyTasksNotAssigned(activeTasks, Set.of(), Set.of());
verifyInStateJoining(membershipManager);
}
@Test
public void testReconcilingWhenReconciliationAbortedBeforeAssignmentDueToNotInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> partitionsToAssignAtSetup = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedPartitionsToRevoke = partitionsToAssignAtSetup;
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
@@ -1013,6 +1016,8 @@ public class StreamsMembershipManagerTest {
expectedNewPartitionsToAssign
);
membershipManager.transitionToFatal();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
onAllTasksLostCallbackExecuted.complete(null);
onTasksRevokedCallbackExecuted.complete(null);
@@ -1021,40 +1026,34 @@ public class StreamsMembershipManagerTest {
verify(memberStateListener, never()).onGroupAssignmentUpdated(expectedFullPartitionsToAssign);
verify(subscriptionState, never())
.enablePartitionsAwaitingCallback(argThat(a -> !a.equals(partitionsToAssignAtSetup)));
- verify(streamsRebalanceEventsProcessor, never()).requestOnTasksAssignedCallbackInvocation(
- makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())
- );
+ verifyTasksNotAssigned(activeTasks, Set.of(), Set.of());
verifyInStateFatal(membershipManager);
}
@Test
public void testReconcilingWhenReconciliationAbortedAfterAssignmentDueToRejoin() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> partitionsToAssignAtSetup = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedPartitionsToRevoke = partitionsToAssignAtSetup;
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
@@ -1067,8 +1066,16 @@ public class StreamsMembershipManagerTest {
onTasksRevokedCallbackExecuted.complete(null);
membershipManager.onPollTimerExpired();
membershipManager.onHeartbeatRequestGenerated();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
onAllTasksLostCallbackExecuted.complete(null);
membershipManager.maybeRejoinStaleMember();
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
onTasksAssignedCallbackExecuted.complete(null);
@@ -1077,32 +1084,28 @@ public class StreamsMembershipManagerTest {
@Test
public void testReconcilingWhenReconciliationAbortedAfterAssignmentDueToNotInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_1)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
when(subscriptionState.assignedPartitions())
- .thenReturn(Collections.emptySet())
+ .thenReturn(Set.of())
.thenReturn(Set.of(new TopicPartition(TOPIC_0, PARTITION_0)));
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup);
final Set<TopicPartition> partitionsToAssignAtSetup = Set.of(new TopicPartition(TOPIC_0, PARTITION_0));
final Set<TopicPartition> expectedPartitionsToRevoke = partitionsToAssignAtSetup;
final Set<TopicPartition> expectedFullPartitionsToAssign = Set.of(new TopicPartition(TOPIC_0, PARTITION_1));
@@ -1114,8 +1117,17 @@ public class StreamsMembershipManagerTest {
);
onTasksRevokedCallbackExecuted.complete(null);
membershipManager.transitionToFatal();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
+
onAllTasksLostCallbackExecuted.complete(null);
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
onTasksAssignedCallbackExecuted.complete(null);
assertNotEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
@@ -1139,7 +1151,7 @@ public class StreamsMembershipManagerTest {
assertFalse(future.isCancelled());
assertFalse(future.isCompletedExceptionally());
verify(subscriptionState).unsubscribe();
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
verifyInStateUnsubscribed(membershipManager);
}
@@ -1154,11 +1166,9 @@ public class StreamsMembershipManagerTest {
}
private void testLeaveGroupOnCloseWhenNotInGroupAndFenced(final Supplier<CompletableFuture<Void>> leaveGroup) {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
fenced();
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
final CompletableFuture<Void> future = leaveGroup.get();
assertFalse(membershipManager.isLeavingGroup());
@@ -1166,28 +1176,30 @@ public class StreamsMembershipManagerTest {
assertFalse(future.isCancelled());
assertFalse(future.isCompletedExceptionally());
verify(subscriptionState).unsubscribe();
- verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
- verify(memberStateListener, times(2)).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(subscriptionState).assignFromSubscribed(Set.of());
+ verify(memberStateListener, times(2)).onGroupAssignmentUpdated(Set.of());
verifyInStateUnsubscribed(membershipManager);
}
@Test
public void testLeaveGroupWhenInGroupWithAssignment() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasks))
- .thenReturn(onTasksRevokedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
final CompletableFuture<Void> onGroupLeft = membershipManager.leaveGroup();
+ final CompletableFuture<Void> onTasksRevokedCallbackExecuted =
+ verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks);
assertFalse(onGroupLeft.isDone());
verify(subscriptionState, never()).unsubscribe();
verifyInStatePrepareLeaving(membershipManager);
@@ -1196,7 +1208,7 @@ public class StreamsMembershipManagerTest {
final CompletableFuture<Void> onGroupLeftOnCloseBeforeRevocationCallback = membershipManager.leaveGroupOnClose();
assertEquals(onGroupLeft, onGroupLeftOnCloseBeforeRevocationCallback);
onTasksRevokedCallbackExecuted.complete(null);
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
verify(subscriptionState).unsubscribe();
assertFalse(onGroupLeft.isDone());
verifyInStateLeaving(membershipManager);
@@ -1212,14 +1224,18 @@ public class StreamsMembershipManagerTest {
@Test
public void testLeaveGroupOnCloseWhenInGroupWithAssignment() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasks,
+ Set.of(),
+ Set.of()
+ );
+
acknowledging(onTasksAssignedCallbackExecutedSetup);
final CompletableFuture<Void> onGroupLeft = membershipManager.leaveGroupOnClose();
@@ -1227,8 +1243,8 @@ public class StreamsMembershipManagerTest {
assertFalse(onGroupLeft.isDone());
verifyInStateLeaving(membershipManager);
verify(subscriptionState).unsubscribe();
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
- verify(streamsRebalanceEventsProcessor, never()).requestOnTasksRevokedCallbackInvocation(any());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
+ verify(backgroundEventHandler, never()).add(any(StreamsOnTasksRevokedCallbackNeededEvent.class));
final CompletableFuture<Void> onGroupLeftBeforeHeartbeatRequestGenerated = membershipManager.leaveGroup();
assertEquals(onGroupLeft, onGroupLeftBeforeHeartbeatRequestGenerated);
final CompletableFuture<Void> onGroupLeftOnCloseBeforeHeartbeatRequestGenerated = membershipManager.leaveGroupOnClose();
@@ -1244,20 +1260,20 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestSkippedWhenInLeaving() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, "topic");
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, "topic");
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onAllTasksRevokedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onAllTasksRevokedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
acknowledging(onTasksAssignedCallbackExecutedSetup);
- CompletableFuture<Void> future = leaving(onAllTasksRevokedCallbackExecuted);
+ CompletableFuture<Void> future = leaving();
membershipManager.onHeartbeatRequestSkipped();
@@ -1269,20 +1285,21 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatSuccessWhenInLeaving() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, "topic");
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, "topic");
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onTasksRevokedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasksSetup))
- .thenReturn(onTasksRevokedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(
+ activeTasksSetup,
+ Set.of(),
+ Set.of()
+ );
+
acknowledging(onTasksAssignedCallbackExecutedSetup);
- CompletableFuture<Void> future = leaving(onTasksRevokedCallbackExecuted);
+ CompletableFuture<Void> future = leaving();
membershipManager.onHeartbeatSuccess(makeHeartbeatResponseWithActiveTasks(
SUBTOPOLOGY_ID_0, List.of(PARTITION_0),
@@ -1308,9 +1325,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatSuccessWhenInFenced() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
fenced();
@@ -1336,9 +1350,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatSuccessWhenInStale() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
membershipManager.onPollTimerExpired();
membershipManager.onHeartbeatRequestGenerated();
@@ -1350,11 +1361,10 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatSuccessWhenInReconciling() {
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(Set.of(), Set.of(), Set.of())))
- .thenReturn(onTasksAssignedCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(List.of(), MEMBER_EPOCH));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecuted =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(Set.of(), Set.of(), Set.of());
onTasksAssignedCallbackExecuted.complete(null);
membershipManager.onHeartbeatRequestGenerated();
@@ -1366,14 +1376,13 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnPollTimerExpired() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
membershipManager.onPollTimerExpired();
@@ -1400,14 +1409,13 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInAcknowleding() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
membershipManager.onHeartbeatRequestGenerated();
@@ -1417,14 +1425,13 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInAcknowledgingAndNewTargetAssignment() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_1)));
acknowledging(onTasksAssignedCallbackExecutedSetup);
@@ -1435,11 +1442,8 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInLeaving() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- leaving(onAllTasksLostCallbackExecuted);
+ leavingAtMemberEpochZero();
membershipManager.onHeartbeatRequestGenerated();
@@ -1448,28 +1452,24 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnHeartbeatRequestGeneratedWhenInLeavingAndPollTimerExpired() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
membershipManager.onPollTimerExpired();
membershipManager.onHeartbeatRequestGenerated();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
verifyInStateStale(membershipManager);
- verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
+ verify(subscriptionState, never()).assignFromSubscribed(Set.of());
onAllTasksLostCallbackExecuted.complete(null);
- verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(subscriptionState).assignFromSubscribed(Set.of());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
}
@Test
public void testOnHeartbeatFailureAfterLeaveRequestGenerated() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- final CompletableFuture<Void> groupLeft = leaving(onAllTasksLostCallbackExecuted);
+ final CompletableFuture<Void> groupLeft = leavingAtMemberEpochZero();
membershipManager.onHeartbeatRequestGenerated();
assertFalse(groupLeft.isDone());
@@ -1485,12 +1485,7 @@ public class StreamsMembershipManagerTest {
"failed-rebalance-total",
CONSUMER_METRIC_GROUP_PREFIX + COORDINATOR_METRICS_SUFFIX
);
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final Set<StreamsRebalanceData.TaskId> activeTasks =
- Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecuted);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
joining();
time.sleep(1);
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
@@ -1516,12 +1511,7 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final Set<StreamsRebalanceData.TaskId> activeTasks =
- Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
@@ -1530,14 +1520,13 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInAcknowledging() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
@@ -1545,14 +1534,13 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInStable() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasks =
Set.of(new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0));
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasks, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasks, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
stable();
@@ -1560,26 +1548,22 @@ public class StreamsMembershipManagerTest {
}
private void testOnFencedWhenInJoiningOrReconcilingOrAcknowledgingOrStable() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
-
membershipManager.onFenced();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
+
verifyInStateFenced(membershipManager);
assertEquals(StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
- verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
+ verify(subscriptionState, never()).assignFromSubscribed(Set.of());
onAllTasksLostCallbackExecuted.complete(null);
- verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(subscriptionState).assignFromSubscribed(Set.of());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
verifyInStateJoining(membershipManager);
}
@Test
public void testOnFencedWhenInPrepareLeaving() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
testOnFencedWhenInPrepareLeavingOrLeaving(prepareLeaving());
@@ -1587,12 +1571,9 @@ public class StreamsMembershipManagerTest {
@Test
public void testOnFencedWhenInLeaving() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testOnFencedWhenInPrepareLeavingOrLeaving(leaving(onAllTasksLostCallbackExecuted));
+ testOnFencedWhenInPrepareLeavingOrLeaving(leavingAtMemberEpochZero());
}
private void testOnFencedWhenInPrepareLeavingOrLeaving(final CompletableFuture<Void> onGroupLeft) {
@@ -1607,9 +1588,6 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInPrepareLeaving() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
testTransitionToFatalWhenInPrepareLeavingOrLeaving(prepareLeaving());
@@ -1619,12 +1597,9 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInLeaving() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testTransitionToFatalWhenInPrepareLeavingOrLeaving(leaving(onAllTasksLostCallbackExecuted));
+ testTransitionToFatalWhenInPrepareLeavingOrLeaving(leavingAtMemberEpochZero());
verify(memberStateListener, times(2)).onMemberEpochUpdated(Optional.empty(), membershipManager.memberId());
}
@@ -1639,80 +1614,70 @@ public class StreamsMembershipManagerTest {
@Test
public void testTransitionToFatalWhenInJoining() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
@Test
public void testTransitionToFatalWhenInReconciling() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
+ acknowledging(onTasksAssignedCallbackExecutedSetup);
+ stable();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
@Test
public void testTransitionToFatalWhenInAcknowledging() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
@Test
public void testTransitionToFatalWhenInStable() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
);
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
+ final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup =
+ verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(activeTasksSetup, Set.of(), Set.of());
acknowledging(onTasksAssignedCallbackExecutedSetup);
stable();
- testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(onAllTasksLostCallbackExecuted);
+ testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable();
}
- private void testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable(final CompletableFuture<Void> future) {
+ private void testTransitionToFatalWhenInJoiningOrReconcilingOrAcknowledgingOrStable() {
membershipManager.transitionToFatal();
- verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
- future.complete(null);
- verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
+
+ verify(subscriptionState, never()).assignFromSubscribed(Set.of());
+ onAllTasksLostCallbackExecuted.complete(null);
+ verify(subscriptionState).assignFromSubscribed(Set.of());
verifyInStateFatal(membershipManager);
verify(memberStateListener).onMemberEpochUpdated(Optional.empty(), membershipManager.memberId());
- verify(memberStateListener).onGroupAssignmentUpdated(Collections.emptySet());
+ verify(memberStateListener).onGroupAssignmentUpdated(Set.of());
}
@Test
@@ -1721,9 +1686,8 @@ public class StreamsMembershipManagerTest {
verifyInStateFatal(membershipManager);
verify(memberStateListener).onMemberEpochUpdated(Optional.empty(), membershipManager.memberId());
- verify(streamsRebalanceEventsProcessor, never()).requestOnAllTasksLostCallbackInvocation();
- verify(streamsRebalanceEventsProcessor, never()).requestOnAllTasksLostCallbackInvocation();
- verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
+ verify(backgroundEventHandler, never()).add(any(StreamsOnAllTasksLostCallbackNeededEvent.class));
+ verify(subscriptionState, never()).assignFromSubscribed(Set.of());
}
@Test
@@ -1836,12 +1800,11 @@ public class StreamsMembershipManagerTest {
@Test
public void testMaybeRejoinStaleMember() {
- final CompletableFuture<Void> onAllTasksLostCallbackExecuted = new CompletableFuture<>();
- when(streamsRebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation())
- .thenReturn(onAllTasksLostCallbackExecuted);
joining();
membershipManager.onPollTimerExpired();
membershipManager.onHeartbeatRequestGenerated();
+ final CompletableFuture<Void> onAllTasksLostCallbackExecuted =
+ verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler();
verifyInStateStale(membershipManager);
membershipManager.maybeRejoinStaleMember();
@@ -1876,13 +1839,7 @@ public class StreamsMembershipManagerTest {
@Test
public void testConsumerPollWhenNotJoining() {
- setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
- final CompletableFuture<Void> onTasksAssignedCallbackExecutedSetup = new CompletableFuture<>();
- final Set<StreamsRebalanceData.TaskId> activeTasksSetup = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_ID_0, PARTITION_0)
- );
- when(streamsRebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(makeTaskAssignment(activeTasksSetup, Collections.emptySet(), Collections.emptySet())))
- .thenReturn(onTasksAssignedCallbackExecutedSetup);
+ setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(SUBTOPOLOGY_ID_0, TOPIC_0);
joining();
reconcile(makeHeartbeatResponseWithActiveTasks(SUBTOPOLOGY_ID_0, List.of(PARTITION_0)));
membershipManager.onSubscriptionUpdated();
@@ -1900,7 +1857,7 @@ public class StreamsMembershipManagerTest {
}
private void verifyThatNoTasksHaveBeenRevoked() {
- verify(streamsRebalanceEventsProcessor, never()).requestOnTasksRevokedCallbackInvocation(any());
+ verify(backgroundEventHandler, never()).add(any(StreamsOnTasksRevokedCallbackNeededEvent.class));
verify(subscriptionState, never()).markPendingRevocation(any());
}
@@ -1995,15 +1952,52 @@ public class StreamsMembershipManagerTest {
assertFalse(membershipManager.isLeavingGroup());
}
- private void setupStreamsAssignmentInterfaceWithOneSubtopologyOneSourceTopic(final String subtopologyId,
- final String topicName) {
+ private CompletableFuture<Void> verifyOnTasksAssignedCallbackNeededEventAddedToBackgroundEventHandler(final Set<StreamsRebalanceData.TaskId> activeTasks,
+ final Set<StreamsRebalanceData.TaskId> standbyTasks,
+ final Set<StreamsRebalanceData.TaskId> warmupTasks) {
+ verify(backgroundEventHandler, times(++onTasksAssignedCallbackNeededAddCount)).add(onTasksAssignedCallbackNeededEventCaptor.capture());
+ final StreamsOnTasksAssignedCallbackNeededEvent onTasksAssignedCallbackNeeded = onTasksAssignedCallbackNeededEventCaptor.getValue();
+ assertEquals(makeTaskAssignment(activeTasks, standbyTasks, warmupTasks), onTasksAssignedCallbackNeeded.assignment());
+ return onTasksAssignedCallbackNeeded.future();
+ }
+
+ private CompletableFuture<Void> verifyOnTasksRevokedCallbackNeededEventAddedToBackgroundEventHandler(final Set<StreamsRebalanceData.TaskId> activeTasksToRevoke) {
+ verify(backgroundEventHandler).add(onTasksRevokedCallbackNeededEventCaptor.capture());
+ final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = onTasksRevokedCallbackNeededEventCaptor.getValue();
+ assertEquals(
+ activeTasksToRevoke,
+ onTasksRevokedCallbackNeededEvent.activeTasksToRevoke()
+ );
+ return onTasksRevokedCallbackNeededEvent.future();
+ }
+
+ private CompletableFuture<Void> verifyOnAllTasksLostCallbackNeededEventAddedToBackgroundEventHandler() {
+ verify(backgroundEventHandler).add(onAllTasksLostCallbackNeededEventCaptor.capture());
+ final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = onAllTasksLostCallbackNeededEventCaptor.getValue();
+ return onAllTasksLostCallbackNeededEvent.future();
+ }
+
+ private void verifyTasksNotAssigned(final Set<StreamsRebalanceData.TaskId> activeTasks,
+ final Set<StreamsRebalanceData.TaskId> standbyTasks,
+ final Set<StreamsRebalanceData.TaskId> warmupTasks) {
+ verify(backgroundEventHandler, never()).add(argThat(a -> {
+ if (a instanceof StreamsOnTasksAssignedCallbackNeededEvent) {
+ return ((StreamsOnTasksAssignedCallbackNeededEvent) a).assignment()
+ .equals(makeTaskAssignment(activeTasks, standbyTasks, warmupTasks));
+ }
+ return false;
+ }));
+ }
+
+ private void setupStreamsRebalanceDataWithOneSubtopologyOneSourceTopic(final String subtopologyId,
+ final String topicName) {
when(streamsRebalanceData.subtopologies()).thenReturn(
mkMap(
mkEntry(
subtopologyId,
new StreamsRebalanceData.Subtopology(
Set.of(topicName),
- Collections.emptySet(),
+ Set.of(),
Collections.emptyMap(),
Collections.emptyMap(),
Collections.emptyList()
@@ -2013,17 +2007,17 @@ public class StreamsMembershipManagerTest {
);
}
- private void setupStreamsAssignmentInterfaceWithTwoSubtopologies(final String subtopologyId1,
- final String topicName1,
- final String subtopologyId2,
- final String topicName2) {
+ private void setupStreamsReabalanceDataWithTwoSubtopologies(final String subtopologyId1,
+ final String topicName1,
+ final String subtopologyId2,
+ final String topicName2) {
when(streamsRebalanceData.subtopologies()).thenReturn(
mkMap(
mkEntry(
subtopologyId1,
new StreamsRebalanceData.Subtopology(
Set.of(topicName1),
- Collections.emptySet(),
+ Set.of(),
Collections.emptyMap(),
Collections.emptyMap(),
Collections.emptyList()
@@ -2033,7 +2027,7 @@ public class StreamsMembershipManagerTest {
subtopologyId2,
new StreamsRebalanceData.Subtopology(
Set.of(topicName2),
- Collections.emptySet(),
+ Set.of(),
Collections.emptyMap(),
Collections.emptyMap(),
Collections.emptyList()
@@ -2167,9 +2161,20 @@ public class StreamsMembershipManagerTest {
return onGroupLeft;
}
- private CompletableFuture<Void> leaving(final CompletableFuture<Void> onAllTasksRevokedCallbackExecuted) {
+ private CompletableFuture<Void> leaving() {
+ final CompletableFuture<Void> future = prepareLeaving();
+ verify(backgroundEventHandler).add(onTasksRevokedCallbackNeededEventCaptor.capture());
+ final StreamsOnTasksRevokedCallbackNeededEvent onTasksRevokedCallbackNeededEvent = onTasksRevokedCallbackNeededEventCaptor.getValue();
+ onTasksRevokedCallbackNeededEvent.future().complete(null);
+ verifyInStateLeaving(membershipManager);
+ return future;
+ }
+
+ private CompletableFuture<Void> leavingAtMemberEpochZero() {
final CompletableFuture<Void> future = prepareLeaving();
- onAllTasksRevokedCallbackExecuted.complete(null);
+ verify(backgroundEventHandler).add(onAllTasksLostCallbackNeededEventCaptor.capture());
+ final StreamsOnAllTasksLostCallbackNeededEvent onAllTasksLostCallbackNeededEvent = onAllTasksLostCallbackNeededEventCaptor.getValue();
+ onAllTasksLostCallbackNeededEvent.future().complete(null);
verifyInStateLeaving(membershipManager);
return future;
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessorTest.java
deleted file mode 100644
index 5bc988cf9c7..00000000000
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceEventsProcessorTest.java
+++ /dev/null
@@ -1,286 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.clients.consumer.internals;
-
-import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
-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.common.KafkaException;
-
-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.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-@ExtendWith(MockitoExtension.class)
-public class StreamsRebalanceEventsProcessorTest {
-
- private static final String SUBTOPOLOGY_0 = "subtopology-0";
- private static final String SUBTOPOLOGY_1 = "subtopology-1";
-
- @Mock
- private StreamsGroupRebalanceCallbacks rebalanceCallbacks;
-
- @Mock
- private ApplicationEventHandler applicationEventHandler;
-
- private StreamsRebalanceData rebalanceData;
-
- @BeforeEach
- public void setup() {
- final UUID processId = UUID.randomUUID();
- final Optional<StreamsRebalanceData.HostInfo> endpoint = Optional.of(new StreamsRebalanceData.HostInfo("localhost", 9090));
- final Map<String, StreamsRebalanceData.Subtopology> subtopologies = new HashMap<>();
- final Map<String, String> clientTags = Map.of("clientTag1", "clientTagValue1");
- rebalanceData = new StreamsRebalanceData(
- processId,
- endpoint,
- subtopologies,
- clientTags
- );
- }
-
- @Test
- public void shouldInvokeOnTasksAssignedCallback() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0)
- );
- final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 2),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 3)
- );
- StreamsRebalanceData.Assignment assignment =
- new StreamsRebalanceData.Assignment(activeTasks, standbyTasks, warmupTasks);
- when(rebalanceCallbacks.onTasksAssigned(assignment)).thenReturn(Optional.empty());
-
- final CompletableFuture<Void> onTasksAssignedExecuted = rebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(assignment);
-
- assertFalse(onTasksAssignedExecuted.isDone());
- rebalanceEventsProcessor.process();
- ArgumentCaptor<StreamsOnTasksAssignedCallbackCompletedEvent> streamsOnTasksAssignedCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnTasksAssignedCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnTasksAssignedCallbackCompletedCaptor.capture());
- StreamsOnTasksAssignedCallbackCompletedEvent streamsOnTasksAssignedCallbackCompletedEvent =
- streamsOnTasksAssignedCallbackCompletedCaptor.getValue();
- assertFalse(streamsOnTasksAssignedCallbackCompletedEvent.future().isDone());
- assertTrue(streamsOnTasksAssignedCallbackCompletedEvent.error().isEmpty());
- assertEquals(assignment, rebalanceData.reconciledAssignment());
- }
-
- @Test
- public void shouldReThrowErrorFromOnTasksAssignedCallbackAndPassErrorToBackground() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0)
- );
- final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 2),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 3)
- );
- StreamsRebalanceData.Assignment assignment =
- new StreamsRebalanceData.Assignment(activeTasks, standbyTasks, warmupTasks);
- final Exception exception = new RuntimeException("Nobody expects the Spanish inquisition.");
- when(rebalanceCallbacks.onTasksAssigned(assignment)).thenReturn(Optional.of(exception));
-
- final CompletableFuture<Void> onTasksAssignedExecuted = rebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(assignment);
-
- assertFalse(onTasksAssignedExecuted.isDone());
- final Exception actualException = assertThrows(KafkaException.class, rebalanceEventsProcessor::process);
- assertEquals("Task assignment callback throws an error", actualException.getMessage());
- ArgumentCaptor<StreamsOnTasksAssignedCallbackCompletedEvent> streamsOnTasksAssignedCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnTasksAssignedCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnTasksAssignedCallbackCompletedCaptor.capture());
- StreamsOnTasksAssignedCallbackCompletedEvent streamsOnTasksAssignedCallbackCompletedEvent =
- streamsOnTasksAssignedCallbackCompletedCaptor.getValue();
- assertFalse(streamsOnTasksAssignedCallbackCompletedEvent.future().isDone());
- assertTrue(streamsOnTasksAssignedCallbackCompletedEvent.error().isPresent());
- assertEquals(exception, streamsOnTasksAssignedCallbackCompletedEvent.error().get().getCause());
- assertEquals(exception, actualException.getCause());
- assertEquals(StreamsRebalanceData.Assignment.EMPTY, rebalanceData.reconciledAssignment());
- }
-
- @Test
- public void shouldInvokeOnTasksRevokedCallback() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- when(rebalanceCallbacks.onTasksRevoked(activeTasks)).thenReturn(Optional.empty());
-
- final CompletableFuture<Void> onTasksRevokedExecuted = rebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasks);
-
- assertFalse(onTasksRevokedExecuted.isDone());
- rebalanceEventsProcessor.process();
- ArgumentCaptor<StreamsOnTasksRevokedCallbackCompletedEvent> streamsOnTasksRevokedCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnTasksRevokedCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnTasksRevokedCallbackCompletedCaptor.capture());
- StreamsOnTasksRevokedCallbackCompletedEvent streamsOnTasksRevokedCallbackCompletedEvent =
- streamsOnTasksRevokedCallbackCompletedCaptor.getValue();
- assertFalse(streamsOnTasksRevokedCallbackCompletedEvent.future().isDone());
- assertTrue(streamsOnTasksRevokedCallbackCompletedEvent.error().isEmpty());
- }
-
- @Test
- public void shouldReThrowErrorFromOnTasksRevokedCallbackAndPassErrorToBackground() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- final Exception exception = new RuntimeException("Nobody expects the Spanish inquisition.");
- when(rebalanceCallbacks.onTasksRevoked(activeTasks)).thenReturn(Optional.of(exception));
-
- final CompletableFuture<Void> onTasksRevokedExecuted = rebalanceEventsProcessor.requestOnTasksRevokedCallbackInvocation(activeTasks);
-
- assertFalse(onTasksRevokedExecuted.isDone());
- final Exception actualException = assertThrows(KafkaException.class, rebalanceEventsProcessor::process);
- assertEquals("Task revocation callback throws an error", actualException.getMessage());
- ArgumentCaptor<StreamsOnTasksRevokedCallbackCompletedEvent> streamsOnTasksRevokedCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnTasksRevokedCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnTasksRevokedCallbackCompletedCaptor.capture());
- StreamsOnTasksRevokedCallbackCompletedEvent streamsOnTasksRevokedCallbackCompletedEvent =
- streamsOnTasksRevokedCallbackCompletedCaptor.getValue();
- assertTrue(streamsOnTasksRevokedCallbackCompletedEvent.error().isPresent());
- assertEquals(exception, streamsOnTasksRevokedCallbackCompletedEvent.error().get().getCause());
- assertEquals(exception, actualException.getCause());
- }
-
- @Test
- public void shouldInvokeOnAllTasksLostCallback() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0)
- );
- final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 2),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 3)
- );
- StreamsRebalanceData.Assignment assignment =
- new StreamsRebalanceData.Assignment(activeTasks, standbyTasks, warmupTasks);
- when(rebalanceCallbacks.onTasksAssigned(assignment)).thenReturn(Optional.empty());
- rebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(assignment);
- rebalanceEventsProcessor.process();
- assertEquals(assignment, rebalanceData.reconciledAssignment());
- when(rebalanceCallbacks.onAllTasksLost()).thenReturn(Optional.empty());
-
- final CompletableFuture<Void> onAllTasksLostExecuted = rebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
-
- assertFalse(onAllTasksLostExecuted.isDone());
- rebalanceEventsProcessor.process();
- ArgumentCaptor<StreamsOnAllTasksLostCallbackCompletedEvent> streamsOnAllTasksLostCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnAllTasksLostCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnAllTasksLostCallbackCompletedCaptor.capture());
- StreamsOnAllTasksLostCallbackCompletedEvent streamsOnAllTasksLostCallbackCompletedEvent =
- streamsOnAllTasksLostCallbackCompletedCaptor.getValue();
- assertFalse(streamsOnAllTasksLostCallbackCompletedEvent.future().isDone());
- assertTrue(streamsOnAllTasksLostCallbackCompletedEvent.error().isEmpty());
- assertEquals(StreamsRebalanceData.Assignment.EMPTY, rebalanceData.reconciledAssignment());
- }
-
- @Test
- public void shouldReThrowErrorFromOnAllTasksLostCallbackAndPassErrorToBackground() {
- final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData);
- rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
- rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
- final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1)
- );
- final Set<StreamsRebalanceData.TaskId> standbyTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 1),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0)
- );
- final Set<StreamsRebalanceData.TaskId> warmupTasks = Set.of(
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_1, 2),
- new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 3)
- );
- StreamsRebalanceData.Assignment assignment =
- new StreamsRebalanceData.Assignment(activeTasks, standbyTasks, warmupTasks);
- when(rebalanceCallbacks.onTasksAssigned(assignment)).thenReturn(Optional.empty());
- rebalanceEventsProcessor.requestOnTasksAssignedCallbackInvocation(assignment);
- rebalanceEventsProcessor.process();
- assertEquals(assignment, rebalanceData.reconciledAssignment());
- final Exception exception = new RuntimeException("Nobody expects the Spanish inquisition.");
- when(rebalanceCallbacks.onAllTasksLost()).thenReturn(Optional.of(exception));
-
- final CompletableFuture<Void> onAllTasksLostExecuted = rebalanceEventsProcessor.requestOnAllTasksLostCallbackInvocation();
-
- assertFalse(onAllTasksLostExecuted.isDone());
- final Exception actualException = assertThrows(KafkaException.class, rebalanceEventsProcessor::process);
- assertEquals("All tasks lost callback throws an error", actualException.getMessage());
- ArgumentCaptor<StreamsOnAllTasksLostCallbackCompletedEvent> streamsOnAllTasksLostCallbackCompletedCaptor =
- ArgumentCaptor.forClass(StreamsOnAllTasksLostCallbackCompletedEvent.class);
- verify(applicationEventHandler).add(streamsOnAllTasksLostCallbackCompletedCaptor.capture());
- StreamsOnAllTasksLostCallbackCompletedEvent streamsOnAllTasksLostCallbackCompletedEvent =
- streamsOnAllTasksLostCallbackCompletedCaptor.getValue();
- assertFalse(streamsOnAllTasksLostCallbackCompletedEvent.future().isDone());
- assertTrue(streamsOnAllTasksLostCallbackCompletedEvent.error().isPresent());
- assertEquals(exception, streamsOnAllTasksLostCallbackCompletedEvent.error().get().getCause());
- assertEquals(exception, actualException.getCause());
- assertEquals(assignment, rebalanceData.reconciledAssignment());
- }
-}
\ No newline at end of file
diff --git a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java
index 67c53e7fe12..604fdd9b6a1 100644
--- a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java
+++ b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/SmokeTestDriverIntegrationTest.java
@@ -114,6 +114,8 @@ public class SmokeTestDriverIntegrationTest {
"true, false, true",
"false, false, false",
"true, false, false",
+ "true, true, true",
+ "true, true, false"
})
public void shouldWorkWithRebalance(
final boolean stateUpdaterEnabled,
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsGroupRebalanceCallbacks.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
similarity index 89%
rename from streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsGroupRebalanceCallbacks.java
rename to streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
index 37e61427ccd..dcc4821f2a8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsGroupRebalanceCallbacks.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsRebalanceListener.java
@@ -16,8 +16,8 @@
*/
package org.apache.kafka.streams.processor.internals;
-import org.apache.kafka.clients.consumer.internals.StreamsGroupRebalanceCallbacks;
import org.apache.kafka.clients.consumer.internals.StreamsRebalanceData;
+import org.apache.kafka.clients.consumer.internals.StreamsRebalanceListener;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.processor.TaskId;
@@ -31,7 +31,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-public class DefaultStreamsGroupRebalanceCallbacks implements StreamsGroupRebalanceCallbacks {
+public class DefaultStreamsRebalanceListener implements StreamsRebalanceListener {
private final Logger log;
private final Time time;
@@ -39,11 +39,11 @@ public class DefaultStreamsGroupRebalanceCallbacks implements StreamsGroupRebala
private final TaskManager taskManager;
private final StreamThread streamThread;
- public DefaultStreamsGroupRebalanceCallbacks(final Logger log,
- final Time time,
- final StreamsRebalanceData streamsRebalanceData,
- final StreamThread streamThread,
- final TaskManager taskManager) {
+ public DefaultStreamsRebalanceListener(final Logger log,
+ final Time time,
+ final StreamsRebalanceData streamsRebalanceData,
+ final StreamThread streamThread,
+ final TaskManager taskManager) {
this.log = log;
this.time = time;
this.streamsRebalanceData = streamsRebalanceData;
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 a96d968ca27..9bcb6cb6627 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
@@ -28,7 +28,6 @@ import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer;
import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.StreamsRebalanceData;
-import org.apache.kafka.clients.consumer.internals.StreamsRebalanceEventsProcessor;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Metric;
@@ -353,7 +352,6 @@ public class StreamThread extends Thread implements ProcessingThread {
private final Queue<StreamsException> nonFatalExceptionsToHandle;
private final Optional<StreamsRebalanceData> streamsRebalanceData;
- private final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor;
private final StreamsMetadataState streamsMetadataState;
// These are used to signal from outside the stream thread, but the variables themselves are internal to the thread
@@ -521,7 +519,6 @@ public class StreamThread extends Thread implements ProcessingThread {
streamsUncaughtExceptionHandler,
cache::resize,
mainConsumerSetup.streamsRebalanceData,
- mainConsumerSetup.streamsRebalanceEventsProcessor,
streamsMetadataState
);
@@ -548,8 +545,6 @@ public class StreamThread extends Thread implements ProcessingThread {
topologyMetadata
)
);
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor =
- Optional.of(new StreamsRebalanceEventsProcessor(streamsRebalanceData.get()));
final ByteArrayDeserializer keyDeserializer = new ByteArrayDeserializer();
final ByteArrayDeserializer valueDeserializer = new ByteArrayDeserializer();
return new MainConsumerSetup(
@@ -557,16 +552,13 @@ public class StreamThread extends Thread implements ProcessingThread {
new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(consumerConfigs, keyDeserializer, valueDeserializer)),
keyDeserializer,
valueDeserializer,
- streamsRebalanceData,
- streamsRebalanceEventsProcessor
+ streamsRebalanceData
),
- streamsRebalanceData,
- streamsRebalanceEventsProcessor
+ streamsRebalanceData
);
} else {
- return new MainConsumerSetup(
+ return new MainConsumerSetup(
clientSupplier.getConsumer(consumerConfigs),
- Optional.empty(),
Optional.empty()
);
}
@@ -575,14 +567,11 @@ public class StreamThread extends Thread implements ProcessingThread {
private static class MainConsumerSetup {
public final Consumer<byte[], byte[]> mainConsumer;
public final Optional<StreamsRebalanceData> streamsRebalanceData;
- public final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor;
public MainConsumerSetup(final Consumer<byte[], byte[]> mainConsumer,
- final Optional<StreamsRebalanceData> streamsRebalanceData,
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor) {
+ final Optional<StreamsRebalanceData> streamsRebalanceData) {
this.mainConsumer = mainConsumer;
this.streamsRebalanceData = streamsRebalanceData;
- this.streamsRebalanceEventsProcessor = streamsRebalanceEventsProcessor;
}
}
@@ -751,7 +740,6 @@ public class StreamThread extends Thread implements ProcessingThread {
final BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler,
final java.util.function.Consumer<Long> cacheResizer,
final Optional<StreamsRebalanceData> streamsRebalanceData,
- final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor,
final StreamsMetadataState streamsMetadataState
) {
super(threadId);
@@ -837,12 +825,6 @@ public class StreamThread extends Thread implements ProcessingThread {
this.logSummaryIntervalMs = config.getLong(StreamsConfig.LOG_SUMMARY_INTERVAL_MS_CONFIG);
this.streamsRebalanceData = streamsRebalanceData;
- this.streamsRebalanceEventsProcessor = streamsRebalanceEventsProcessor;
- if (streamsRebalanceData.isPresent() && streamsRebalanceEventsProcessor.isPresent()) {
- streamsRebalanceEventsProcessor.get().setRebalanceCallbacks(
- new DefaultStreamsGroupRebalanceCallbacks(log, time, streamsRebalanceData.get(), this, taskManager)
- );
- }
this.streamsMetadataState = streamsMetadataState;
}
@@ -1106,9 +1088,26 @@ public class StreamThread extends Thread implements ProcessingThread {
private void subscribeConsumer() {
if (topologyMetadata.usesPatternSubscription()) {
+ if (streamsRebalanceData.isPresent()) {
+ throw new IllegalArgumentException("Pattern subscription is not yet supported with the Streams rebalance " +
+ "protocol");
+ }
mainConsumer.subscribe(topologyMetadata.sourceTopicPattern(), rebalanceListener);
} else {
- mainConsumer.subscribe(topologyMetadata.allFullSourceTopicNames(), rebalanceListener);
+ if (streamsRebalanceData.isPresent()) {
+ ((AsyncKafkaConsumer<byte[], byte[]>) mainConsumer).subscribe(
+ topologyMetadata.allFullSourceTopicNames(),
+ new DefaultStreamsRebalanceListener(
+ log,
+ time,
+ streamsRebalanceData.get(),
+ this,
+ taskManager
+ )
+ );
+ } else {
+ mainConsumer.subscribe(topologyMetadata.allFullSourceTopicNames(), rebalanceListener);
+ }
}
}
@@ -1139,13 +1138,13 @@ public class StreamThread extends Thread implements ProcessingThread {
final long startMs = time.milliseconds();
now = startMs;
- maybeHandleAssignmentFromStreamsRebalanceProtocol();
-
final long pollLatency;
taskManager.resumePollingForPartitionsWithAvailableSpace();
pollLatency = pollPhase();
totalPolledSinceLastSummary += 1;
+ handleStreamsRebalanceData();
+
// Shutdown hook could potentially be triggered and transit the thread state to PENDING_SHUTDOWN during #pollRequests().
// The task manager internal states could be uninitialized if the state transition happens during #onPartitionsAssigned().
// Should only proceed when the thread is still running after #pollRequests(), because no external state mutation
@@ -1295,6 +1294,8 @@ public class StreamThread extends Thread implements ProcessingThread {
taskManager.resumePollingForPartitionsWithAvailableSpace();
pollLatency = pollPhase();
+ handleStreamsRebalanceData();
+
// Shutdown hook could potentially be triggered and transit the thread state to PENDING_SHUTDOWN during #pollRequests().
// The task manager internal states could be uninitialized if the state transition happens during #onPartitionsAssigned().
// Should only proceed when the thread is still running after #pollRequests(), because no external state mutation
@@ -1476,17 +1477,12 @@ public class StreamThread extends Thread implements ProcessingThread {
return records;
}
- public void maybeHandleAssignmentFromStreamsRebalanceProtocol() {
+ public void handleStreamsRebalanceData() {
if (streamsRebalanceData.isPresent()) {
if (streamsRebalanceData.get().shutdownRequested()) {
assignmentErrorCode.set(AssignorError.SHUTDOWN_REQUESTED.code());
}
-
- // ToDo: process IQ-related metadata
-
- // Process assignment from Streams Rebalance Protocol
- streamsRebalanceEventsProcessor.get().process();
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index fa72b116f00..8ef4e09bb27 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -1449,7 +1449,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
).updateThreadMetadata(adminClientId(CLIENT_ID));
@@ -2671,7 +2670,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -2733,7 +2731,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -2804,7 +2801,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -2871,7 +2867,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -2935,7 +2930,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -3170,7 +3164,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
);
final MetricName testMetricName = new MetricName("test_metric", "", "", new HashMap<>());
@@ -3230,7 +3223,6 @@ public class StreamThreadTest {
(e, b) -> { },
null,
Optional.empty(),
- Optional.empty(),
null
) {
@Override
@@ -3613,7 +3605,6 @@ public class StreamThreadTest {
null,
null,
Optional.empty(),
- Optional.empty(),
null
);
}
@@ -3738,7 +3729,6 @@ public class StreamThreadTest {
HANDLER,
null,
Optional.empty(),
- Optional.empty(),
null
);
}