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/09 11:26:57 UTC
(kafka) branch trunk updated: KAFKA-19037: Integrate consumer-side code with Streams (#19377)
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 2a370ed7213 KAFKA-19037: Integrate consumer-side code with Streams (#19377)
2a370ed7213 is described below
commit 2a370ed7213c7de74acef8b66955c4b91ba791e2
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Wed Apr 9 13:26:51 2025 +0200
KAFKA-19037: Integrate consumer-side code with Streams (#19377)
The consumer adaptations for the new Streams rebalance protocol need to
be integrated into the Streams code. This commit does the following:
- creates an async Kafka consumer
- with a Streams heartbeat request manager
- with a Streams membership manager
- integrates consumer code with the Streams membership manager and the
Streams heartbeat request manager
- processes the events from the consumer network thread (a.k.a.
background thread)
that request the invocation of the "on tasks revoked", "on tasks
assigned", and "on all tasks lost"
callbacks
- executes the callbacks
- sends to the consumer network thread the events signalling the
execution of the callbacks
- adapts SmokeTestDriverIntegrationTest to use the new Streams rebalance
protocol
This commit misses some unit test coverage, but it also unblocks other
work on trunk regarding the new Streams rebalance protocol. The missing
unit tests will be added soon.
Reviewers: Lucas Brutschy <lb...@confluent.io>
---
.../consumer/internals/AsyncKafkaConsumer.java | 30 ++-
.../internals/ConsumerDelegateCreator.java | 3 +-
.../consumer/internals/RequestManagers.java | 93 ++++++---
.../StreamsGroupHeartbeatRequestManager.java | 21 +-
.../consumer/internals/StreamsRebalanceData.java | 5 +-
.../internals/StreamsRebalanceEventsProcessor.java | 13 +-
.../events/ApplicationEventProcessor.java | 97 +++++++--
.../consumer/internals/AsyncKafkaConsumerTest.java | 12 +-
.../consumer/internals/RequestManagersTest.java | 4 +-
.../StreamsRebalanceEventsProcessorTest.java | 18 +-
.../events/ApplicationEventProcessorTest.java | 5 +-
.../SmokeTestDriverIntegrationTest.java | 44 +++-
.../org/apache/kafka/streams/GroupProtocol.java | 43 ++++
.../org/apache/kafka/streams/StreamsConfig.java | 24 +++
.../DefaultStreamsGroupRebalanceCallbacks.java | 130 ++++++++++++
.../streams/processor/internals/StreamThread.java | 229 ++++++++++++++++++++-
.../apache/kafka/streams/StreamsConfigTest.java | 20 ++
.../processor/internals/StreamThreadTest.java | 30 +++
.../kafka/streams/tests/SmokeTestClient.java | 4 +-
19 files changed, 733 insertions(+), 92 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 ba2e145d327..e421ef93191 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
@@ -280,10 +280,12 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
setGroupAssignmentSnapshot(partitions);
}
};
-
- AsyncKafkaConsumer(final ConsumerConfig config,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer) {
+
+ public AsyncKafkaConsumer(final ConsumerConfig config,
+ final Deserializer<K> keyDeserializer,
+ final Deserializer<V> valueDeserializer,
+ final Optional<StreamsRebalanceData> streamsRebalanceData,
+ final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor) {
this(
config,
keyDeserializer,
@@ -293,11 +295,14 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
CompletableEventReaper::new,
FetchCollector::new,
ConsumerMetadata::new,
- new LinkedBlockingQueue<>()
+ new LinkedBlockingQueue<>(),
+ streamsRebalanceData,
+ streamsRebalanceEventsProcessor
);
}
// Visible for testing
+ @SuppressWarnings({"this-escape"})
AsyncKafkaConsumer(final ConsumerConfig config,
final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer,
@@ -306,7 +311,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
final CompletableEventReaperFactory backgroundEventReaperFactory,
final FetchCollectorFactory<K, V> fetchCollectorFactory,
final ConsumerMetadataFactory metadataFactory,
- final LinkedBlockingQueue<BackgroundEvent> backgroundEventQueue) {
+ final LinkedBlockingQueue<BackgroundEvent> backgroundEventQueue,
+ final Optional<StreamsRebalanceData> streamsRebalanceData,
+ final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor) {
try {
GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
config,
@@ -382,7 +389,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
clientTelemetryReporter,
metrics,
offsetCommitCallbackInvoker,
- memberStateListener
+ memberStateListener,
+ streamsRebalanceData,
+ streamsRebalanceEventsProcessor
);
final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(logContext,
metadata,
@@ -398,6 +407,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
requestManagersSupplier,
kafkaConsumerMetrics
);
+ streamsRebalanceEventsProcessor.ifPresent(
+ processor -> processor.setApplicationEventHandler(applicationEventHandler)
+ );
this.rebalanceListenerInvoker = new ConsumerRebalanceListenerInvoker(
logContext,
@@ -568,7 +580,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
clientTelemetryReporter,
metrics,
offsetCommitCallbackInvoker,
- memberStateListener
+ memberStateListener,
+ Optional.empty(),
+ Optional.empty()
);
Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier = ApplicationEventProcessor.supplier(
logContext,
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 74592972b9d..60d12aec864 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
@@ -29,6 +29,7 @@ import org.apache.kafka.common.utils.Time;
import java.util.List;
import java.util.Locale;
+import java.util.Optional;
/**
* {@code ConsumerDelegateCreator} implements a quasi-factory pattern to allow the caller to remain unaware of the
@@ -60,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);
+ return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty(), 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 9b9fb48482b..9f406bd2090 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -54,10 +54,12 @@ public class RequestManagers implements Closeable {
public final Optional<ShareHeartbeatRequestManager> shareHeartbeatRequestManager;
public final Optional<ConsumerMembershipManager> consumerMembershipManager;
public final Optional<ShareMembershipManager> shareMembershipManager;
+ public final Optional<StreamsMembershipManager> streamsMembershipManager;
public final OffsetsRequestManager offsetsRequestManager;
public final TopicMetadataRequestManager topicMetadataRequestManager;
public final FetchRequestManager fetchRequestManager;
public final Optional<ShareConsumeRequestManager> shareConsumeRequestManager;
+ public final Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager;
private final List<Optional<? extends RequestManager>> entries;
private final IdempotentCloser closer = new IdempotentCloser();
@@ -68,7 +70,9 @@ public class RequestManagers implements Closeable {
Optional<CoordinatorRequestManager> coordinatorRequestManager,
Optional<CommitRequestManager> commitRequestManager,
Optional<ConsumerHeartbeatRequestManager> heartbeatRequestManager,
- Optional<ConsumerMembershipManager> membershipManager) {
+ Optional<ConsumerMembershipManager> membershipManager,
+ Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager,
+ Optional<StreamsMembershipManager> streamsMembershipManager) {
this.log = logContext.logger(RequestManagers.class);
this.offsetsRequestManager = requireNonNull(offsetsRequestManager, "OffsetsRequestManager cannot be null");
this.coordinatorRequestManager = coordinatorRequestManager;
@@ -78,7 +82,9 @@ public class RequestManagers implements Closeable {
this.shareConsumeRequestManager = Optional.empty();
this.consumerHeartbeatRequestManager = heartbeatRequestManager;
this.shareHeartbeatRequestManager = Optional.empty();
+ this.streamsGroupHeartbeatRequestManager = streamsGroupHeartbeatRequestManager;
this.consumerMembershipManager = membershipManager;
+ this.streamsMembershipManager = streamsMembershipManager;
this.shareMembershipManager = Optional.empty();
List<Optional<? extends RequestManager>> list = new ArrayList<>();
@@ -86,6 +92,8 @@ public class RequestManagers implements Closeable {
list.add(commitRequestManager);
list.add(heartbeatRequestManager);
list.add(membershipManager);
+ list.add(streamsGroupHeartbeatRequestManager);
+ list.add(streamsMembershipManager);
list.add(Optional.of(offsetsRequestManager));
list.add(Optional.of(topicMetadataRequestManager));
list.add(Optional.of(fetchRequestManager));
@@ -102,8 +110,10 @@ public class RequestManagers implements Closeable {
this.coordinatorRequestManager = coordinatorRequestManager;
this.commitRequestManager = Optional.empty();
this.consumerHeartbeatRequestManager = Optional.empty();
+ this.streamsGroupHeartbeatRequestManager = Optional.empty();
this.shareHeartbeatRequestManager = shareHeartbeatRequestManager;
this.consumerMembershipManager = Optional.empty();
+ this.streamsMembershipManager = Optional.empty();
this.shareMembershipManager = shareMembershipManager;
this.offsetsRequestManager = null;
this.topicMetadataRequestManager = null;
@@ -158,8 +168,10 @@ public class RequestManagers implements Closeable {
final Optional<ClientTelemetryReporter> clientTelemetryReporter,
final Metrics metrics,
final OffsetCommitCallbackInvoker offsetCommitCallbackInvoker,
- final MemberStateListener applicationThreadMemberStateListener
- ) {
+ final MemberStateListener applicationThreadMemberStateListener,
+ final Optional<StreamsRebalanceData> streamsRebalanceData,
+ final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor
+ ) {
return new CachedSupplier<>() {
@Override
protected RequestManagers create() {
@@ -187,26 +199,56 @@ public class RequestManagers implements Closeable {
ConsumerMembershipManager membershipManager = null;
CoordinatorRequestManager coordinator = null;
CommitRequestManager commitRequestManager = null;
+ StreamsGroupHeartbeatRequestManager streamsGroupHeartbeatRequestManager = null;
+ StreamsMembershipManager streamsMembershipManager = null;
if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
Optional<String> serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
coordinator = new CoordinatorRequestManager(
- logContext,
- retryBackoffMs,
- retryBackoffMaxMs,
- groupRebalanceConfig.groupId);
+ logContext,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ groupRebalanceConfig.groupId);
commitRequestManager = new CommitRequestManager(
+ time,
+ logContext,
+ subscriptions,
+ config,
+ coordinator,
+ offsetCommitCallbackInvoker,
+ groupRebalanceConfig.groupId,
+ groupRebalanceConfig.groupInstanceId,
+ metrics,
+ metadata);
+ if (streamsRebalanceEventsProcessor.isPresent()) {
+ streamsMembershipManager = new StreamsMembershipManager(
+ groupRebalanceConfig.groupId,
+ streamsRebalanceEventsProcessor.get(),
+ streamsRebalanceData.get(),
+ subscriptions,
+ logContext,
time,
+ metrics);
+ streamsMembershipManager.registerStateListener(commitRequestManager);
+ streamsMembershipManager.registerStateListener(applicationThreadMemberStateListener);
+
+ if (clientTelemetryReporter.isPresent()) {
+ clientTelemetryReporter.get()
+ .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, streamsMembershipManager.memberId()));
+ }
+
+ streamsGroupHeartbeatRequestManager = new StreamsGroupHeartbeatRequestManager(
logContext,
- subscriptions,
+ time,
config,
coordinator,
- offsetCommitCallbackInvoker,
- groupRebalanceConfig.groupId,
- groupRebalanceConfig.groupInstanceId,
+ streamsMembershipManager,
+ backgroundEventHandler,
metrics,
- metadata);
- membershipManager = new ConsumerMembershipManager(
+ streamsRebalanceData.get()
+ );
+ } else {
+ membershipManager = new ConsumerMembershipManager(
groupRebalanceConfig.groupId,
groupRebalanceConfig.groupInstanceId,
groupRebalanceConfig.rebalanceTimeoutMs,
@@ -220,17 +262,17 @@ public class RequestManagers implements Closeable {
metrics,
config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
- // Update the group member ID label in the client telemetry reporter.
- // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
- // Therefore, we can update the group member ID during initialization.
- if (clientTelemetryReporter.isPresent()) {
- clientTelemetryReporter.get()
- .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, membershipManager.memberId()));
- }
+ // Update the group member ID label in the client telemetry reporter.
+ // According to KIP-1082, the consumer will generate the member ID as the incarnation ID of the process.
+ // Therefore, we can update the group member ID during initialization.
+ if (clientTelemetryReporter.isPresent()) {
+ clientTelemetryReporter.get()
+ .updateMetricsLabels(Map.of(ClientTelemetryProvider.GROUP_MEMBER_ID, membershipManager.memberId()));
+ }
- membershipManager.registerStateListener(commitRequestManager);
- membershipManager.registerStateListener(applicationThreadMemberStateListener);
- heartbeatRequestManager = new ConsumerHeartbeatRequestManager(
+ membershipManager.registerStateListener(commitRequestManager);
+ membershipManager.registerStateListener(applicationThreadMemberStateListener);
+ heartbeatRequestManager = new ConsumerHeartbeatRequestManager(
logContext,
time,
config,
@@ -239,6 +281,7 @@ public class RequestManagers implements Closeable {
membershipManager,
backgroundEventHandler,
metrics);
+ }
}
final OffsetsRequestManager listOffsets = new OffsetsRequestManager(subscriptions,
@@ -261,7 +304,9 @@ public class RequestManagers implements Closeable {
Optional.ofNullable(coordinator),
Optional.ofNullable(commitRequestManager),
Optional.ofNullable(heartbeatRequestManager),
- Optional.ofNullable(membershipManager)
+ Optional.ofNullable(membershipManager),
+ Optional.ofNullable(streamsGroupHeartbeatRequestManager),
+ Optional.ofNullable(streamsMembershipManager)
);
}
};
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
index 319a708e216..55741114b34 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
@@ -74,13 +74,13 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
// Fields of StreamsGroupHeartbeatRequest sent in the most recent request
static class LastSentFields {
- private StreamsRebalanceData.Assignment assignment = StreamsRebalanceData.Assignment.EMPTY;
+ private StreamsRebalanceData.Assignment assignment = null;
LastSentFields() {
}
void reset() {
- assignment = StreamsRebalanceData.Assignment.EMPTY;
+ assignment = null;
}
}
@@ -402,6 +402,10 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
return EMPTY;
}
+ public StreamsMembershipManager membershipManager() {
+ return membershipManager;
+ }
+
/**
* Returns the delay for which the application thread can safely wait before it should be responsive
* to results from the request managers. For example, the subscription state can change when heartbeats
@@ -425,6 +429,17 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
return Math.min(pollTimer.remainingMs() / 2, heartbeatRequestState.timeToNextHeartbeatMs(currentTimeMs));
}
+ public void resetPollTimer(final long pollMs) {
+ pollTimer.update(pollMs);
+ if (pollTimer.isExpired()) {
+ logger.warn("Time between subsequent calls to poll() was longer than the configured " +
+ "max.poll.interval.ms, exceeded approximately by {} ms. Member {} will rejoin the group now.",
+ pollTimer.isExpiredBy(), membershipManager.memberId());
+ membershipManager.maybeRejoinStaleMember();
+ }
+ pollTimer.reset(maxPollIntervalMs);
+ }
+
/**
* A heartbeat should be sent without waiting for the heartbeat interval to expire if:
* - the member is leaving the group
@@ -508,7 +523,7 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
String statusDetails = statuses.stream()
.map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
.collect(Collectors.joining(", "));
- logger.warn("Membership is in the following statuses: {}.", statusDetails);
+ logger.warn("Membership is in the following statuses: {}", statusDetails);
}
membershipManager.onHeartbeatSuccess(response);
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
index 43e804d84bd..6157b66cf16 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java
@@ -75,10 +75,7 @@ public class StreamsRebalanceData {
@Override
public String toString() {
- return "TaskId{" +
- "subtopologyId=" + subtopologyId +
- ", partitionId=" + partitionId +
- '}';
+ return subtopologyId + "_" + partitionId;
}
}
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
index db91e8fcece..30bc7dcfb07 100644
--- 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
@@ -45,20 +45,17 @@ import java.util.concurrent.LinkedBlockingQueue;
public class StreamsRebalanceEventsProcessor {
private final BlockingQueue<BackgroundEvent> onCallbackRequests = new LinkedBlockingQueue<>();
- private ApplicationEventHandler applicationEventHandler = null;
- private final StreamsGroupRebalanceCallbacks rebalanceCallbacks;
private final StreamsRebalanceData streamsRebalanceData;
+ private ApplicationEventHandler applicationEventHandler;
+ private StreamsGroupRebalanceCallbacks rebalanceCallbacks;
/**
* Constructs the Streams rebalance processor.
*
* @param streamsRebalanceData
- * @param rebalanceCallbacks
*/
- public StreamsRebalanceEventsProcessor(StreamsRebalanceData streamsRebalanceData,
- StreamsGroupRebalanceCallbacks rebalanceCallbacks) {
+ public StreamsRebalanceEventsProcessor(StreamsRebalanceData streamsRebalanceData) {
this.streamsRebalanceData = streamsRebalanceData;
- this.rebalanceCallbacks = rebalanceCallbacks;
}
/**
@@ -107,6 +104,10 @@ public class StreamsRebalanceEventsProcessor {
this.applicationEventHandler = applicationEventHandler;
}
+ public void setRebalanceCallbacks(final StreamsGroupRebalanceCallbacks rebalanceCallbacks) {
+ this.rebalanceCallbacks = rebalanceCallbacks;
+ }
+
private void process(final BackgroundEvent event) {
switch (event.type()) {
case ERROR:
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
index 42dd1711b96..1ca51cca62e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
@@ -72,7 +72,7 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
this.metadataVersionSnapshot = metadata.updateVersion();
}
- @SuppressWarnings({"CyclomaticComplexity"})
+ @SuppressWarnings({"CyclomaticComplexity", "JavaNCSSCheck"})
@Override
public void process(ApplicationEvent event) {
switch (event.type()) {
@@ -200,6 +200,18 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
process((CurrentLagEvent) event);
return;
+ case STREAMS_ON_TASKS_REVOKED_CALLBACK_COMPLETED:
+ process((StreamsOnTasksRevokedCallbackCompletedEvent) event);
+ return;
+
+ case STREAMS_ON_TASKS_ASSIGNED_CALLBACK_COMPLETED:
+ process((StreamsOnTasksAssignedCallbackCompletedEvent) event);
+ return;
+
+ case STREAMS_ON_ALL_TASKS_LOST_CALLBACK_COMPLETED:
+ process((StreamsOnAllTasksLostCallbackCompletedEvent) event);
+ return;
+
default:
log.warn("Application event type {} was not expected", event.type());
}
@@ -220,6 +232,10 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
hrm.membershipManager().onConsumerPoll();
hrm.resetPollTimer(event.pollTimeMs());
});
+ requestManagers.streamsGroupHeartbeatRequestManager.ifPresent(hrm -> {
+ hrm.membershipManager().onConsumerPoll();
+ hrm.resetPollTimer(event.pollTimeMs());
+ });
} else {
// safe to unblock - no auto-commit risk here:
// 1. commitRequestManager is not present
@@ -320,22 +336,32 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
* it is already a member on the next poll.
*/
private void process(final TopicSubscriptionChangeEvent event) {
- if (requestManagers.consumerHeartbeatRequestManager.isEmpty()) {
+ if (requestManagers.consumerHeartbeatRequestManager.isPresent()) {
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener())) {
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ }
+ // Join the group if not already part of it, or just send the new subscription to the broker on the next poll.
+ requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ } else if (requestManagers.streamsGroupHeartbeatRequestManager.isPresent()) {
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener())) {
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+ }
+ requestManagers.streamsMembershipManager.get().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ } else {
log.warn("Group membership manager not present when processing a subscribe event");
event.future().complete(null);
- return;
}
- try {
- if (subscriptions.subscribe(event.topics(), event.listener()))
- this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
-
- // Join the group if not already part of it, or just send the new subscription to the broker on the next poll.
- requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
- event.future().complete(null);
- } catch (Exception e) {
- event.future().completeExceptionally(e);
- }
}
/**
@@ -405,6 +431,9 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
if (requestManagers.consumerHeartbeatRequestManager.isPresent()) {
CompletableFuture<Void> future = requestManagers.consumerHeartbeatRequestManager.get().membershipManager().leaveGroup();
future.whenComplete(complete(event.future()));
+ } else if (requestManagers.streamsGroupHeartbeatRequestManager.isPresent()) {
+ CompletableFuture<Void> future = requestManagers.streamsGroupHeartbeatRequestManager.get().membershipManager().leaveGroup();
+ future.whenComplete(complete(event.future()));
} else {
// If the consumer is not using the group management capabilities, we still need to clear all assignments it may have.
subscriptions.unsubscribe();
@@ -463,12 +492,15 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
}
private void process(final LeaveGroupOnCloseEvent event) {
- if (requestManagers.consumerMembershipManager.isEmpty())
- return;
-
- log.debug("Signal the ConsumerMembershipManager to leave the consumer group since the consumer is closing");
- CompletableFuture<Void> future = requestManagers.consumerMembershipManager.get().leaveGroupOnClose(event.membershipOperation());
- future.whenComplete(complete(event.future()));
+ if (requestManagers.consumerMembershipManager.isPresent()) {
+ log.debug("Signal the ConsumerMembershipManager to leave the consumer group since the consumer is closing");
+ CompletableFuture<Void> future = requestManagers.consumerMembershipManager.get().leaveGroupOnClose(event.membershipOperation());
+ future.whenComplete(complete(event.future()));
+ } else if (requestManagers.streamsMembershipManager.isPresent()) {
+ log.debug("Signal the StreamsMembershipManager to leave the Streams group since the member is closing");
+ CompletableFuture<Void> future = requestManagers.streamsMembershipManager.get().leaveGroupOnClose();
+ future.whenComplete(complete(event.future()));
+ }
}
private void process(@SuppressWarnings("unused") final StopFindCoordinatorOnCloseEvent event) {
@@ -667,6 +699,33 @@ public class ApplicationEventProcessor implements EventProcessor<ApplicationEven
}
}
+ private void process(final StreamsOnTasksRevokedCallbackCompletedEvent event) {
+ if (requestManagers.streamsMembershipManager.isEmpty()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onTasksRevoked callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onTasksRevokedCallbackCompleted(event);
+ }
+
+ private void process(final StreamsOnTasksAssignedCallbackCompletedEvent event) {
+ if (requestManagers.streamsMembershipManager.isEmpty()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onTasksAssigned callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onTasksAssignedCallbackCompleted(event);
+ }
+
+ private void process(final StreamsOnAllTasksLostCallbackCompletedEvent event) {
+ if (requestManagers.streamsMembershipManager.isEmpty()) {
+ log.warn("An internal error occurred; the Streams membership manager was not present, so the notification " +
+ "of the onAllTasksLost callback execution could not be sent");
+ return;
+ }
+ requestManagers.streamsMembershipManager.get().onAllTasksLostCallbackCompleted(event);
+ }
+
private <T> BiConsumer<? super T, ? super Throwable> complete(final CompletableFuture<T> b) {
return (value, exception) -> {
if (exception != null)
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index e501f3aeea3..df091bfccb6 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
@@ -219,7 +219,9 @@ public class AsyncKafkaConsumerTest {
a -> backgroundEventReaper,
(a, b, c, d, e, f, g) -> fetchCollector,
(a, b, c, d) -> metadata,
- backgroundEventQueue
+ backgroundEventQueue,
+ Optional.empty(),
+ Optional.empty()
);
}
@@ -233,7 +235,9 @@ public class AsyncKafkaConsumerTest {
a -> backgroundEventReaper,
(a, b, c, d, e, f, g) -> fetchCollector,
(a, b, c, d) -> metadata,
- backgroundEventQueue
+ backgroundEventQueue,
+ Optional.empty(),
+ Optional.empty()
);
}
@@ -1315,7 +1319,9 @@ public class AsyncKafkaConsumerTest {
any(),
any(),
any(),
- applicationThreadMemberStateListener.capture()
+ 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 405ecabcf16..ef3ba79dbf5 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
@@ -62,7 +62,9 @@ public class RequestManagersTest {
Optional.empty(),
new Metrics(),
mock(OffsetCommitCallbackInvoker.class),
- listener
+ listener,
+ Optional.empty(),
+ Optional.empty()
).get();
requestManagers.consumerMembershipManager.ifPresent(
membershipManager -> assertTrue(membershipManager.stateListeners().contains(listener))
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
index 6c7b164826b..5bc988cf9c7 100644
--- 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
@@ -74,7 +74,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldInvokeOnTasksAssignedCallback() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
@@ -109,7 +110,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldReThrowErrorFromOnTasksAssignedCallbackAndPassErrorToBackground() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
@@ -148,7 +150,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldInvokeOnTasksRevokedCallback() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
@@ -172,7 +175,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldReThrowErrorFromOnTasksRevokedCallbackAndPassErrorToBackground() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
@@ -199,7 +203,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldInvokeOnAllTasksLostCallback() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
@@ -238,7 +243,8 @@ public class StreamsRebalanceEventsProcessorTest {
@Test
public void shouldReThrowErrorFromOnAllTasksLostCallbackAndPassErrorToBackground() {
final StreamsRebalanceEventsProcessor rebalanceEventsProcessor =
- new StreamsRebalanceEventsProcessor(rebalanceData, rebalanceCallbacks);
+ new StreamsRebalanceEventsProcessor(rebalanceData);
+ rebalanceEventsProcessor.setRebalanceCallbacks(rebalanceCallbacks);
rebalanceEventsProcessor.setApplicationEventHandler(applicationEventHandler);
final Set<StreamsRebalanceData.TaskId> activeTasks = Set.of(
new StreamsRebalanceData.TaskId(SUBTOPOLOGY_0, 0),
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index 75d696806a4..9cd306a9be1 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -97,7 +97,10 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(mock(CoordinatorRequestManager.class)) : Optional.empty(),
withGroupId ? Optional.of(commitRequestManager) : Optional.empty(),
withGroupId ? Optional.of(heartbeatRequestManager) : Optional.empty(),
- withGroupId ? Optional.of(membershipManager) : Optional.empty());
+ withGroupId ? Optional.of(membershipManager) : Optional.empty(),
+ Optional.empty(),
+ Optional.empty()
+ );
processor = new ApplicationEventProcessor(
new LogContext(),
requestManagers,
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 aaf2af7b1cc..01e899ecc21 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
@@ -18,6 +18,9 @@ package org.apache.kafka.streams.integration;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ServerConfigs;
+import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
@@ -27,7 +30,9 @@ import org.apache.kafka.streams.tests.SmokeTestDriver;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
@@ -35,29 +40,40 @@ import org.junit.jupiter.params.provider.CsvSource;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import static org.apache.kafka.streams.tests.SmokeTestDriver.generate;
import static org.apache.kafka.streams.tests.SmokeTestDriver.verify;
+import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
import static org.junit.jupiter.api.Assertions.assertTrue;
@Timeout(600)
@Tag("integration")
public class SmokeTestDriverIntegrationTest {
- public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);
+ public static EmbeddedKafkaCluster cluster;
+ public TestInfo testInfo;
@BeforeAll
public static void startCluster() throws IOException {
- CLUSTER.start();
+ final Properties props = new Properties();
+ props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
+ props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
+ cluster = new EmbeddedKafkaCluster(3, props);
+ cluster.start();
}
@AfterAll
public static void closeCluster() {
- CLUSTER.stop();
+ cluster.stop();
}
+ @BeforeEach
+ public void setUp(final TestInfo testInfo) {
+ this.testInfo = testInfo;
+ }
private static class Driver extends Thread {
private final String bootstrapServers;
@@ -99,8 +115,17 @@ public class SmokeTestDriverIntegrationTest {
// We set 2 timeout condition to fail the test before passing the verification:
// (1) 10 min timeout, (2) 30 tries of polling without getting any data
@ParameterizedTest
- @CsvSource({"false, false", "true, false"})
- public void shouldWorkWithRebalance(final boolean stateUpdaterEnabled, final boolean processingThreadsEnabled) throws InterruptedException {
+ @CsvSource({
+ "false, false, true",
+ "true, false, true",
+ "false, false, false",
+ "true, false, false",
+ })
+ public void shouldWorkWithRebalance(
+ final boolean stateUpdaterEnabled,
+ final boolean processingThreadsEnabled,
+ final boolean streamsProtocolEnabled
+ ) throws InterruptedException {
Exit.setExitProcedure((statusCode, message) -> {
throw new AssertionError("Test called exit(). code:" + statusCode + " message:" + message);
});
@@ -110,9 +135,9 @@ public class SmokeTestDriverIntegrationTest {
int numClientsCreated = 0;
final ArrayList<SmokeTestClient> clients = new ArrayList<>();
- IntegrationTestUtils.cleanStateBeforeTest(CLUSTER, SmokeTestDriver.topics());
+ IntegrationTestUtils.cleanStateBeforeTest(cluster, SmokeTestDriver.topics());
- final String bootstrapServers = CLUSTER.bootstrapServers();
+ final String bootstrapServers = cluster.bootstrapServers();
final Driver driver = new Driver(bootstrapServers, 10, 1000);
driver.start();
System.out.println("started driver");
@@ -120,10 +145,15 @@ public class SmokeTestDriverIntegrationTest {
final Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, safeUniqueTestName(testInfo));
props.put(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled);
props.put(InternalConfig.PROCESSING_THREADS_ENABLED, processingThreadsEnabled);
// decrease the session timeout so that we can trigger the rebalance soon after old client left closed
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
+ props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 500);
+ if (streamsProtocolEnabled) {
+ props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+ }
// cycle out Streams instances as long as the test is running.
while (driver.isAlive()) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/GroupProtocol.java b/streams/src/main/java/org/apache/kafka/streams/GroupProtocol.java
new file mode 100644
index 00000000000..146a5e6e9de
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/GroupProtocol.java
@@ -0,0 +1,43 @@
+/*
+ * 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.streams;
+
+import java.util.Locale;
+
+public enum GroupProtocol {
+ /** Classic group protocol. */
+ CLASSIC("CLASSIC"),
+
+ /** Streams group protocol */
+ STREAMS("STREAMS");
+
+ /**
+ * String representation of the group protocol.
+ */
+ public final String name;
+
+ GroupProtocol(final String name) {
+ this.name = name;
+ }
+
+ /**
+ * Case-insensitive group protocol lookup by string name.
+ */
+ public static GroupProtocol of(final String name) {
+ return GroupProtocol.valueOf(name.toUpperCase(Locale.ROOT));
+ }
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index b98806590a1..e19cb03b207 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -68,6 +68,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
@@ -608,6 +609,16 @@ public class StreamsConfig extends AbstractConfig {
" topics (e.g., changelog and repartition topics) and their associated state stores." +
" When enabled, the application will refuse to start if any internal resource has an auto-generated name.";
+ /**
+ * <code>group.protocol</code>
+ */
+ public static final String GROUP_PROTOCOL_CONFIG = "group.protocol";
+ public static final String DEFAULT_GROUP_PROTOCOL = GroupProtocol.CLASSIC.name().toLowerCase(
+ Locale.ROOT);
+ private static final String GROUP_PROTOCOL_DOC = "The group protocol streams should use. We currently " +
+ "support \"classic\" or \"streams\". If \"streams\" is specified, then the streams rebalance protocol will be " +
+ "used. Otherwise, the classic group protocol will be used.";
+
/** {@code log.summary.interval.ms} */
public static final String LOG_SUMMARY_INTERVAL_MS_CONFIG = "log.summary.interval.ms";
private static final String LOG_SUMMARY_INTERVAL_MS_DOC = "The output interval in milliseconds for logging summary information.\n" +
@@ -1042,6 +1053,12 @@ public class StreamsConfig extends AbstractConfig {
TOPOLOGY_OPTIMIZATION_CONFIGS::toString),
Importance.MEDIUM,
TOPOLOGY_OPTIMIZATION_DOC)
+ .define(GROUP_PROTOCOL_CONFIG,
+ Type.STRING,
+ DEFAULT_GROUP_PROTOCOL,
+ ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(GroupProtocol.class)),
+ Importance.MEDIUM,
+ GROUP_PROTOCOL_DOC)
// LOW
@@ -1505,6 +1522,11 @@ public class StreamsConfig extends AbstractConfig {
}
verifyTopologyOptimizationConfigs(getString(TOPOLOGY_OPTIMIZATION_CONFIG));
verifyClientTelemetryConfigs();
+
+ if (doLog && getString(GROUP_PROTOCOL_CONFIG).equals(GroupProtocol.STREAMS.name().toLowerCase(Locale.ROOT))) {
+ log.warn("The streams rebalance protocol is still in development and should not be used in production. "
+ + "Please set group.protocol=classic (default) in all production use cases.");
+ }
}
private void verifyEOSTransactionTimeoutCompatibility() {
@@ -1627,6 +1649,8 @@ public class StreamsConfig extends AbstractConfig {
private Map<String, Object> getCommonConsumerConfigs() {
final Map<String, Object> clientProvidedProps = getClientPropsWithPrefix(CONSUMER_PREFIX, ConsumerConfig.configNames());
+ clientProvidedProps.remove(GROUP_PROTOCOL_CONFIG);
+
checkIfUnexpectedUserSpecifiedConsumerConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS);
checkIfUnexpectedUserSpecifiedConsumerConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_EOS_CONFIGS);
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/DefaultStreamsGroupRebalanceCallbacks.java
new file mode 100644
index 00000000000..37e61427ccd
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStreamsGroupRebalanceCallbacks.java
@@ -0,0 +1,130 @@
+/*
+ * 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.streams.processor.internals;
+
+import org.apache.kafka.clients.consumer.internals.StreamsGroupRebalanceCallbacks;
+import org.apache.kafka.clients.consumer.internals.StreamsRebalanceData;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.processor.TaskId;
+
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class DefaultStreamsGroupRebalanceCallbacks implements StreamsGroupRebalanceCallbacks {
+
+ private final Logger log;
+ private final Time time;
+ private final StreamsRebalanceData streamsRebalanceData;
+ 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) {
+ this.log = log;
+ this.time = time;
+ this.streamsRebalanceData = streamsRebalanceData;
+ this.streamThread = streamThread;
+ this.taskManager = taskManager;
+ }
+
+ @Override
+ public Optional<Exception> onTasksRevoked(final Set<StreamsRebalanceData.TaskId> tasks) {
+ try {
+ final Map<TaskId, Set<TopicPartition>> activeTasksToRevokeWithPartitions =
+ pairWithTopicPartitions(tasks.stream());
+ final Set<TopicPartition> partitionsToRevoke = activeTasksToRevokeWithPartitions.values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toSet());
+
+ final long start = time.milliseconds();
+ try {
+ log.info("Revoking active tasks {}.", tasks);
+ taskManager.handleRevocation(partitionsToRevoke);
+ } finally {
+ log.info("partition revocation took {} ms.", time.milliseconds() - start);
+ }
+ if (streamThread.state() != StreamThread.State.PENDING_SHUTDOWN) {
+ streamThread.setState(StreamThread.State.PARTITIONS_REVOKED);
+ }
+ } catch (final Exception exception) {
+ return Optional.of(exception);
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<Exception> onTasksAssigned(final StreamsRebalanceData.Assignment assignment) {
+ try {
+ final Map<TaskId, Set<TopicPartition>> activeTasksWithPartitions =
+ pairWithTopicPartitions(assignment.activeTasks().stream());
+ final Map<TaskId, Set<TopicPartition>> standbyTasksWithPartitions =
+ pairWithTopicPartitions(Stream.concat(assignment.standbyTasks().stream(), assignment.warmupTasks().stream()));
+
+ log.info("Processing new assignment {} from Streams Rebalance Protocol", assignment);
+
+ taskManager.handleAssignment(activeTasksWithPartitions, standbyTasksWithPartitions);
+ streamThread.setState(StreamThread.State.PARTITIONS_ASSIGNED);
+ taskManager.handleRebalanceComplete();
+ } catch (final Exception exception) {
+ return Optional.of(exception);
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<Exception> onAllTasksLost() {
+ try {
+ taskManager.handleLostAll();
+ } catch (final Exception exception) {
+ return Optional.of(exception);
+ }
+ return Optional.empty();
+ }
+
+ private Map<TaskId, Set<TopicPartition>> pairWithTopicPartitions(final Stream<StreamsRebalanceData.TaskId> taskIdStream) {
+ return taskIdStream
+ .collect(Collectors.toMap(
+ this::toTaskId,
+ task -> toTopicPartitions(task, streamsRebalanceData.subtopologies().get(task.subtopologyId()))
+ ));
+ }
+
+ private TaskId toTaskId(final StreamsRebalanceData.TaskId task) {
+ return new TaskId(Integer.parseInt(task.subtopologyId()), task.partitionId());
+ }
+
+ private Set<TopicPartition> toTopicPartitions(final StreamsRebalanceData.TaskId task,
+ final StreamsRebalanceData.Subtopology subTopology) {
+ return
+ Stream.concat(
+ subTopology.sourceTopics().stream(),
+ subTopology.repartitionSourceTopics().keySet().stream()
+ )
+ .map(t -> new TopicPartition(t, task.partitionId()))
+ .collect(Collectors.toSet());
+ }
+}
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 74e7a27ceb9..a96d968ca27 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
@@ -25,11 +25,16 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.InvalidOffsetException;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
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;
import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.TimeoutException;
@@ -40,6 +45,7 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig.InternalConfig;
@@ -60,12 +66,14 @@ import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.processor.internals.metrics.ThreadMetrics;
import org.apache.kafka.streams.processor.internals.tasks.DefaultTaskManager;
import org.apache.kafka.streams.processor.internals.tasks.DefaultTaskManager.DefaultTaskExecutorCreator;
+import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.internals.ThreadCache;
import org.slf4j.Logger;
import java.time.Duration;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -86,6 +94,7 @@ import static org.apache.kafka.streams.processor.internals.ClientUtils.adminClie
import static org.apache.kafka.streams.processor.internals.ClientUtils.consumerClientId;
import static org.apache.kafka.streams.processor.internals.ClientUtils.restoreConsumerClientId;
+@SuppressWarnings("ClassDataAbstractionCoupling")
public class StreamThread extends Thread implements ProcessingThread {
private static final String THREAD_ID_SUBSTRING = "-StreamThread-";
@@ -343,6 +352,10 @@ public class StreamThread extends Thread implements ProcessingThread {
// handler for, eg MissingSourceTopicException with named topologies
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
private final AtomicLong cacheResizeSize = new AtomicLong(-1L);
private final AtomicBoolean leaveGroupRequested = new AtomicBoolean(false);
@@ -478,18 +491,19 @@ public class StreamThread extends Thread implements ProcessingThread {
consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
}
- final Consumer<byte[], byte[]> mainConsumer = clientSupplier.getConsumer(consumerConfigs);
- taskManager.setMainConsumer(mainConsumer);
- referenceContainer.mainConsumer = mainConsumer;
+ final MainConsumerSetup mainConsumerSetup = setupMainConsumer(topologyMetadata, config, clientSupplier, processId, log, consumerConfigs);
+
+ taskManager.setMainConsumer(mainConsumerSetup.mainConsumer);
+ referenceContainer.mainConsumer = mainConsumerSetup.mainConsumer;
- final StreamsThreadMetricsDelegatingReporter reporter = new StreamsThreadMetricsDelegatingReporter(mainConsumer, threadId, Optional.of(stateUpdaterId));
+ final StreamsThreadMetricsDelegatingReporter reporter = new StreamsThreadMetricsDelegatingReporter(mainConsumerSetup.mainConsumer, threadId, Optional.of(stateUpdaterId));
streamsMetrics.metricsRegistry().addReporter(reporter);
final StreamThread streamThread = new StreamThread(
time,
config,
adminClient,
- mainConsumer,
+ mainConsumerSetup.mainConsumer,
restoreConsumer,
changelogReader,
originalReset,
@@ -505,12 +519,73 @@ public class StreamThread extends Thread implements ProcessingThread {
referenceContainer.nonFatalExceptionsToHandle,
shutdownErrorHook,
streamsUncaughtExceptionHandler,
- cache::resize
+ cache::resize,
+ mainConsumerSetup.streamsRebalanceData,
+ mainConsumerSetup.streamsRebalanceEventsProcessor,
+ streamsMetadataState
);
return streamThread.updateThreadMetadata(adminClientId(clientId));
}
+ private static MainConsumerSetup setupMainConsumer(final TopologyMetadata topologyMetadata,
+ final StreamsConfig config,
+ final KafkaClientSupplier clientSupplier,
+ final UUID processId,
+ final Logger log,
+ final Map<String, Object> consumerConfigs) {
+ if (config.getString(StreamsConfig.GROUP_PROTOCOL_CONFIG).equalsIgnoreCase(GroupProtocol.STREAMS.name)) {
+ if (topologyMetadata.hasNamedTopologies()) {
+ throw new IllegalStateException("Named topologies and the CONSUMER protocol cannot be used at the same time.");
+ }
+ log.info("Streams rebalance protocol enabled");
+
+ final Optional<StreamsRebalanceData> streamsRebalanceData = Optional.of(
+ initStreamsRebalanceData(
+ processId,
+ config,
+ parseHostInfo(config.getString(StreamsConfig.APPLICATION_SERVER_CONFIG)),
+ topologyMetadata
+ )
+ );
+ final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor =
+ Optional.of(new StreamsRebalanceEventsProcessor(streamsRebalanceData.get()));
+ final ByteArrayDeserializer keyDeserializer = new ByteArrayDeserializer();
+ final ByteArrayDeserializer valueDeserializer = new ByteArrayDeserializer();
+ return new MainConsumerSetup(
+ new AsyncKafkaConsumer<>(
+ new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(consumerConfigs, keyDeserializer, valueDeserializer)),
+ keyDeserializer,
+ valueDeserializer,
+ streamsRebalanceData,
+ streamsRebalanceEventsProcessor
+ ),
+ streamsRebalanceData,
+ streamsRebalanceEventsProcessor
+ );
+ } else {
+ return new MainConsumerSetup(
+ clientSupplier.getConsumer(consumerConfigs),
+ Optional.empty(),
+ Optional.empty()
+ );
+ }
+ }
+
+ 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) {
+ this.mainConsumer = mainConsumer;
+ this.streamsRebalanceData = streamsRebalanceData;
+ this.streamsRebalanceEventsProcessor = streamsRebalanceEventsProcessor;
+ }
+ }
+
private static DefaultTaskManager maybeCreateSchedulingTaskManager(final boolean processingThreadsEnabled,
final boolean stateUpdaterEnabled,
final TopologyMetadata topologyMetadata,
@@ -563,7 +638,98 @@ public class StreamThread extends Thread implements ProcessingThread {
}
}
- @SuppressWarnings("this-escape")
+ private static Optional<StreamsRebalanceData.HostInfo> parseHostInfo(final String endpoint) {
+ final HostInfo hostInfo = HostInfo.buildFromEndpoint(endpoint);
+ if (hostInfo == null) {
+ return Optional.empty();
+ } else {
+ return Optional.of(new StreamsRebalanceData.HostInfo(hostInfo.host(), hostInfo.port()));
+ }
+ }
+
+ private static StreamsRebalanceData initStreamsRebalanceData(final UUID processId,
+ final StreamsConfig config,
+ final Optional<StreamsRebalanceData.HostInfo> endpoint,
+ final TopologyMetadata topologyMetadata) {
+ final InternalTopologyBuilder internalTopologyBuilder = topologyMetadata.lookupBuilderForNamedTopology(null);
+
+ final Map<String, StreamsRebalanceData.Subtopology> subtopologies = initBrokerTopology(config, internalTopologyBuilder);
+
+ return new StreamsRebalanceData(
+ processId,
+ endpoint,
+ subtopologies,
+ config.getClientTags()
+ );
+ }
+
+ private static Map<String, StreamsRebalanceData.Subtopology> initBrokerTopology(final StreamsConfig config,
+ final InternalTopologyBuilder internalTopologyBuilder) {
+ final Map<String, String> defaultTopicConfigs = new HashMap<>();
+ for (final Map.Entry<String, Object> entry : config.originalsWithPrefix(StreamsConfig.TOPIC_PREFIX).entrySet()) {
+ if (entry.getValue() != null) {
+ defaultTopicConfigs.put(entry.getKey(), entry.getValue().toString());
+ }
+ }
+ final long windowChangeLogAdditionalRetention = config.getLong(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG);
+
+ final Map<String, StreamsRebalanceData.Subtopology> subtopologies = new HashMap<>();
+ final Collection<Set<String>> copartitionGroups = internalTopologyBuilder.copartitionGroups();
+
+ final Set<String> allRepartitionSourceTopics = internalTopologyBuilder.subtopologyToTopicsInfo().values().stream()
+ .flatMap(t -> t.repartitionSourceTopics.keySet().stream())
+ .collect(Collectors.toSet());
+
+ for (final Map.Entry<TopologyMetadata.Subtopology, InternalTopologyBuilder.TopicsInfo> topicsInfoEntry : internalTopologyBuilder.subtopologyToTopicsInfo()
+ .entrySet()) {
+
+ final HashSet<String> allSourceTopics = new HashSet<>(topicsInfoEntry.getValue().sourceTopics);
+ topicsInfoEntry.getValue().repartitionSourceTopics.forEach(
+ (repartitionSourceTopic, repartitionTopicInfo) -> {
+ allSourceTopics.add(repartitionSourceTopic);
+ });
+
+ final Set<String> sourceTopics = topicsInfoEntry.getValue().sourceTopics.stream()
+ .filter(topic -> !topicsInfoEntry.getValue().repartitionSourceTopics.containsKey(topic))
+ .collect(Collectors.toSet());
+ final Set<String> repartitionSinkTopics = topicsInfoEntry.getValue().sinkTopics.stream()
+ .filter(allRepartitionSourceTopics::contains)
+ .collect(Collectors.toSet());
+ final Map<String, StreamsRebalanceData.TopicInfo> repartitionSourceTopics = topicsInfoEntry.getValue().repartitionSourceTopics.entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e ->
+ new StreamsRebalanceData.TopicInfo(e.getValue().numberOfPartitions(),
+ Optional.of(config.getInt(StreamsConfig.REPLICATION_FACTOR_CONFIG).shortValue()),
+ e.getValue().properties(defaultTopicConfigs, windowChangeLogAdditionalRetention))));
+ final Map<String, StreamsRebalanceData.TopicInfo> stateChangelogTopics = topicsInfoEntry.getValue().stateChangelogTopics.entrySet()
+ .stream()
+ .collect(Collectors.toMap(Map.Entry::getKey, e ->
+ new StreamsRebalanceData.TopicInfo(e.getValue().numberOfPartitions(),
+ Optional.of(config.getInt(StreamsConfig.REPLICATION_FACTOR_CONFIG).shortValue()),
+ e.getValue().properties(defaultTopicConfigs, windowChangeLogAdditionalRetention))));
+
+ subtopologies.put(
+ String.valueOf(topicsInfoEntry.getKey().nodeGroupId),
+ new StreamsRebalanceData.Subtopology(
+ sourceTopics,
+ repartitionSinkTopics,
+ repartitionSourceTopics,
+ stateChangelogTopics,
+ copartitionGroups.stream().filter(allSourceTopics::containsAll).collect(Collectors.toList())
+ )
+ );
+ }
+
+ if (subtopologies.values().stream().mapToInt(x -> x.copartitionGroups().size()).sum()
+ != copartitionGroups.size()) {
+ throw new IllegalStateException(
+ "Not all copartition groups were converted to broker topology");
+ }
+
+ return subtopologies;
+ }
+
+ @SuppressWarnings({"this-escape"})
public StreamThread(final Time time,
final StreamsConfig config,
final Admin adminClient,
@@ -583,7 +749,10 @@ public class StreamThread extends Thread implements ProcessingThread {
final Queue<StreamsException> nonFatalExceptionsToHandle,
final Runnable shutdownErrorHook,
final BiConsumer<Throwable, Boolean> streamsUncaughtExceptionHandler,
- final java.util.function.Consumer<Long> cacheResizer
+ final java.util.function.Consumer<Long> cacheResizer,
+ final Optional<StreamsRebalanceData> streamsRebalanceData,
+ final Optional<StreamsRebalanceEventsProcessor> streamsRebalanceEventsProcessor,
+ final StreamsMetadataState streamsMetadataState
) {
super(threadId);
this.stateLock = new Object();
@@ -666,6 +835,15 @@ public class StreamThread extends Thread implements ProcessingThread {
this.stateUpdaterEnabled = InternalConfig.stateUpdaterEnabled(config.originals());
this.processingThreadsEnabled = InternalConfig.processingThreadsEnabled(config.originals());
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;
}
private static final class InternalConsumerConfig extends ConsumerConfig {
@@ -961,6 +1139,8 @@ public class StreamThread extends Thread implements ProcessingThread {
final long startMs = time.milliseconds();
now = startMs;
+ maybeHandleAssignmentFromStreamsRebalanceProtocol();
+
final long pollLatency;
taskManager.resumePollingForPartitionsWithAvailableSpace();
pollLatency = pollPhase();
@@ -1296,6 +1476,39 @@ public class StreamThread extends Thread implements ProcessingThread {
return records;
}
+ public void maybeHandleAssignmentFromStreamsRebalanceProtocol() {
+ 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();
+ }
+ }
+
+ static Map<TopicPartition, PartitionInfo> getTopicPartitionInfo(final Map<HostInfo, Set<TopicPartition>> partitionsByHost) {
+ final Map<TopicPartition, PartitionInfo> topicToPartitionInfo = new HashMap<>();
+ for (final Set<TopicPartition> value : partitionsByHost.values()) {
+ for (final TopicPartition topicPartition : value) {
+ topicToPartitionInfo.put(
+ topicPartition,
+ new PartitionInfo(
+ topicPartition.topic(),
+ topicPartition.partition(),
+ null,
+ new Node[0],
+ new Node[0]
+ )
+ );
+ }
+ }
+ return topicToPartitionInfo;
+ }
+
private void resetOffsets(final Set<TopicPartition> partitions, final Exception cause) {
final Set<String> loggedTopics = new HashSet<>();
final Set<TopicPartition> seekToBeginning = new HashSet<>();
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index 7e9447851e1..a7f657ec54b 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -71,6 +71,7 @@ import static org.apache.kafka.streams.StreamsConfig.DSL_STORE_SUPPLIERS_CLASS_C
import static org.apache.kafka.streams.StreamsConfig.ENABLE_METRICS_PUSH_CONFIG;
import static org.apache.kafka.streams.StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG;
import static org.apache.kafka.streams.StreamsConfig.EXACTLY_ONCE_V2;
+import static org.apache.kafka.streams.StreamsConfig.GROUP_PROTOCOL_CONFIG;
import static org.apache.kafka.streams.StreamsConfig.MAX_RACK_AWARE_ASSIGNMENT_TAG_KEY_LENGTH;
import static org.apache.kafka.streams.StreamsConfig.MAX_RACK_AWARE_ASSIGNMENT_TAG_VALUE_LENGTH;
import static org.apache.kafka.streams.StreamsConfig.PROCESSOR_WRAPPER_CLASS_CONFIG;
@@ -1596,6 +1597,25 @@ public class StreamsConfigTest {
assertTrue(streamsConfig.getBoolean(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG));
}
+ @Test
+ public void shouldSetGroupProtocolToClassicByDefault() {
+ assertTrue(GroupProtocol.CLASSIC.name().equalsIgnoreCase(streamsConfig.getString(GROUP_PROTOCOL_CONFIG)));
+ }
+
+ @Test
+ public void shouldSetGroupProtocolToClassic() {
+ props.put(GROUP_PROTOCOL_CONFIG, GroupProtocol.CLASSIC.name());
+ streamsConfig = new StreamsConfig(props);
+ assertTrue(GroupProtocol.CLASSIC.name().equalsIgnoreCase(streamsConfig.getString(GROUP_PROTOCOL_CONFIG)));
+ }
+
+ @Test
+ public void shouldSetGroupProtocolToStreams() {
+ props.put(GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streamsConfig = new StreamsConfig(props);
+ assertTrue(GroupProtocol.STREAMS.name().equalsIgnoreCase(streamsConfig.getString(GROUP_PROTOCOL_CONFIG)));
+ }
+
static class MisconfiguredSerde implements Serde<Object> {
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
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 5d32efe8329..fa72b116f00 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
@@ -1447,6 +1447,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
).updateThreadMetadata(adminClientId(CLIENT_ID));
@@ -2666,6 +2669,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -2725,6 +2731,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -2793,6 +2802,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -2857,6 +2869,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -2918,6 +2933,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -3150,6 +3168,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
);
final MetricName testMetricName = new MetricName("test_metric", "", "", new HashMap<>());
@@ -3207,6 +3228,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
(e, b) -> { },
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
) {
@Override
@@ -3587,6 +3611,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
null,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
);
}
@@ -3709,6 +3736,9 @@ public class StreamThreadTest {
new LinkedList<>(),
null,
HANDLER,
+ null,
+ Optional.empty(),
+ Optional.empty(),
null
);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
index bf53c76921b..5bc679245d7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
+++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java
@@ -129,7 +129,9 @@ public class SmokeTestClient extends SmokeTestUtil {
private Properties getStreamsConfig(final Properties props) {
final Properties fullProps = new Properties(props);
- fullProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "SmokeTest");
+ if (!props.containsKey(StreamsConfig.APPLICATION_ID_CONFIG)) {
+ fullProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "SmokeTest");
+ }
fullProps.put(StreamsConfig.CLIENT_ID_CONFIG, "SmokeTest-" + name);
fullProps.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
fullProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);