You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2024/11/28 16:59:16 UTC
(kafka) 37/42: Resolve conflicts from 11/15 trunk rebase - Initialize topologies as part of heartbeat (#17695)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch kip1071
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 6b98ec67ee34e283f50957728784d0103699e4dc
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Wed Nov 6 15:49:06 2024 +0100
Resolve conflicts from 11/15 trunk rebase - Initialize topologies as part of heartbeat (#17695)
We merge the initialization of the Streams group into the heartbeat of the Streams group.
As a result the first heartbeat of a member (member epoch is zero) contains the topology
metadata required for initialization.
---
.../consumer/internals/RequestManagers.java | 17 +-
.../internals/StreamsAssignmentInterface.java | 8 +-
.../StreamsGroupHeartbeatRequestManager.java | 116 +++++-
.../StreamsGroupInitializeRequestManager.java | 210 ----------
.../org/apache/kafka/common/protocol/ApiKeys.java | 1 -
.../kafka/common/requests/AbstractRequest.java | 2 -
.../kafka/common/requests/AbstractResponse.java | 2 -
.../requests/StreamsGroupHeartbeatResponse.java | 1 +
.../requests/StreamsGroupInitializeRequest.java | 77 ----
.../requests/StreamsGroupInitializeResponse.java | 71 ----
.../message/StreamsGroupDescribeRequest.json | 2 +-
.../message/StreamsGroupDescribeResponse.json | 6 +-
.../message/StreamsGroupHeartbeatRequest.json | 65 ++-
.../message/StreamsGroupHeartbeatResponse.json | 15 +-
.../message/StreamsGroupInitializeRequest.json | 77 ----
.../message/StreamsGroupInitializeResponse.json | 42 --
.../StreamsGroupHeartbeatRequestManagerTest.java | 84 +++-
.../StreamsGroupInitializeRequestManagerTest.java | 152 -------
.../events/ApplicationEventProcessorTest.java | 1 -
.../kafka/common/requests/RequestResponseTest.java | 12 -
.../group/GroupCoordinatorAdapter.scala | 15 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 131 +++---
.../group/GroupCoordinatorAdapterTest.scala | 18 +-
.../scala/unit/kafka/server/KafkaApisTest.scala | 194 +++++++--
.../scala/unit/kafka/server/RequestQuotaTest.scala | 3 -
.../kafka/coordinator/group/GroupCoordinator.java | 22 +-
.../coordinator/group/GroupCoordinatorService.java | 49 +--
.../coordinator/group/GroupCoordinatorShard.java | 25 +-
.../coordinator/group/GroupMetadataManager.java | 318 ++++-----------
.../metrics/GroupCoordinatorMetricsShard.java | 8 +-
.../streams/CoordinatorStreamsRecordHelpers.java | 18 +-
...opology.java => StreamsConfiguredTopology.java} | 80 ++--
.../coordinator/group/streams/StreamsGroup.java | 119 ++++--
...esult.java => StreamsGroupHeartbeatResult.java} | 16 +-
.../coordinator/group/streams/StreamsTopology.java | 48 +--
.../group/streams/TargetAssignmentBuilder.java | 8 +-
.../group/streams/topics/InternalTopicManager.java | 31 +-
.../group/GroupCoordinatorServiceTest.java | 146 +------
.../group/GroupCoordinatorShardTest.java | 40 +-
.../group/GroupMetadataManagerTest.java | 439 ++++++++-------------
.../group/GroupMetadataManagerTestContext.java | 39 +-
.../CoordinatorStreamsRecordHelpersTest.java | 18 +-
.../streams/StreamsConfiguredTopologyTest.java | 93 +++++
.../group/streams/StreamsGroupTest.java | 43 +-
.../group/streams/StreamsTopologyTest.java | 75 +---
.../group/streams/TargetAssignmentBuilderTest.java | 2 +-
.../streams/topics/InternalTopicManagerTest.java | 45 +--
.../apache/kafka/network/RequestConvertToJson.java | 16 +
streams/src/test/resources/log4j.properties | 1 -
49 files changed, 1057 insertions(+), 1964 deletions(-)
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 9128ee7889b..4058bd72725 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
@@ -59,7 +59,6 @@ public class RequestManagers implements Closeable {
public final FetchRequestManager fetchRequestManager;
public final Optional<ShareConsumeRequestManager> shareConsumeRequestManager;
public final Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager;
- public final Optional<StreamsGroupInitializeRequestManager> streamsGroupInitializeRequestManager;
private final List<Optional<? extends RequestManager>> entries;
private final IdempotentCloser closer = new IdempotentCloser();
@@ -71,8 +70,7 @@ public class RequestManagers implements Closeable {
Optional<CommitRequestManager> commitRequestManager,
Optional<ConsumerHeartbeatRequestManager> heartbeatRequestManager,
Optional<ConsumerMembershipManager> membershipManager,
- Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager,
- Optional<StreamsGroupInitializeRequestManager> streamsGroupInitializeRequestManager) {
+ Optional<StreamsGroupHeartbeatRequestManager> streamsGroupHeartbeatRequestManager) {
this.log = logContext.logger(RequestManagers.class);
this.offsetsRequestManager = requireNonNull(offsetsRequestManager, "OffsetsRequestManager cannot be null");
this.coordinatorRequestManager = coordinatorRequestManager;
@@ -85,7 +83,6 @@ public class RequestManagers implements Closeable {
this.consumerMembershipManager = membershipManager;
this.shareMembershipManager = Optional.empty();
this.streamsGroupHeartbeatRequestManager = streamsGroupHeartbeatRequestManager;
- this.streamsGroupInitializeRequestManager = streamsGroupInitializeRequestManager;
List<Optional<? extends RequestManager>> list = new ArrayList<>();
list.add(coordinatorRequestManager);
@@ -96,7 +93,6 @@ public class RequestManagers implements Closeable {
list.add(Optional.of(topicMetadataRequestManager));
list.add(Optional.of(fetchRequestManager));
list.add(streamsGroupHeartbeatRequestManager);
- list.add(streamsGroupInitializeRequestManager);
entries = Collections.unmodifiableList(list);
}
@@ -114,7 +110,6 @@ public class RequestManagers implements Closeable {
this.consumerMembershipManager = Optional.empty();
this.shareMembershipManager = shareMembershipManager;
this.streamsGroupHeartbeatRequestManager = Optional.empty();
- this.streamsGroupInitializeRequestManager = Optional.empty();
this.offsetsRequestManager = null;
this.topicMetadataRequestManager = null;
this.fetchRequestManager = null;
@@ -199,7 +194,6 @@ public class RequestManagers implements Closeable {
CoordinatorRequestManager coordinator = null;
CommitRequestManager commitRequestManager = null;
StreamsGroupHeartbeatRequestManager streamsGroupHeartbeatRequestManager = null;
- StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager = null;
if (groupRebalanceConfig != null && groupRebalanceConfig.groupId != null) {
Optional<String> serverAssignor = Optional.ofNullable(config.getString(ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG));
@@ -245,17 +239,11 @@ public class RequestManagers implements Closeable {
membershipManager.registerStateListener(applicationThreadMemberStateListener);
if (streamsInstanceMetadata.isPresent()) {
- streamsGroupInitializeRequestManager = new StreamsGroupInitializeRequestManager(
- logContext,
- groupRebalanceConfig.groupId,
- streamsInstanceMetadata.get(),
- coordinator);
streamsGroupHeartbeatRequestManager = new StreamsGroupHeartbeatRequestManager(
logContext,
time,
config,
coordinator,
- streamsGroupInitializeRequestManager,
membershipManager,
backgroundEventHandler,
metrics,
@@ -296,8 +284,7 @@ public class RequestManagers implements Closeable {
Optional.ofNullable(commitRequestManager),
Optional.ofNullable(heartbeatRequestManager),
Optional.ofNullable(membershipManager),
- Optional.ofNullable(streamsGroupHeartbeatRequestManager),
- Optional.ofNullable(streamsGroupInitializeRequestManager)
+ Optional.ofNullable(streamsGroupHeartbeatRequestManager)
);
}
};
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
index 650edaa2de2..de6cb9e2098 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsAssignmentInterface.java
@@ -251,19 +251,19 @@ public class StreamsAssignmentInterface {
public static class Subtopology {
public final Set<String> sourceTopics;
- public final Set<String> sinkTopics;
+ public final Set<String> repartitionSinkTopics;
public final Map<String, TopicInfo> stateChangelogTopics;
public final Map<String, TopicInfo> repartitionSourceTopics;
public final Collection<Set<String>> copartitionGroups;
public Subtopology(final Set<String> sourceTopics,
- final Set<String> sinkTopics,
+ final Set<String> repartitionSinkTopics,
final Map<String, TopicInfo> repartitionSourceTopics,
final Map<String, TopicInfo> stateChangelogTopics,
final Collection<Set<String>> copartitionGroups
) {
this.sourceTopics = sourceTopics;
- this.sinkTopics = sinkTopics;
+ this.repartitionSinkTopics = repartitionSinkTopics;
this.stateChangelogTopics = stateChangelogTopics;
this.repartitionSourceTopics = repartitionSourceTopics;
this.copartitionGroups = copartitionGroups;
@@ -273,7 +273,7 @@ public class StreamsAssignmentInterface {
public String toString() {
return "Subtopology{" +
"sourceTopics=" + sourceTopics +
- ", sinkTopics=" + sinkTopics +
+ ", repartitionSinkTopics=" + repartitionSinkTopics +
", stateChangelogTopics=" + stateChangelogTopics +
", repartitionSourceTopics=" + repartitionSourceTopics +
", copartitionGroups=" + copartitionGroups +
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 5fc15b7e436..7506a497939 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
@@ -29,6 +29,7 @@ import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.TopicPartitions;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.CopartitionGroup;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TaskIds;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData.Endpoint;
@@ -44,13 +45,16 @@ import org.apache.kafka.common.utils.Timer;
import org.slf4j.Logger;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import java.util.stream.Stream;
public class StreamsGroupHeartbeatRequestManager implements RequestManager {
@@ -67,8 +71,6 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
private final ConsumerMembershipManager membershipManager;
- private final StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager;
-
private final BackgroundEventHandler backgroundEventHandler;
private final Timer pollTimer;
@@ -86,7 +88,6 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
final Time time,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
- final StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager,
final ConsumerMembershipManager membershipManager,
final BackgroundEventHandler backgroundEventHandler,
final Metrics metrics,
@@ -96,7 +97,6 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
this.coordinatorRequestManager = coordinatorRequestManager;
this.logger = logContext.logger(getClass());
this.membershipManager = membershipManager;
- this.streamsGroupInitializeRequestManager = streamsGroupInitializeRequestManager;
this.backgroundEventHandler = backgroundEventHandler;
this.maxPollIntervalMs = config.getInt(CommonClientConfigs.MAX_POLL_INTERVAL_MS_CONFIG);
long retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
@@ -248,9 +248,6 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
heartbeatRequestState.onSuccessfulAttempt(currentTimeMs);
heartbeatRequestState.resetTimer();
- if (data.shouldInitializeTopology()) {
- streamsGroupInitializeRequestManager.initialize();
- }
if (data.partitionsByUserEndpoint() != null) {
streamsInterface.partitionsByHost.set(convertHostInfoMap(data));
}
@@ -286,7 +283,8 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
membershipManager.onHeartbeatSuccess(new ConsumerGroupHeartbeatResponse(cgData));
}
- private void setTargetAssignmentForConsumerGroup(final StreamsGroupHeartbeatResponseData data, final ConsumerGroupHeartbeatResponseData cgData) {
+ private void setTargetAssignmentForConsumerGroup(final StreamsGroupHeartbeatResponseData data,
+ final ConsumerGroupHeartbeatResponseData cgData) {
Map<String, TopicPartitions> tps = new HashMap<>();
data.activeTasks().forEach(taskId -> Stream.concat(
streamsInterface.subtopologyMap().get(taskId.subtopologyId()).sourceTopics.stream(),
@@ -321,7 +319,8 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
streamsInterface.targetAssignment.set(targetAssignment);
}
- private static Map<StreamsAssignmentInterface.HostInfo, List<TopicPartition>> convertHostInfoMap(final StreamsGroupHeartbeatResponseData data) {
+ private static Map<StreamsAssignmentInterface.HostInfo, List<TopicPartition>> convertHostInfoMap(
+ final StreamsGroupHeartbeatResponseData data) {
Map<StreamsAssignmentInterface.HostInfo, List<TopicPartition>> partitionsByHost = new HashMap<>();
data.partitionsByUserEndpoint().forEach(endpoint -> {
List<TopicPartition> topicPartitions = endpoint.partitions().stream()
@@ -470,7 +469,8 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
final long retryBackoffMs,
final long retryBackoffMaxMs,
final double jitter) {
- super(logContext, StreamsGroupHeartbeatRequestManager.HeartbeatRequestState.class.getName(), retryBackoffMs, 2, retryBackoffMaxMs,
+ super(logContext, StreamsGroupHeartbeatRequestManager.HeartbeatRequestState.class.getName(), retryBackoffMs, 2,
+ retryBackoffMaxMs,
jitter);
this.heartbeatIntervalMs = heartbeatIntervalMs;
this.heartbeatTimer = time.timer(heartbeatIntervalMs);
@@ -574,6 +574,8 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
// Immutable -- only sent when joining
if (joining) {
+ // Topology -- sent when joining
+ data.setTopology(getTopologyFromStreams());
data.setProcessId(streamsInterface.processId().toString());
data.setActiveTasks(Collections.emptyList());
data.setStandbyTasks(Collections.emptyList());
@@ -627,6 +629,100 @@ public class StreamsGroupHeartbeatRequestManager implements RequestManager {
.collect(Collectors.toList());
}
+ private List<StreamsGroupHeartbeatRequestData.Subtopology> getTopologyFromStreams() {
+ final Map<String, StreamsAssignmentInterface.Subtopology> subTopologyMap = streamsInterface.subtopologyMap();
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> subtopologies = new ArrayList<>(subTopologyMap.size());
+ for (final Map.Entry<String, StreamsAssignmentInterface.Subtopology> subtopology : subTopologyMap.entrySet()) {
+ subtopologies.add(getSubtopologyFromStreams(subtopology.getKey(), subtopology.getValue()));
+ }
+ subtopologies.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.Subtopology::subtopologyId));
+ return subtopologies;
+ }
+
+ private static StreamsGroupHeartbeatRequestData.Subtopology getSubtopologyFromStreams(final String subtopologyName,
+ final StreamsAssignmentInterface.Subtopology subtopology) {
+ final StreamsGroupHeartbeatRequestData.Subtopology subtopologyData = new StreamsGroupHeartbeatRequestData.Subtopology();
+ subtopologyData.setSubtopologyId(subtopologyName);
+ ArrayList<String> sortedSourceTopics = new ArrayList<>(subtopology.sourceTopics);
+ Collections.sort(sortedSourceTopics);
+ subtopologyData.setSourceTopics(sortedSourceTopics);
+ ArrayList<String> sortedSinkTopics = new ArrayList<>(subtopology.repartitionSinkTopics);
+ Collections.sort(sortedSinkTopics);
+ subtopologyData.setRepartitionSinkTopics(sortedSinkTopics);
+ subtopologyData.setRepartitionSourceTopics(getRepartitionTopicsInfoFromStreams(subtopology));
+ subtopologyData.setStateChangelogTopics(getChangelogTopicsInfoFromStreams(subtopology));
+ subtopologyData.setCopartitionGroups(
+ getCopartitionGroupsFromStreams(subtopology.copartitionGroups, subtopologyData));
+ return subtopologyData;
+ }
+
+ private static List<CopartitionGroup> getCopartitionGroupsFromStreams(final Collection<Set<String>> copartitionGroups,
+ final StreamsGroupHeartbeatRequestData.Subtopology subtopologyData) {
+
+ final Map<String, Short> sourceTopicsMap =
+ IntStream.range(0, subtopologyData.sourceTopics().size())
+ .boxed()
+ .collect(Collectors.toMap(subtopologyData.sourceTopics()::get, Integer::shortValue));
+
+ final Map<String, Short> repartitionSourceTopics =
+ IntStream.range(0, subtopologyData.repartitionSourceTopics().size())
+ .boxed()
+ .collect(
+ Collectors.toMap(x -> subtopologyData.repartitionSourceTopics().get(x).name(),
+ Integer::shortValue));
+
+ return copartitionGroups.stream()
+ .map(x -> getCopartitionGroupFromStreams(x, sourceTopicsMap, repartitionSourceTopics))
+ .collect(Collectors.toList());
+ }
+
+ private static CopartitionGroup getCopartitionGroupFromStreams(final Set<String> topicNames,
+ final Map<String, Short> sourceTopicsMap,
+ final Map<String, Short> repartitionSourceTopics) {
+ CopartitionGroup copartitionGroup = new CopartitionGroup();
+
+ topicNames.forEach(topicName -> {
+ if (sourceTopicsMap.containsKey(topicName)) {
+ copartitionGroup.sourceTopics().add(sourceTopicsMap.get(topicName));
+ } else if (repartitionSourceTopics.containsKey(topicName)) {
+ copartitionGroup.repartitionSourceTopics()
+ .add(repartitionSourceTopics.get(topicName));
+ } else {
+ throw new IllegalStateException(
+ "Source topic not found in subtopology: " + topicName);
+ }
+ });
+
+ return copartitionGroup;
+ }
+
+ private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getRepartitionTopicsInfoFromStreams(
+ final StreamsAssignmentInterface.Subtopology subtopologyDataFromStreams) {
+ final List<StreamsGroupHeartbeatRequestData.TopicInfo> repartitionTopicsInfo = new ArrayList<>();
+ for (final Map.Entry<String, StreamsAssignmentInterface.TopicInfo> repartitionTopic : subtopologyDataFromStreams.repartitionSourceTopics.entrySet()) {
+ final StreamsGroupHeartbeatRequestData.TopicInfo repartitionTopicInfo = new StreamsGroupHeartbeatRequestData.TopicInfo();
+ repartitionTopicInfo.setName(repartitionTopic.getKey());
+ repartitionTopic.getValue().numPartitions.ifPresent(repartitionTopicInfo::setPartitions);
+ repartitionTopic.getValue().replicationFactor.ifPresent(repartitionTopicInfo::setReplicationFactor);
+ repartitionTopicsInfo.add(repartitionTopicInfo);
+ }
+ repartitionTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));
+ return repartitionTopicsInfo;
+ }
+
+ private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getChangelogTopicsInfoFromStreams(
+ final StreamsAssignmentInterface.Subtopology subtopologyDataFromStreams) {
+ final List<StreamsGroupHeartbeatRequestData.TopicInfo> changelogTopicsInfo = new ArrayList<>();
+ for (final Map.Entry<String, StreamsAssignmentInterface.TopicInfo> changelogTopic : subtopologyDataFromStreams.stateChangelogTopics.entrySet()) {
+ final StreamsGroupHeartbeatRequestData.TopicInfo changelogTopicInfo = new StreamsGroupHeartbeatRequestData.TopicInfo();
+ changelogTopicInfo.setName(changelogTopic.getKey());
+ changelogTopic.getValue().replicationFactor.ifPresent(changelogTopicInfo::setReplicationFactor);
+ changelogTopicsInfo.add(changelogTopicInfo);
+ }
+ changelogTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));
+ return changelogTopicsInfo;
+ }
+
// Fields of StreamsGroupHeartbeatRequest sent in the most recent request
static class SentFields {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManager.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManager.java
deleted file mode 100644
index 8c7671efe67..00000000000
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManager.java
+++ /dev/null
@@ -1,210 +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.ClientResponse;
-import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData.CopartitionGroup;
-import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.requests.StreamsGroupInitializeRequest;
-import org.apache.kafka.common.requests.StreamsGroupInitializeResponse;
-import org.apache.kafka.common.utils.LogContext;
-
-import org.slf4j.Logger;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-public class StreamsGroupInitializeRequestManager implements RequestManager {
-
- private final Logger logger;
- private final String groupId;
- private final StreamsAssignmentInterface streamsAssignmentInterface;
- private final CoordinatorRequestManager coordinatorRequestManager;
-
- private Optional<NetworkClientDelegate.UnsentRequest> unsentRequest = Optional.empty();
-
-
- StreamsGroupInitializeRequestManager(final LogContext logContext,
- final String groupId,
- final StreamsAssignmentInterface streamsAssignmentInterface,
- final CoordinatorRequestManager coordinatorRequestManager) {
- this.logger = logContext.logger(getClass());
- this.groupId = groupId;
- this.streamsAssignmentInterface = streamsAssignmentInterface;
- this.coordinatorRequestManager = coordinatorRequestManager;
- }
-
- @Override
- public PollResult poll(final long currentTimeMs) {
- final PollResult pollResult = unsentRequest.map(PollResult::new).orElse(PollResult.EMPTY);
- unsentRequest = Optional.empty();
- return pollResult;
- }
-
- public void initialize() {
- final NetworkClientDelegate.UnsentRequest unsentRequest = makeRequest();
-
- unsentRequest.whenComplete(this::onResponse);
-
- this.unsentRequest = Optional.of(unsentRequest);
- }
-
- private NetworkClientDelegate.UnsentRequest makeRequest() {
- final StreamsGroupInitializeRequestData streamsGroupInitializeRequestData = new StreamsGroupInitializeRequestData();
- streamsGroupInitializeRequestData.setGroupId(groupId);
- streamsGroupInitializeRequestData.setTopologyId(streamsAssignmentInterface.topologyId());
- final List<StreamsGroupInitializeRequestData.Subtopology> topology = getTopologyFromStreams();
- streamsGroupInitializeRequestData.setTopology(topology);
- final StreamsGroupInitializeRequest.Builder streamsGroupInitializeRequestBuilder =
- new StreamsGroupInitializeRequest.Builder(streamsGroupInitializeRequestData);
- return new NetworkClientDelegate.UnsentRequest(
- streamsGroupInitializeRequestBuilder,
- coordinatorRequestManager.coordinator()
- );
- }
-
- private List<StreamsGroupInitializeRequestData.Subtopology> getTopologyFromStreams() {
- final Map<String, StreamsAssignmentInterface.Subtopology> subTopologyMap = streamsAssignmentInterface.subtopologyMap();
- final List<StreamsGroupInitializeRequestData.Subtopology> topology = new ArrayList<>(subTopologyMap.size());
- for (final Map.Entry<String, StreamsAssignmentInterface.Subtopology> subtopology : subTopologyMap.entrySet()) {
- topology.add(getSubtopologyFromStreams(subtopology.getKey(), subtopology.getValue()));
- }
- return topology;
- }
-
- private static StreamsGroupInitializeRequestData.Subtopology getSubtopologyFromStreams(final String subtopologyName,
- final StreamsAssignmentInterface.Subtopology subtopology) {
- final StreamsGroupInitializeRequestData.Subtopology subtopologyData = new StreamsGroupInitializeRequestData.Subtopology();
- subtopologyData.setSubtopologyId(subtopologyName);
- ArrayList<String> sortedSourceTopics = new ArrayList<>(subtopology.sourceTopics);
- Collections.sort(sortedSourceTopics);
- subtopologyData.setSourceTopics(sortedSourceTopics);
- // TODO: We should only encode the repartition sink topics here.
- ArrayList<String> sortedSinkTopics = new ArrayList<>(subtopology.sinkTopics);
- Collections.sort(sortedSinkTopics);
- subtopologyData.setRepartitionSinkTopics(sortedSinkTopics);
- subtopologyData.setRepartitionSourceTopics(getRepartitionTopicsInfoFromStreams(subtopology));
- subtopologyData.setStateChangelogTopics(getChangelogTopicsInfoFromStreams(subtopology));
- subtopologyData.setCopartitionGroups(
- getCopartitionGroupsFromStreams(subtopology.copartitionGroups, subtopologyData));
- return subtopologyData;
- }
-
- private static List<CopartitionGroup> getCopartitionGroupsFromStreams(
- final Collection<Set<String>> copartitionGroups,
- final StreamsGroupInitializeRequestData.Subtopology subtopologyData) {
-
- final Map<String, Short> sourceTopicsMap =
- IntStream.range(0, subtopologyData.sourceTopics().size())
- .boxed()
- .collect(Collectors.toMap(subtopologyData.sourceTopics()::get, Integer::shortValue));
-
- final Map<String, Short> repartitionSourceTopics =
- IntStream.range(0, subtopologyData.repartitionSourceTopics().size())
- .boxed()
- .collect(
- Collectors.toMap(x -> subtopologyData.repartitionSourceTopics().get(x).name(),
- Integer::shortValue));
-
- return copartitionGroups.stream()
- .map(x -> getCopartitionGroupFromStreams(x, sourceTopicsMap, repartitionSourceTopics))
- .collect(Collectors.toList());
- }
-
- private static CopartitionGroup getCopartitionGroupFromStreams(
- final Set<String> topicNames,
- final Map<String, Short> sourceTopicsMap,
- final Map<String, Short> repartitionSourceTopics) {
- CopartitionGroup copartitionGroup = new CopartitionGroup();
-
- topicNames.forEach(topicName -> {
- if (sourceTopicsMap.containsKey(topicName)) {
- copartitionGroup.sourceTopics().add(sourceTopicsMap.get(topicName));
- } else if (repartitionSourceTopics.containsKey(topicName)) {
- copartitionGroup.repartitionSourceTopics()
- .add(repartitionSourceTopics.get(topicName));
- } else {
- throw new IllegalStateException(
- "Source topic not found in subtopology: " + topicName);
- }
- });
-
- return copartitionGroup;
- }
-
- private static List<StreamsGroupInitializeRequestData.TopicInfo> getRepartitionTopicsInfoFromStreams(final StreamsAssignmentInterface.Subtopology subtopologyDataFromStreams) {
- final List<StreamsGroupInitializeRequestData.TopicInfo> repartitionTopicsInfo = new ArrayList<>();
- for (final Map.Entry<String, StreamsAssignmentInterface.TopicInfo> repartitionTopic : subtopologyDataFromStreams.repartitionSourceTopics.entrySet()) {
- final StreamsGroupInitializeRequestData.TopicInfo repartitionTopicInfo = new StreamsGroupInitializeRequestData.TopicInfo();
- repartitionTopicInfo.setName(repartitionTopic.getKey());
- repartitionTopic.getValue().numPartitions.ifPresent(repartitionTopicInfo::setPartitions);
- repartitionTopic.getValue().replicationFactor.ifPresent(repartitionTopicInfo::setReplicationFactor);
- repartitionTopicsInfo.add(repartitionTopicInfo);
- }
- repartitionTopicsInfo.sort(Comparator.comparing(StreamsGroupInitializeRequestData.TopicInfo::name));
- return repartitionTopicsInfo;
- }
-
- private static List<StreamsGroupInitializeRequestData.TopicInfo> getChangelogTopicsInfoFromStreams(final StreamsAssignmentInterface.Subtopology subtopologyDataFromStreams) {
- final List<StreamsGroupInitializeRequestData.TopicInfo> changelogTopicsInfo = new ArrayList<>();
- for (final Map.Entry<String, StreamsAssignmentInterface.TopicInfo> changelogTopic : subtopologyDataFromStreams.stateChangelogTopics.entrySet()) {
- final StreamsGroupInitializeRequestData.TopicInfo changelogTopicInfo = new StreamsGroupInitializeRequestData.TopicInfo();
- changelogTopicInfo.setName(changelogTopic.getKey());
- changelogTopic.getValue().replicationFactor.ifPresent(changelogTopicInfo::setReplicationFactor);
- changelogTopicsInfo.add(changelogTopicInfo);
- }
- changelogTopicsInfo.sort(Comparator.comparing(StreamsGroupInitializeRequestData.TopicInfo::name));
- return changelogTopicsInfo;
- }
-
- private void onResponse(final ClientResponse response, final Throwable exception) {
- if (exception != null) {
- // todo: handle error
- logger.error("Error during Streams initialization: ", exception);
- } else {
- onResponse((StreamsGroupInitializeResponse) response.responseBody());
- }
- }
-
- private void onResponse(final StreamsGroupInitializeResponse response) {
- if (Errors.forCode(response.data().errorCode()) == Errors.NONE) {
- onSuccessResponse(response);
- } else {
- onErrorResponse(response);
- }
- }
-
- private void onErrorResponse(final StreamsGroupInitializeResponse response) {
- // todo: handle error
- logger.error("Error during Streams initialization: {}", response);
- }
-
- private void onSuccessResponse(final StreamsGroupInitializeResponse response) {
- // todo: handle success
- logger.info("Streams initialization successful {}", response);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
index 345b9b808a8..35e225b8956 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
@@ -131,7 +131,6 @@ public enum ApiKeys {
WRITE_SHARE_GROUP_STATE(ApiMessageType.WRITE_SHARE_GROUP_STATE, true),
DELETE_SHARE_GROUP_STATE(ApiMessageType.DELETE_SHARE_GROUP_STATE, true),
READ_SHARE_GROUP_STATE_SUMMARY(ApiMessageType.READ_SHARE_GROUP_STATE_SUMMARY, true),
- STREAMS_GROUP_INITIALIZE(ApiMessageType.STREAMS_GROUP_INITIALIZE),
STREAMS_GROUP_HEARTBEAT(ApiMessageType.STREAMS_GROUP_HEARTBEAT),
STREAMS_GROUP_DESCRIBE(ApiMessageType.STREAMS_GROUP_DESCRIBE);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
index 4e5d21251f8..c16c71903d6 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java
@@ -352,8 +352,6 @@ public abstract class AbstractRequest implements AbstractRequestResponse {
return ReadShareGroupStateSummaryRequest.parse(buffer, apiVersion);
case STREAMS_GROUP_HEARTBEAT:
return StreamsGroupHeartbeatRequest.parse(buffer, apiVersion);
- case STREAMS_GROUP_INITIALIZE:
- return StreamsGroupInitializeRequest.parse(buffer, apiVersion);
case STREAMS_GROUP_DESCRIBE:
return StreamsGroupDescribeRequest.parse(buffer, apiVersion);
default:
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index 25f3afe0d5b..8f7b12d4fa4 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -289,8 +289,6 @@ public abstract class AbstractResponse implements AbstractRequestResponse {
return ReadShareGroupStateSummaryResponse.parse(responseBuffer, version);
case STREAMS_GROUP_HEARTBEAT:
return StreamsGroupHeartbeatResponse.parse(responseBuffer, version);
- case STREAMS_GROUP_INITIALIZE:
- return StreamsGroupInitializeResponse.parse(responseBuffer, version);
case STREAMS_GROUP_DESCRIBE:
return StreamsGroupDescribeResponse.parse(responseBuffer, version);
default:
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatResponse.java
index 2785bef35f7..153a2257f18 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupHeartbeatResponse.java
@@ -38,6 +38,7 @@ import java.util.Map;
* - {@link Errors#UNSUPPORTED_ASSIGNOR}
* - {@link Errors#UNRELEASED_INSTANCE_ID}
* - {@link Errors#GROUP_MAX_SIZE_REACHED}
+ * - {@link Errors#STREAMS_INVALID_TOPOLOGY}
* - {@link Errors#STREAMS_SHUTDOWN_APPLICATION}
* - {@link Errors#STREAMS_INCONSISTENT_TOPOLOGY}
* - {@link Errors#STREAMS_MISSING_SOURCE_TOPICS}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeRequest.java
deleted file mode 100644
index bebcf103cf9..00000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeRequest.java
+++ /dev/null
@@ -1,77 +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.common.requests;
-
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-
-public class StreamsGroupInitializeRequest extends AbstractRequest {
-
- public static class Builder extends AbstractRequest.Builder<StreamsGroupInitializeRequest> {
- private final StreamsGroupInitializeRequestData data;
-
- public Builder(StreamsGroupInitializeRequestData data) {
- this(data, false);
- }
-
- public Builder(StreamsGroupInitializeRequestData data, boolean enableUnstableLastVersion) {
- super(ApiKeys.STREAMS_GROUP_INITIALIZE, enableUnstableLastVersion);
- this.data = data;
- }
-
- @Override
- public StreamsGroupInitializeRequest build(short version) {
- return new StreamsGroupInitializeRequest(data, version);
- }
-
- @Override
- public String toString() {
- return data.toString();
- }
- }
-
- private final StreamsGroupInitializeRequestData data;
-
- public StreamsGroupInitializeRequest(StreamsGroupInitializeRequestData data, short version) {
- super(ApiKeys.STREAMS_GROUP_INITIALIZE, version);
- this.data = data;
- }
-
- @Override
- public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
- return new StreamsGroupInitializeResponse(
- new StreamsGroupInitializeResponseData()
- .setThrottleTimeMs(throttleTimeMs)
- .setErrorCode(Errors.forException(e).code())
- );
- }
-
- @Override
- public StreamsGroupInitializeRequestData data() {
- return data;
- }
-
- public static StreamsGroupInitializeRequest parse(ByteBuffer buffer, short version) {
- return new StreamsGroupInitializeRequest(new StreamsGroupInitializeRequestData(
- new ByteBufferAccessor(buffer), version), version);
- }
-}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeResponse.java
deleted file mode 100644
index cb224b3c6d7..00000000000
--- a/clients/src/main/java/org/apache/kafka/common/requests/StreamsGroupInitializeResponse.java
+++ /dev/null
@@ -1,71 +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.common.requests;
-
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.protocol.ByteBufferAccessor;
-import org.apache.kafka.common.protocol.Errors;
-
-import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
-
-/**
- * Possible error codes.
- *
- * - {@link Errors#GROUP_AUTHORIZATION_FAILED}
- * - {@link Errors#NOT_COORDINATOR}
- * - {@link Errors#COORDINATOR_NOT_AVAILABLE}
- * - {@link Errors#COORDINATOR_LOAD_IN_PROGRESS}
- * - {@link Errors#INVALID_REQUEST}
- * - {@link Errors#STREAMS_INVALID_TOPOLOGY}
- */
-public class StreamsGroupInitializeResponse extends AbstractResponse {
-
- private final StreamsGroupInitializeResponseData data;
-
- public StreamsGroupInitializeResponse(StreamsGroupInitializeResponseData data) {
- super(ApiKeys.STREAMS_GROUP_INITIALIZE);
- this.data = data;
- }
-
- @Override
- public StreamsGroupInitializeResponseData data() {
- return data;
- }
-
- @Override
- public Map<Errors, Integer> errorCounts() {
- return Collections.singletonMap(Errors.forCode(data.errorCode()), 1);
- }
-
- @Override
- public int throttleTimeMs() {
- return data.throttleTimeMs();
- }
-
- @Override
- public void maybeSetThrottleTimeMs(int throttleTimeMs) {
- data.setThrottleTimeMs(throttleTimeMs);
- }
-
- public static StreamsGroupInitializeResponse parse(ByteBuffer buffer, short version) {
- return new StreamsGroupInitializeResponse(new StreamsGroupInitializeResponseData(
- new ByteBufferAccessor(buffer), version));
- }
-}
diff --git a/clients/src/main/resources/common/message/StreamsGroupDescribeRequest.json b/clients/src/main/resources/common/message/StreamsGroupDescribeRequest.json
index b7047141e8a..722ac7fc053 100644
--- a/clients/src/main/resources/common/message/StreamsGroupDescribeRequest.json
+++ b/clients/src/main/resources/common/message/StreamsGroupDescribeRequest.json
@@ -14,7 +14,7 @@
// limitations under the License.
{
- "apiKey": 90,
+ "apiKey": 89,
"type": "request",
"listeners": ["broker", "zkBroker"],
"name": "StreamsGroupDescribeRequest",
diff --git a/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json b/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
index 71d74eee831..36526b241db 100644
--- a/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
+++ b/clients/src/main/resources/common/message/StreamsGroupDescribeResponse.json
@@ -14,7 +14,7 @@
// limitations under the License.
{
- "apiKey": 90,
+ "apiKey": 89,
"type": "response",
"name": "StreamsGroupDescribeResponse",
"validVersions": "0",
@@ -46,7 +46,7 @@
{ "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
"about": "The assignment epoch." },
{ "name": "Topology", "type": "[]Subtopology", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The sub-topologies of the streams application. Null if uninitialized.",
+ "about": "The resolved sub-topologies of the streams application. This contains the configured sub-topologies, where the number of partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or inconsistent.",
"fields": [
{ "name": "SubtopologyId", "type": "string", "versions": "0+",
"about": "String to uniquely identify the subtopology. Deterministically generated from the topology." },
@@ -54,8 +54,6 @@
"about": "The topics the topology reads from." },
{ "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
"about": "The repartition topics the topology writes to." },
- { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
- "about": "Regular expressions identifying topics the subtopology reads from." },
{ "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
"about": "The set of state changelog topics associated with this subtopology. Created automatically." },
{ "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/StreamsGroupHeartbeatRequest.json b/clients/src/main/resources/common/message/StreamsGroupHeartbeatRequest.json
index ec28c2766fb..d4ae5df26c2 100644
--- a/clients/src/main/resources/common/message/StreamsGroupHeartbeatRequest.json
+++ b/clients/src/main/resources/common/message/StreamsGroupHeartbeatRequest.json
@@ -14,7 +14,7 @@
// limitations under the License.
{
- "apiKey": 89,
+ "apiKey": 88,
"type": "request",
"listeners": ["broker", "zkBroker"],
"name": "StreamsGroupHeartbeatRequest",
@@ -24,18 +24,44 @@
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The group identifier." },
{ "name": "MemberId", "type": "string", "versions": "0+",
- "about": "The member id generated by the coordinator. The member id must be kept during the entire lifetime of the member." },
+ "about": "The member ID generated by the coordinator. The member ID must be kept during the entire lifetime of the member." },
{ "name": "MemberEpoch", "type": "int32", "versions": "0+",
"about": "The current member epoch; 0 to join the group; -1 to leave the group; -2 to indicate that the static member will rejoin." },
{ "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "null if not provided or if it didn't change since the last heartbeat; the instance Id otherwise." },
+ "about": "null if not provided or if it didn't change since the last heartbeat; the instance ID for static membership otherwise." },
{ "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
{ "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
"about": "-1 if it didn't change since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its partitions otherwise." },
- { "name": "TopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
- "about": "The ID of the topology. Only sent when memberEpoch = 0, must be non-empty. Null otherwise. Used to check if topology corresponds to server-topology. " },
+ { "name": "TopologyId", "type": "string", "versions": "0+",
+ "about": "The ID of the topology. Used to check if the topology corresponds to the topology initialized on the brokers." },
+ { "name": "Topology", "type": "[]Subtopology", "versions": "0+", "nullableVersions": "0+", "default": "null",
+ "about": "The topology data of the streams application. Used to initialize the topology of the group. Can only sent when memberEpoch = 0, must then be non-empty. Null otherwise.",
+ "fields": [
+ { "name": "SubtopologyId", "type": "string", "versions": "0+",
+ "about": "String to uniquely identify the sub-topology. Deterministically generated from the topology" },
+ { "name": "SourceTopics", "type": "[]string", "versions": "0+",
+ "about": "The topics the topology reads from." },
+ { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
+ "about": "The regular expressions identifying topics the sub-topology reads from." },
+ { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
+ "about": "The set of state changelog topics associated with this sub-topology. Created automatically." },
+ { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
+ "about": "The repartition topics the sub-topology writes to." },
+ { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
+ "about": "The set of source topics that are internally created repartition topics. Created automatically." },
+ { "name": "CopartitionGroups", "type": "[]CopartitionGroup", "versions": "0+",
+ "about": "A subset of source topics that must be copartitioned.",
+ "fields": [
+ { "name": "SourceTopics", "type": "[]int16", "versions": "0+",
+ "about": "The topics the topology reads from. Index into the array on the subtopology level." },
+ { "name": "SourceTopicRegex", "type": "[]int16", "versions": "0+",
+ "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
+ { "name": "RepartitionSourceTopics", "type": "[]int16", "versions": "0+",
+ "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology level." }
+ ]}
+ ]},
{ "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Currently owned active tasks for this client. Null if unchanged since last heartbeat." },
@@ -52,25 +78,36 @@
"about": "Used for rack-aware assignment algorithm. Null if unchanged since last heartbeat." },
{ "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "Cumulative offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
+ "about": "Cumulative changelog offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "Cumulative end-offsets for tasks. Only updated when a warm up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
+ "about": "Cumulative changelog end-offsets for tasks. Only updated when a warm-up task has caught up, and according to the task offset interval. Null if unchanged since last heartbeat." },
{ "name": "ShutdownApplication", "type": "bool", "versions": "0+", "default": false,
- "about": "Whether all Streams clients in the group should shut down. Can only be defined if memberEpoch = -1" }
+ "about": "Whether all Streams clients in the group should shut down." }
],
"commonStructs": [
- { "name": "Endpoint", "versions": "0+", "fields": [
- { "name": "Host", "type": "string", "versions": "0+",
- "about": "host of the endpoint" },
- { "name": "Port", "type": "uint16", "versions": "0+",
- "about": "port of the endpoint" }
- ]},
{ "name": "KeyValue", "versions": "0+", "fields": [
{ "name": "Key", "type": "string", "versions": "0+",
"about": "key of the config" },
{ "name": "Value", "type": "string", "versions": "0+",
"about": "value of the config" }
]},
+ { "name": "TopicInfo", "versions": "0+", "fields": [
+ { "name": "Name", "type": "string", "versions": "0+",
+ "about": "The name of the topic." },
+ { "name": "Partitions", "type": "int32", "versions": "0+",
+ "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
+ { "name": "ReplicationFactor", "type": "int16", "versions": "0+",
+ "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
+ { "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+",
+ "about": "Topic-level configurations as key-value pairs."
+ }
+ ]},
+ { "name": "Endpoint", "versions": "0+", "fields": [
+ { "name": "Host", "type": "string", "versions": "0+",
+ "about": "host of the endpoint" },
+ { "name": "Port", "type": "int32", "versions": "0+",
+ "about": "port of the endpoint" }
+ ]},
{ "name": "TaskOffset", "versions": "0+", "fields": [
{ "name": "SubtopologyId", "type": "string", "versions": "0+",
"about": "The subtopology identifier." },
diff --git a/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json b/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
index 40e134dc407..7b254a17de6 100644
--- a/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
+++ b/clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json
@@ -14,13 +14,14 @@
// limitations under the License.
{
- "apiKey": 89,
+ "apiKey": 88,
"type": "response",
"name": "StreamsGroupHeartbeatResponse",
"validVersions": "0",
"flexibleVersions": "0+",
// Supported errors:
// - GROUP_AUTHORIZATION_FAILED (version 0+)
+ // - GROUP_ID_NOT_FOUND (version 0+)
// - NOT_COORDINATOR (version 0+)
// - COORDINATOR_NOT_AVAILABLE (version 0+)
// - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
@@ -29,6 +30,11 @@
// - FENCED_MEMBER_EPOCH (version 0+)
// - UNRELEASED_INSTANCE_ID (version 0+)
// - GROUP_MAX_SIZE_REACHED (version 0+)
+ // - TOPIC_AUTHORIZATION_FAILED (version 0+)
+ // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
+ // - STREAMS_INVALID_TOPOLOGY (version 0+)
+ // - STREAMS_MISSING_SOURCE_TOPICS (version 0+)
+ // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+)
"fields": [
// Same as consumer group heart beat
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
@@ -46,13 +52,12 @@
{ "name": "AcceptableRecoveryLag", "type": "int32", "versions": "0+",
"about": "The maximal lag a warm-up task can have to be considered caught-up." },
{ "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
- "about": "The interval in which the changelog task offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
+ "about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
// Topology updating
- { "name": "ShouldInitializeTopology", "type": "bool", "versions": "0+", "default": false,
- "about": "true if this streams application should initialize the topology on the broker" },
{ "name": "GroupTopologyId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": null,
- "about": "The current ID of the topology for the group. Null if unchanged since last heartbeat, empty if group is unitialized." },
+ "about": "The current ID of the topology for the group. Null if unchanged since last heartbeat." },
+
{ "name": "Status", "type": "[]Status", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "Indicate zero or more status for the group. Null if unchanged since last heartbeat." },
diff --git a/clients/src/main/resources/common/message/StreamsGroupInitializeRequest.json b/clients/src/main/resources/common/message/StreamsGroupInitializeRequest.json
deleted file mode 100644
index aa68fde6bf5..00000000000
--- a/clients/src/main/resources/common/message/StreamsGroupInitializeRequest.json
+++ /dev/null
@@ -1,77 +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.
-
-{
- "apiKey": 88,
- "type": "request",
- "listeners": ["broker", "zkBroker"],
- "name": "StreamsGroupInitializeRequest",
- "validVersions": "0",
- "flexibleVersions": "0+",
- "fields": [
- { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
- "about": "The group identifier." },
- { "name": "TopologyId", "type": "string", "versions": "0+",
- "about": "The ID of the topology." },
- { "name": "Topology", "type": "[]Subtopology", "versions": "0+",
- "about": "The sub-topologies of the streams application.",
- "fields": [
- { "name": "SubtopologyId", "type": "string", "versions": "0+",
- "about": "String to uniquely identify the subtopology. Deterministically generated from the topology." },
- { "name": "SourceTopics", "type": "[]string", "versions": "0+",
- "about": "The topics the topology reads from." },
- { "name": "SourceTopicRegex", "type": "[]string", "versions": "0+",
- "about": "Regular expressions identifying topics the subtopology reads from." },
- { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
- "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
- { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
- "about": "The repartition topics the subtopology writes to." },
- { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
- "about": "The set of source topics that are internally created repartition topics. Created automatically." },
- { "name": "CopartitionGroups", "type": "[]CopartitionGroup", "versions": "0+",
- "about": "A subset of source topics that must be copartitioned.",
- "fields": [
- { "name": "SourceTopics", "type": "[]int16", "versions": "0+",
- "about": "The topics the topology reads from. Index into the array on the subtopology level." },
- { "name": "SourceTopicRegex", "type": "[]int16", "versions": "0+",
- "about": "Regular expressions identifying topics the subtopology reads from. Index into the array on the subtopology level." },
- { "name": "RepartitionSourceTopics", "type": "[]int16", "versions": "0+",
- "about": "The set of source topics that are internally created repartition topics. Index into the array on the subtopology level." }
- ]
- }
- ]
- }
- ],
- "commonStructs": [
- { "name": "TopicConfig", "versions": "0+", "fields": [
- { "name": "key", "type": "string", "versions": "0+",
- "about": "The key of the topic-level configuration." },
- { "name": "value", "type": "string", "versions": "0+",
- "about": "The value of the topic-level configuration," }
- ]
- },
- { "name": "TopicInfo", "versions": "0+", "fields": [
- { "name": "Name", "type": "string", "versions": "0+",
- "about": "The name of the topic." },
- { "name": "Partitions", "type": "int32", "versions": "0+",
- "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
- { "name": "ReplicationFactor", "type": "int16", "versions": "0+",
- "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
- { "name": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+",
- "about": "Topic-level configurations as key-value pairs."
- }
- ]}
- ]
-}
diff --git a/clients/src/main/resources/common/message/StreamsGroupInitializeResponse.json b/clients/src/main/resources/common/message/StreamsGroupInitializeResponse.json
deleted file mode 100644
index 327548577ea..00000000000
--- a/clients/src/main/resources/common/message/StreamsGroupInitializeResponse.json
+++ /dev/null
@@ -1,42 +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.
-
-{
- "apiKey": 88,
- "type": "response",
- "name": "StreamsGroupInitializeResponse",
- "validVersions": "0",
- "flexibleVersions": "0+",
- // Supported errors:
- // - GROUP_AUTHORIZATION_FAILED (version 0+)
- // - GROUP_ID_NOT_FOUND (version 0+)
- // - NOT_COORDINATOR (version 0+)
- // - COORDINATOR_NOT_AVAILABLE (version 0+)
- // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
- // - INVALID_REQUEST (version 0+)
- // - TOPIC_AUTHORIZATION_FAILED (version 0+)
- // - CLUSTER_AUTHORIZATION_FAILED (version 0+)
- // - STREAMS_INVALID_TOPOLOGY (version 0+)
- // - STREAMS_MISSING_SOURCE_TOPICS (version 0+)
- // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+)
- "fields": [
- { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
- "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
- { "name": "ErrorCode", "type": "int16", "versions": "0+",
- "about": "The top-level error code, or 0 if there was no error" },
- { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
- "about": "The top-level error message, or null if there was no error." }
- ]
-}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
index bec93f98f22..9e88e2c7569 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
@@ -28,6 +28,7 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData.TopicPartitions;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
@@ -48,6 +49,8 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -61,7 +64,9 @@ import java.util.stream.Collectors;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -87,9 +92,6 @@ class StreamsGroupHeartbeatRequestManagerTest {
@Mock
private CoordinatorRequestManager coordinatorRequestManager;
- @Mock
- private StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager;
-
@Mock
private ConsumerMembershipManager membershipManager;
@@ -136,7 +138,6 @@ class StreamsGroupHeartbeatRequestManagerTest {
time,
config,
coordinatorRequestManager,
- streamsGroupInitializeRequestManager,
membershipManager,
backgroundEventHandler,
metrics,
@@ -188,16 +189,44 @@ class StreamsGroupHeartbeatRequestManagerTest {
assertEquals(TEST_MEMBER_ID, request.data().memberId());
assertEquals(TEST_MEMBER_EPOCH, request.data().memberEpoch());
assertEquals(TEST_INSTANCE_ID, request.data().instanceId());
+ assertEquals(streamsAssignmentInterface.topologyId(), request.data().topologyId());
// Static information is null
assertNull(request.data().processId());
assertNull(request.data().userEndpoint());
assertNull(request.data().clientTags());
+ assertNull(request.data().topology());
}
@Test
void testFullStaticInformationWhenJoining() {
mockJoiningState();
+
+ final Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ final Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2", "repartitionSinkTopic3");
+ final Map<String, StreamsAssignmentInterface.TopicInfo> repartitionSourceTopics = mkMap(
+ mkEntry("repartitionTopic1", new StreamsAssignmentInterface.TopicInfo(Optional.of(2), Optional.of((short) 1), Collections.emptyMap())),
+ mkEntry("repartitionTopic2", new StreamsAssignmentInterface.TopicInfo(Optional.of(3), Optional.of((short) 3), Collections.emptyMap()))
+ );
+ final Map<String, StreamsAssignmentInterface.TopicInfo> changelogTopics = mkMap(
+ mkEntry("changelogTopic1", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 1), Collections.emptyMap())),
+ mkEntry("changelogTopic2", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 2), Collections.emptyMap())),
+ mkEntry("changelogTopic3", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 3), Collections.emptyMap()))
+ );
+ final Collection<Set<String>> copartitionGroup = Set.of(
+ Set.of("sourceTopic1", "repartitionTopic2"),
+ Set.of("sourceTopic2", "repartitionTopic1")
+ );
+ final StreamsAssignmentInterface.Subtopology subtopology1 = new StreamsAssignmentInterface.Subtopology(
+ sourceTopics,
+ repartitionSinkTopics,
+ repartitionSourceTopics,
+ changelogTopics,
+ copartitionGroup
+ );
+ final String subtopologyName1 = "subtopology1";
+ subtopologyMap.put(subtopologyName1, subtopology1);
+
clientTags.put("clientTag1", "value2");
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
@@ -213,6 +242,38 @@ class StreamsGroupHeartbeatRequestManagerTest {
assertEquals(1, request.data().clientTags().size());
assertEquals("clientTag1", request.data().clientTags().get(0).key());
assertEquals("value2", request.data().clientTags().get(0).value());
+ assertEquals(streamsAssignmentInterface.topologyId(), request.data().topologyId());
+ assertNotNull(request.data().topology());
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> subtopologies = request.data().topology();
+ assertEquals(1, subtopologies.size());
+ final StreamsGroupHeartbeatRequestData.Subtopology subtopology = subtopologies.get(0);
+ assertEquals(subtopologyName1, subtopology.subtopologyId());
+ assertEquals(Arrays.asList("sourceTopic1", "sourceTopic2"), subtopology.sourceTopics());
+ assertEquals(Arrays.asList("repartitionSinkTopic1", "repartitionSinkTopic2", "repartitionSinkTopic3"), subtopology.repartitionSinkTopics());
+ assertEquals(repartitionSourceTopics.size(), subtopology.repartitionSourceTopics().size());
+ subtopology.repartitionSourceTopics().forEach(topicInfo -> {
+ final StreamsAssignmentInterface.TopicInfo repartitionTopic = repartitionSourceTopics.get(topicInfo.name());
+ assertEquals(repartitionTopic.numPartitions.get(), topicInfo.partitions());
+ assertEquals(repartitionTopic.replicationFactor.get(), topicInfo.replicationFactor());
+ });
+ assertEquals(changelogTopics.size(), subtopology.stateChangelogTopics().size());
+ subtopology.stateChangelogTopics().forEach(topicInfo -> {
+ assertTrue(changelogTopics.containsKey(topicInfo.name()));
+ assertEquals(0, topicInfo.partitions());
+ final StreamsAssignmentInterface.TopicInfo changelogTopic = changelogTopics.get(topicInfo.name());
+ assertEquals(changelogTopic.replicationFactor.get(), topicInfo.replicationFactor());
+ });
+ assertEquals(2, subtopology.copartitionGroups().size());
+ final StreamsGroupHeartbeatRequestData.CopartitionGroup expectedCopartitionGroupData1 =
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setRepartitionSourceTopics(Collections.singletonList((short) 0))
+ .setSourceTopics(Collections.singletonList((short) 1));
+ final StreamsGroupHeartbeatRequestData.CopartitionGroup expectedCopartitionGroupData2 =
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
+ .setRepartitionSourceTopics(Collections.singletonList((short) 1))
+ .setSourceTopics(Collections.singletonList((short) 0));
+ assertTrue(subtopology.copartitionGroups().contains(expectedCopartitionGroupData1));
+ assertTrue(subtopology.copartitionGroups().contains(expectedCopartitionGroupData2));
}
@Test
@@ -318,21 +379,6 @@ class StreamsGroupHeartbeatRequestManagerTest {
}
- @Test
- void testInitializeTopology() {
- mockJoiningState();
-
- StreamsGroupHeartbeatResponseData data = new StreamsGroupHeartbeatResponseData()
- .setErrorCode(Errors.NONE.code())
- .setThrottleTimeMs(0)
- .setMemberEpoch(TEST_MEMBER_EPOCH)
- .setShouldInitializeTopology(true);
-
- mockResponse(data);
-
- verify(streamsGroupInitializeRequestManager).initialize();
- }
-
private void mockResponse(final StreamsGroupHeartbeatResponseData data) {
NetworkClientDelegate.PollResult result = heartbeatRequestManager.poll(time.milliseconds());
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManagerTest.java
deleted file mode 100644
index 35eeaaf4baf..00000000000
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupInitializeRequestManagerTest.java
+++ /dev/null
@@ -1,152 +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.common.Node;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.protocol.ApiKeys;
-import org.apache.kafka.common.requests.StreamsGroupInitializeRequest;
-import org.apache.kafka.common.utils.LogContext;
-
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-
-import static org.apache.kafka.common.utils.Utils.mkEntry;
-import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-@ExtendWith(MockitoExtension.class)
-class StreamsGroupInitializeRequestManagerTest {
-
- private final String groupId = "groupId";
- private final LogContext logContext = new LogContext("test");
-
- @Test
- public void shouldPollEmptyResult() {
- final CoordinatorRequestManager coordinatorRequestManager = mock(CoordinatorRequestManager.class);
- final StreamsAssignmentInterface streamsAssignmentInterface = mock(StreamsAssignmentInterface.class);
- final StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager = new StreamsGroupInitializeRequestManager(
- logContext,
- groupId,
- streamsAssignmentInterface,
- coordinatorRequestManager
- );
-
- final NetworkClientDelegate.PollResult pollResult = streamsGroupInitializeRequestManager.poll(0);
-
- assertEquals(NetworkClientDelegate.PollResult.EMPTY, pollResult);
- }
-
- @Test
- public void shouldPollStreamsGroupInitializeRequest() {
- final Node node = mock(Node.class);
- final CoordinatorRequestManager coordinatorRequestManager = mock(CoordinatorRequestManager.class);
- when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(node));
- final StreamsAssignmentInterface streamsAssignmentInterface = mock(StreamsAssignmentInterface.class);
- final Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
- final Set<String> sinkTopics = Set.of("sinkTopic1", "sinkTopic2", "sinkTopic3");
- final Map<String, StreamsAssignmentInterface.TopicInfo> repartitionTopics = mkMap(
- mkEntry("repartitionTopic1", new StreamsAssignmentInterface.TopicInfo(Optional.of(2), Optional.of((short) 1), Collections.emptyMap())),
- mkEntry("repartitionTopic2", new StreamsAssignmentInterface.TopicInfo(Optional.of(3), Optional.of((short) 3), Collections.emptyMap()))
- );
- final Map<String, StreamsAssignmentInterface.TopicInfo> changelogTopics = mkMap(
- mkEntry("changelogTopic1", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 1), Collections.emptyMap())),
- mkEntry("changelogTopic2", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 2), Collections.emptyMap())),
- mkEntry("changelogTopic3", new StreamsAssignmentInterface.TopicInfo(Optional.empty(), Optional.of((short) 3), Collections.emptyMap()))
- );
- final Collection<Set<String>> copartitionGroup = Set.of(
- Set.of("sourceTopic1", "repartitionTopic2"),
- Set.of("sourceTopic2", "repartitionTopic1")
- );
- final StreamsAssignmentInterface.Subtopology subtopology1 = new StreamsAssignmentInterface.Subtopology(
- sourceTopics,
- sinkTopics,
- repartitionTopics,
- changelogTopics,
- copartitionGroup
- );
- final String subtopologyName1 = "subtopology1";
- when(streamsAssignmentInterface.subtopologyMap()).thenReturn(
- mkMap(mkEntry(subtopologyName1, subtopology1))
- );
- final StreamsGroupInitializeRequestManager streamsGroupInitializeRequestManager = new StreamsGroupInitializeRequestManager(
- logContext,
- groupId,
- streamsAssignmentInterface,
- coordinatorRequestManager
- );
-
- streamsGroupInitializeRequestManager.initialize();
- final NetworkClientDelegate.PollResult pollResult = streamsGroupInitializeRequestManager.poll(0);
-
- assertEquals(1, pollResult.unsentRequests.size());
- final NetworkClientDelegate.UnsentRequest unsentRequest = pollResult.unsentRequests.get(0);
- assertTrue(unsentRequest.node().isPresent());
- assertEquals(node, unsentRequest.node().get());
- assertEquals(ApiKeys.STREAMS_GROUP_INITIALIZE, unsentRequest.requestBuilder().apiKey());
- final StreamsGroupInitializeRequest.Builder streamsGroupInitializeRequestBuilder = (StreamsGroupInitializeRequest.Builder) unsentRequest.requestBuilder();
- final StreamsGroupInitializeRequest streamsGroupInitializeRequest = streamsGroupInitializeRequestBuilder.build();
- final StreamsGroupInitializeRequestData streamsGroupInitializeRequestData = streamsGroupInitializeRequest.data();
- assertEquals(ApiKeys.STREAMS_GROUP_INITIALIZE.id, streamsGroupInitializeRequestData.apiKey());
- assertEquals(groupId, streamsGroupInitializeRequestData.groupId());
- assertNotNull(streamsGroupInitializeRequestData.topology());
- final List<StreamsGroupInitializeRequestData.Subtopology> subtopologies = streamsGroupInitializeRequestData.topology();
- assertEquals(1, subtopologies.size());
- final StreamsGroupInitializeRequestData.Subtopology subtopology = subtopologies.get(0);
- assertEquals(subtopologyName1, subtopology.subtopologyId());
- assertEquals(Arrays.asList("sourceTopic1", "sourceTopic2"), subtopology.sourceTopics());
- assertEquals(Arrays.asList("sinkTopic1", "sinkTopic2", "sinkTopic3"), subtopology.repartitionSinkTopics());
- assertEquals(repartitionTopics.size(), subtopology.repartitionSourceTopics().size());
- subtopology.repartitionSourceTopics().forEach(topicInfo -> {
- final StreamsAssignmentInterface.TopicInfo repartitionTopic = repartitionTopics.get(topicInfo.name());
- assertEquals(repartitionTopic.numPartitions.get(), topicInfo.partitions());
- assertEquals(repartitionTopic.replicationFactor.get(), topicInfo.replicationFactor());
- });
- assertEquals(changelogTopics.size(), subtopology.stateChangelogTopics().size());
- subtopology.stateChangelogTopics().forEach(topicInfo -> {
- assertTrue(changelogTopics.containsKey(topicInfo.name()));
- assertEquals(0, topicInfo.partitions());
- final StreamsAssignmentInterface.TopicInfo changelogTopic = changelogTopics.get(topicInfo.name());
- assertEquals(changelogTopic.replicationFactor.get(), topicInfo.replicationFactor());
- });
-
- assertEquals(2, subtopology.copartitionGroups().size());
- final StreamsGroupInitializeRequestData.CopartitionGroup expectedCopartitionGroupData1 =
- new StreamsGroupInitializeRequestData.CopartitionGroup()
- .setRepartitionSourceTopics(Collections.singletonList((short) 0))
- .setSourceTopics(Collections.singletonList((short) 1));
- final StreamsGroupInitializeRequestData.CopartitionGroup expectedCopartitionGroupData2 =
- new StreamsGroupInitializeRequestData.CopartitionGroup()
- .setRepartitionSourceTopics(Collections.singletonList((short) 1))
- .setSourceTopics(Collections.singletonList((short) 0));
- assertTrue(subtopology.copartitionGroups().contains(expectedCopartitionGroupData1));
- assertTrue(subtopology.copartitionGroups().contains(expectedCopartitionGroupData2));
- }
-}
\ No newline at end of file
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 c56bef00c9f..25e41a88257 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
@@ -96,7 +96,6 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(commitRequestManager) : Optional.empty(),
withGroupId ? Optional.of(heartbeatRequestManager) : Optional.empty(),
withGroupId ? Optional.of(membershipManager) : Optional.empty(),
- Optional.empty(),
Optional.empty()
);
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index 1e5a1c4c864..4e07487a222 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -238,8 +238,6 @@ import org.apache.kafka.common.message.StreamsGroupDescribeRequestData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment;
import org.apache.kafka.common.message.SyncGroupResponseData;
@@ -1130,7 +1128,6 @@ public class RequestResponseTest {
case DELETE_SHARE_GROUP_STATE: return createDeleteShareGroupStateRequest(version);
case READ_SHARE_GROUP_STATE_SUMMARY: return createReadShareGroupStateSummaryRequest(version);
case STREAMS_GROUP_HEARTBEAT: return createStreamsGroupHeartbeatRequest(version);
- case STREAMS_GROUP_INITIALIZE: return createStreamsGroupInitializeRequest(version);
case STREAMS_GROUP_DESCRIBE: return createStreamsGroupDescribeRequest(version);
default: throw new IllegalArgumentException("Unknown API key " + apikey);
}
@@ -1227,7 +1224,6 @@ public class RequestResponseTest {
case DELETE_SHARE_GROUP_STATE: return createDeleteShareGroupStateResponse();
case READ_SHARE_GROUP_STATE_SUMMARY: return createReadShareGroupStateSummaryResponse();
case STREAMS_GROUP_HEARTBEAT: return createStreamsGroupHeartbeatResponse();
- case STREAMS_GROUP_INITIALIZE: return createStreamsGroupInitializeResponse();
case STREAMS_GROUP_DESCRIBE: return createStreamsGroupDescribeResponse();
default: throw new IllegalArgumentException("Unknown API key " + apikey);
}
@@ -4033,10 +4029,6 @@ public class RequestResponseTest {
.setIncludeAuthorizedOperations(false)).build(version);
}
- private AbstractRequest createStreamsGroupInitializeRequest(final short version) {
- return new StreamsGroupInitializeRequest.Builder(new StreamsGroupInitializeRequestData()).build(version);
- }
-
private AbstractRequest createStreamsGroupHeartbeatRequest(final short version) {
return new StreamsGroupHeartbeatRequest.Builder(new StreamsGroupHeartbeatRequestData()).build(version);
}
@@ -4058,10 +4050,6 @@ public class RequestResponseTest {
return new StreamsGroupDescribeResponse(data);
}
- private AbstractResponse createStreamsGroupInitializeResponse() {
- return new StreamsGroupInitializeResponse(new StreamsGroupInitializeResponseData());
- }
-
private AbstractResponse createStreamsGroupHeartbeatResponse() {
return new StreamsGroupHeartbeatResponse(new StreamsGroupHeartbeatResponseData());
}
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
index ed4322fe67a..58b1ac4ddf7 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
@@ -18,15 +18,15 @@ package kafka.coordinator.group
import kafka.server.{KafkaConfig, ReplicaManager}
import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
-import org.apache.kafka.common.message.{ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, Offset [...]
+import org.apache.kafka.common.message.{ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, Offset [...]
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.RecordBatch
import org.apache.kafka.common.requests.{OffsetCommitRequest, RequestContext, TransactionResult}
import org.apache.kafka.common.utils.{BufferSupplier, Time}
import org.apache.kafka.coordinator.group
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult
import org.apache.kafka.coordinator.group.OffsetAndMetadata
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult
import org.apache.kafka.image.{MetadataDelta, MetadataImage}
import org.apache.kafka.server.common.RequestLocal
import org.apache.kafka.server.util.FutureUtils
@@ -78,19 +78,10 @@ private[group] class GroupCoordinatorAdapter(
))
}
- override def streamsGroupInitialize(
- context: RequestContext,
- request: StreamsGroupInitializeRequestData
- ): CompletableFuture[StreamsGroupInitializeResult] = {
- FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- s"The old group coordinator does not support ${ApiKeys.STREAMS_GROUP_INITIALIZE.name} API."
- ))
- }
-
override def streamsGroupHeartbeat(
context: RequestContext,
request: StreamsGroupHeartbeatRequestData
- ): CompletableFuture[StreamsGroupHeartbeatResponseData] = {
+ ): CompletableFuture[StreamsGroupHeartbeatResult] = {
FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
s"The old group coordinator does not support ${ApiKeys.STREAMS_GROUP_HEARTBEAT.name} API."
))
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 31c5aaeeccf..dc45b3e3233 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -277,7 +277,6 @@ class KafkaApis(val requestChannel: RequestChannel,
case ApiKeys.DELETE_SHARE_GROUP_STATE => handleDeleteShareGroupStateRequest(request)
case ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY => handleReadShareGroupStateSummaryRequest(request)
case ApiKeys.STREAMS_GROUP_DESCRIBE => handleStreamsGroupDescribe(request).exceptionally(handleError)
- case ApiKeys.STREAMS_GROUP_INITIALIZE => handleStreamsGroupInitialize(request).exceptionally(handleError)
case ApiKeys.STREAMS_GROUP_HEARTBEAT => handleStreamsGroupHeartbeat(request).exceptionally(handleError)
case _ => throw new IllegalStateException(s"No handler for request api key ${request.header.apiKey}")
}
@@ -3890,103 +3889,87 @@ class KafkaApis(val requestChannel: RequestChannel,
config.groupCoordinatorRebalanceProtocols.contains(Group.GroupType.STREAMS)
}
- def handleStreamsGroupInitialize(request: RequestChannel.Request): CompletableFuture[Unit] = {
- // TODO: The unit tests for this method are insufficient. Once we merge initialize with group heartbeat, we have to extend the tests to cover ACLs and internal topic creation
- val streamsGroupInitializeRequest = request.body[StreamsGroupInitializeRequest]
+ def handleStreamsGroupHeartbeat(request: RequestChannel.Request): CompletableFuture[Unit] = {
+ val streamsGroupHeartbeatRequest = request.body[StreamsGroupHeartbeatRequest]
if (!isStreamsGroupProtocolEnabled()) {
// The API is not supported by the "old" group coordinator (the default). If the
// new one is not enabled, we fail directly here.
- requestHelper.sendMaybeThrottle(request, streamsGroupInitializeRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
CompletableFuture.completedFuture[Unit](())
- } else if (!authHelper.authorize(request.context, READ, GROUP, streamsGroupInitializeRequest.data.groupId)) {
- requestHelper.sendMaybeThrottle(request, streamsGroupInitializeRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
+ } else if (!authHelper.authorize(request.context, READ, GROUP, streamsGroupHeartbeatRequest.data.groupId)) {
+ requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
CompletableFuture.completedFuture[Unit](())
} else {
val requestContext = request.context
- val internalTopics: Map[String, StreamsGroupInitializeRequestData.TopicInfo] = {
- streamsGroupInitializeRequest.data().topology().asScala.flatMap(subtopology =>
- subtopology.repartitionSourceTopics().iterator().asScala ++ subtopology.stateChangelogTopics().iterator().asScala
- ).map(x => x.name() -> x).toMap
- }
-
- val prohibitedInternalTopics = internalTopics.keys.filter(Topic.isInternal)
- if (prohibitedInternalTopics.nonEmpty) {
- val errorResponse = new StreamsGroupInitializeResponseData()
- errorResponse.setErrorCode(Errors.STREAMS_INVALID_TOPOLOGY.code)
- errorResponse.setErrorMessage(f"Use of Kafka internal topics ${prohibitedInternalTopics.mkString(",")} as Kafka Streams internal topics is prohibited.")
- requestHelper.sendMaybeThrottle(request, new StreamsGroupInitializeResponse(errorResponse))
- return CompletableFuture.completedFuture[Unit](())
- }
+ if (streamsGroupHeartbeatRequest.data().topology() != null) {
+ val requiredTopics: Seq[String] =
+ streamsGroupHeartbeatRequest.data().topology().iterator().asScala.flatMap(subtopology =>
+ (subtopology.sourceTopics().iterator().asScala:Iterator[String])
+ ++ (subtopology.repartitionSinkTopics().iterator().asScala:Iterator[String])
+ ++ (subtopology.repartitionSourceTopics().iterator().asScala.map(_.name()):Iterator[String])
+ ++ (subtopology.stateChangelogTopics().iterator().asScala.map(_.name()):Iterator[String])
+ ).toSeq
+
+ // While correctness of the heartbeat request is checked inside the group coordinator,
+ // we are checking early that topics in the topology have valid names and are not internal
+ // kafka topics, since we need to pass it to the authorization helper before passing the
+ // request to the group coordinator.
+
+ val prohibitedTopics = requiredTopics.filter(Topic.isInternal)
+ if (prohibitedTopics.nonEmpty) {
+ val errorResponse = new StreamsGroupHeartbeatResponseData()
+ errorResponse.setErrorCode(Errors.STREAMS_INVALID_TOPOLOGY.code)
+ errorResponse.setErrorMessage(f"Use of Kafka internal topics ${prohibitedTopics.mkString(",")} in a Kafka Streams topology is prohibited.")
+ requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(errorResponse))
+ return CompletableFuture.completedFuture[Unit](())
+ }
- val invalidTopics = internalTopics.keys.filterNot(Topic.isValid)
- if (invalidTopics.nonEmpty) {
- val errorResponse = new StreamsGroupInitializeResponseData()
- errorResponse.setErrorCode(Errors.STREAMS_INVALID_TOPOLOGY.code)
- errorResponse.setErrorMessage(f"Internal topic names ${invalidTopics.mkString(",")} are not valid topic names.")
- requestHelper.sendMaybeThrottle(request, new StreamsGroupInitializeResponse(errorResponse))
- return CompletableFuture.completedFuture[Unit](())
- }
+ val invalidTopics = requiredTopics.filterNot(Topic.isValid)
+ if (invalidTopics.nonEmpty) {
+ val errorResponse = new StreamsGroupHeartbeatResponseData()
+ errorResponse.setErrorCode(Errors.STREAMS_INVALID_TOPOLOGY.code)
+ errorResponse.setErrorMessage(f"Topic names ${invalidTopics.mkString(",")} are not valid topic names.")
+ requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(errorResponse))
+ return CompletableFuture.completedFuture[Unit](())
+ }
- // TODO: Once we move initialization to the heartbeat, we should only require these permissions if there are missing internal topics.
- if(!authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME, logIfDenied = false)) {
- val (_, createTopicUnauthorized) = authHelper.partitionSeqByAuthorized(request.context, CREATE, TOPIC, internalTopics.keys.toSeq)(identity[String])
- if (createTopicUnauthorized.nonEmpty) {
- val errorResponse = new StreamsGroupInitializeResponseData()
+ val (_, describeConfigsAuthorized) = authHelper.partitionSeqByAuthorized(request.context, DESCRIBE_CONFIGS, TOPIC, requiredTopics.toSeq)(identity[String])
+ if (describeConfigsAuthorized.nonEmpty) {
+ val errorResponse = new StreamsGroupHeartbeatResponseData()
errorResponse.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
- errorResponse.setErrorMessage(f"Unauthorized to CREATE TOPIC ${createTopicUnauthorized.mkString(",")}.")
- requestHelper.sendMaybeThrottle(request, new StreamsGroupInitializeResponse(errorResponse))
+ errorResponse.setErrorMessage(f"Unauthorized to DESCRIBE_CONFIGS on topics ${describeConfigsAuthorized.mkString(",")} are unauthorized.")
+ requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(errorResponse))
return CompletableFuture.completedFuture[Unit](())
}
}
- val (_, describeConfigsAuthorized) = authHelper.partitionSeqByAuthorized(request.context, DESCRIBE_CONFIGS, TOPIC, internalTopics.keys.toSeq)(identity[String])
- if (describeConfigsAuthorized.nonEmpty) {
- val errorResponse = new StreamsGroupInitializeResponseData()
- errorResponse.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
- errorResponse.setErrorMessage(f"Unauthorized to DESCRIBE_CONFIGS on topics ${describeConfigsAuthorized.mkString(",")} are unauthorized.")
- requestHelper.sendMaybeThrottle(request, new StreamsGroupInitializeResponse(errorResponse))
- return CompletableFuture.completedFuture[Unit](())
- }
-
- groupCoordinator.streamsGroupInitialize(
+ groupCoordinator.streamsGroupHeartbeat(
request.context,
- streamsGroupInitializeRequest.data,
+ streamsGroupHeartbeatRequest.data,
).handle[Unit] { (response, exception) =>
if (exception != null) {
- requestHelper.sendMaybeThrottle(request, streamsGroupInitializeRequest.getErrorResponse(exception))
+ requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(exception))
} else {
+ var responseData = response.responseData();
if (!response.creatableTopics().isEmpty) {
- // TODO: Once we move this code to the heartbeat, we should indicate which topics are being created. We should also find a way to propagate failures to the client
+
+ if(!authHelper.authorize(request.context, CREATE, CLUSTER, CLUSTER_NAME, logIfDenied = false)) {
+ val (_, createTopicUnauthorized) = authHelper.partitionSeqByAuthorized(request.context, CREATE, TOPIC, response.creatableTopics().keySet().asScala.toSeq)(identity[String])
+ if (createTopicUnauthorized.nonEmpty) {
+ // TODO: Once topic validation is done, this should just add a status to the response
+ val errorResponse = new StreamsGroupHeartbeatResponseData()
+ errorResponse.setErrorCode(Errors.TOPIC_AUTHORIZATION_FAILED.code)
+ errorResponse.setErrorMessage(f"Unauthorized to CREATE on topics ${createTopicUnauthorized.mkString(",")}.")
+ responseData = new StreamsGroupHeartbeatResponse(errorResponse).data();
+ }
+ }
+
autoTopicCreationManager.createStreamsInternalTopics(response.creatableTopics().asScala, requestContext);
}
- requestHelper.sendMaybeThrottle(request, new StreamsGroupInitializeResponse(response.responseData()))
- }
- }
- }
- }
-
- def handleStreamsGroupHeartbeat(request: RequestChannel.Request): CompletableFuture[Unit] = {
- val streamsGroupHeartbeatRequest = request.body[StreamsGroupHeartbeatRequest]
- if (!isStreamsGroupProtocolEnabled()) {
- // The API is not supported by the "old" group coordinator (the default). If the
- // new one is not enabled, we fail directly here.
- requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
- CompletableFuture.completedFuture[Unit](())
- } else if (!authHelper.authorize(request.context, READ, GROUP, streamsGroupHeartbeatRequest.data.groupId)) {
- requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
- CompletableFuture.completedFuture[Unit](())
- } else {
- groupCoordinator.streamsGroupHeartbeat(
- request.context,
- streamsGroupHeartbeatRequest.data,
- ).handle[Unit] { (response, exception) =>
- if (exception != null) {
- requestHelper.sendMaybeThrottle(request, streamsGroupHeartbeatRequest.getErrorResponse(exception))
- } else {
- requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(response))
+ requestHelper.sendMaybeThrottle(request, new StreamsGroupHeartbeatResponse(responseData))
}
}
}
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
index 68959acc23a..74feeaa7ac2 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
@@ -19,7 +19,7 @@ package kafka.coordinator.group
import kafka.coordinator.group.GroupCoordinatorConcurrencyTest.{JoinGroupCallback, SyncGroupCallback}
import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.common.errors.{InvalidGroupIdException, UnsupportedVersionException}
-import org.apache.kafka.common.message.{ConsumerGroupHeartbeatRequestData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchRequestData, OffsetFetchResponseData, ShareGroupHeartbeatRequestDa [...]
+import org.apache.kafka.common.message.{ConsumerGroupHeartbeatRequestData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchRequestData, OffsetFetchResponseData, ShareGroupHeartbeatRequestDa [...]
import org.apache.kafka.common.message.JoinGroupRequestData.JoinGroupRequestProtocol
import org.apache.kafka.common.message.JoinGroupResponseData.JoinGroupResponseMember
import org.apache.kafka.common.message.OffsetDeleteRequestData.{OffsetDeleteRequestPartition, OffsetDeleteRequestTopic, OffsetDeleteRequestTopicCollection}
@@ -78,22 +78,6 @@ class GroupCoordinatorAdapterTest {
assertFutureThrows(future, classOf[UnsupportedVersionException])
}
- @Test
- def testStreamsGroupInitialize(): Unit = {
- val groupCoordinator = mock(classOf[GroupCoordinator])
- val adapter = new GroupCoordinatorAdapter(groupCoordinator, Time.SYSTEM)
-
- val ctx = makeContext(ApiKeys.STREAMS_GROUP_INITIALIZE, ApiKeys.STREAMS_GROUP_INITIALIZE.latestVersion)
- val request = new StreamsGroupInitializeRequestData()
- .setGroupId("group")
-
- val future = adapter.streamsGroupInitialize(ctx, request)
-
- assertTrue(future.isDone)
- assertTrue(future.isCompletedExceptionally)
- assertFutureThrows(future, classOf[UnsupportedVersionException])
- }
-
@Test
def testStreamsGroupHeartbeat(): Unit = {
val groupCoordinator = mock(classOf[GroupCoordinator])
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index d31e6fb27e7..d3edc955211 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -78,7 +78,7 @@ import org.apache.kafka.coordinator.group.GroupConfig.{CONSUMER_HEARTBEAT_INTERV
import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig
import org.apache.kafka.coordinator.group.{GroupConfig, GroupCoordinator, GroupCoordinatorConfig}
import org.apache.kafka.coordinator.group.{GroupCoordinator, GroupCoordinatorConfig}
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult
import org.apache.kafka.coordinator.share.{ShareCoordinator, ShareCoordinatorConfigTest}
import org.apache.kafka.coordinator.transaction.TransactionLogConfig
import org.apache.kafka.metadata.LeaderAndIsr
@@ -11237,17 +11237,17 @@ class KafkaApisTest extends Logging {
}
@Test
- def testStreamsGroupInitializeRequest(): Unit = {
+ def testStreamsGroupHeartbeatRequest(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
- val streamsGroupInitializeRequest = new StreamsGroupInitializeRequestData().setGroupId("group")
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
- val requestChannelRequest = buildRequest(new StreamsGroupInitializeRequest.Builder(streamsGroupInitializeRequest, true).build())
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
- val future = new CompletableFuture[StreamsGroupInitializeResult]()
- when(groupCoordinator.streamsGroupInitialize(
+ val future = new CompletableFuture[StreamsGroupHeartbeatResult]()
+ when(groupCoordinator.streamsGroupHeartbeat(
requestChannelRequest.context,
- streamsGroupInitializeRequest
+ streamsGroupHeartbeatRequest
)).thenReturn(future)
kafkaApis = createKafkaApis(
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
@@ -11255,25 +11255,26 @@ class KafkaApisTest extends Logging {
)
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
- val streamsGroupInitializeResponse = new StreamsGroupInitializeResult(new StreamsGroupInitializeResponseData())
+ val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
+ .setMemberId("member")
- future.complete(streamsGroupInitializeResponse)
- val response = verifyNoThrottling[StreamsGroupInitializeResponse](requestChannelRequest)
- assertEquals(streamsGroupInitializeResponse, response.data)
+ future.complete(new StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse))
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
+ assertEquals(streamsGroupHeartbeatResponse, response.data)
}
@Test
- def testStreamsGroupInitializeRequestFutureFailed(): Unit = {
+ def testStreamsGroupHeartbeatRequestFutureFailed(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
- val streamsGroupInitializeRequest = new StreamsGroupInitializeRequestData().setGroupId("group")
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
- val requestChannelRequest = buildRequest(new StreamsGroupInitializeRequest.Builder(streamsGroupInitializeRequest, true).build())
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
- val future = new CompletableFuture[StreamsGroupInitializeResult]()
- when(groupCoordinator.streamsGroupInitialize(
+ val future = new CompletableFuture[StreamsGroupHeartbeatResult]()
+ when(groupCoordinator.streamsGroupHeartbeat(
requestChannelRequest.context,
- streamsGroupInitializeRequest
+ streamsGroupHeartbeatRequest
)).thenReturn(future)
kafkaApis = createKafkaApis(
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
@@ -11282,17 +11283,17 @@ class KafkaApisTest extends Logging {
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
- val response = verifyNoThrottling[StreamsGroupInitializeResponse](requestChannelRequest)
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(Errors.FENCED_MEMBER_EPOCH.code, response.data.errorCode)
}
@Test
- def testStreamsGroupInitializeRequestAuthorizationFailed(): Unit = {
+ def testStreamsGroupHeartbeatRequestAuthorizationFailed(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
- val streamsGroupInitializeRequest = new StreamsGroupInitializeRequestData().setGroupId("group")
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
- val requestChannelRequest = buildRequest(new StreamsGroupInitializeRequest.Builder(streamsGroupInitializeRequest, true).build())
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
val authorizer: Authorizer = mock(classOf[Authorizer])
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
@@ -11304,73 +11305,174 @@ class KafkaApisTest extends Logging {
)
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
- val response = verifyNoThrottling[StreamsGroupInitializeResponse](requestChannelRequest)
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.errorCode)
}
-
@Test
- def testStreamsGroupHeartbeatRequest(): Unit = {
+ def testStreamsGroupHeartbeatRequestProtocolDisabled(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
- val future = new CompletableFuture[StreamsGroupHeartbeatResponseData]()
- when(groupCoordinator.streamsGroupHeartbeat(
- requestChannelRequest.context,
- streamsGroupHeartbeatRequest
- )).thenReturn(future)
+ kafkaApis = createKafkaApis(
+ overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,consumer"),
+ raftSupport = true
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
+ assertEquals(Errors.UNSUPPORTED_VERSION.code, response.data.errorCode)
+ }
+
+ @Test
+ def testStreamsGroupHeartbeatRequestInvalidTopicNames(): Unit = {
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group").setTopology(
+ Collections.singletonList(new StreamsGroupHeartbeatRequestData.Subtopology().setSubtopologyId("subtopology")
+ .setSourceTopics(Collections.singletonList("a "))
+ .setRepartitionSinkTopics(Collections.singletonList("b?"))
+ .setRepartitionSourceTopics(Collections.singletonList(new StreamsGroupHeartbeatRequestData.TopicInfo().setName("c!")))
+ .setStateChangelogTopics(Collections.singletonList(new StreamsGroupHeartbeatRequestData.TopicInfo().setName("d/")))
+ )
+ )
+
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
+
kafkaApis = createKafkaApis(
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
raftSupport = true
)
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
- val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
- .setMemberId("member")
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
+ assertEquals(Errors.STREAMS_INVALID_TOPOLOGY.code, response.data.errorCode)
+ assertEquals("Topic names a ,b?,c!,d/ are not valid topic names.", response.data.errorMessage())
+ }
+
+ @Test
+ def testStreamsGroupHeartbeatRequestInternalTopicNames(): Unit = {
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group").setTopology(
+ Collections.singletonList(new StreamsGroupHeartbeatRequestData.Subtopology().setSubtopologyId("subtopology")
+ .setSourceTopics(Collections.singletonList("__consumer_offsets"))
+ .setRepartitionSinkTopics(Collections.singletonList("__transaction_state"))
+ .setRepartitionSourceTopics(Collections.singletonList(new StreamsGroupHeartbeatRequestData.TopicInfo().setName("__share_group_state")))
+ )
+ )
+
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
+
+ kafkaApis = createKafkaApis(
+ overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
+ raftSupport = true
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
- future.complete(streamsGroupHeartbeatResponse)
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
- assertEquals(streamsGroupHeartbeatResponse, response.data)
+ assertEquals(Errors.STREAMS_INVALID_TOPOLOGY.code, response.data.errorCode)
+ assertEquals("Use of Kafka internal topics __consumer_offsets,__transaction_state,__share_group_state in a Kafka Streams topology is prohibited.", response.data.errorMessage())
}
@Test
- def testStreamsGroupHeartbeatRequestFutureFailed(): Unit = {
+ def testStreamsGroupHeartbeatRequestMissingDescribeConfigsPermission(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
- val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group").setTopology(
+ Collections.singletonList(new StreamsGroupHeartbeatRequestData.Subtopology().setSubtopologyId("subtopology")
+ .setSourceTopics(Collections.singletonList("a"))
+ .setRepartitionSinkTopics(Collections.singletonList("b"))
+ .setRepartitionSourceTopics(Collections.singletonList(new StreamsGroupHeartbeatRequestData.TopicInfo().setName("c")))
+ .setStateChangelogTopics(Collections.singletonList(new StreamsGroupHeartbeatRequestData.TopicInfo().setName("d")))
+ )
+ )
+
+ val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
+
+ val authorizer: Authorizer = mock(classOf[Authorizer])
+ when(authorizer.authorize(any[RequestContext], isNotNull[util.List[Action]])).thenAnswer(invocation => {
+ val actions = invocation.getArgument(1).asInstanceOf[util.List[Action]]
+ actions.asScala.map { action =>
+ if (!action.resourcePattern.name.equals("a") && action.operation() == AclOperation.DESCRIBE_CONFIGS) {
+ AuthorizationResult.DENIED
+ } else {
+ AuthorizationResult.ALLOWED
+ }
+ }.asJava
+ })
+ kafkaApis = createKafkaApis(
+ authorizer = Some(authorizer),
+ overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
+ raftSupport = true
+ )
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
+ assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED.code, response.data.errorCode)
+ assertEquals("Unauthorized to DESCRIBE_CONFIGS on topics b,c,d are unauthorized.", response.data.errorMessage())
+ }
+
+ @Test
+ def testStreamsGroupHeartbeatRequestMissingTopics(): Unit = {
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group");
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
- val future = new CompletableFuture[StreamsGroupHeartbeatResponseData]()
+ val future = new CompletableFuture[StreamsGroupHeartbeatResult]()
when(groupCoordinator.streamsGroupHeartbeat(
requestChannelRequest.context,
streamsGroupHeartbeatRequest
)).thenReturn(future)
+
kafkaApis = createKafkaApis(
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
raftSupport = true
)
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
- future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
+ val missingTopics = Map("test" -> new CreatableTopic())
+ val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
+ .setMemberId("member")
+
+ future.complete(new StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics.asJava))
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
- assertEquals(Errors.FENCED_MEMBER_EPOCH.code, response.data.errorCode)
+ assertEquals(streamsGroupHeartbeatResponse, response.data)
+ verify(autoTopicCreationManager).createStreamsInternalTopics(missingTopics, requestChannelRequest.context)
}
@Test
- def testStreamsGroupHeartbeatRequestAuthorizationFailed(): Unit = {
+ def testStreamsGroupHeartbeatRequestMissingTopicsMissingCreateACL(): Unit = {
metadataCache = mock(classOf[KRaftMetadataCache])
- val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
+ val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group");
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
+ val future = new CompletableFuture[StreamsGroupHeartbeatResult]()
+ when(groupCoordinator.streamsGroupHeartbeat(
+ requestChannelRequest.context,
+ streamsGroupHeartbeatRequest
+ )).thenReturn(future)
+
val authorizer: Authorizer = mock(classOf[Authorizer])
- when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
- .thenReturn(Seq(AuthorizationResult.DENIED).asJava)
+ when(authorizer.authorize(any[RequestContext], isNotNull[util.List[Action]])).thenAnswer(invocation => {
+ val actions = invocation.getArgument(1).asInstanceOf[util.List[Action]]
+ actions.asScala.map { action =>
+ if (action.resourcePattern.name.equals("test") && action.operation() == AclOperation.CREATE && action.resourcePattern().resourceType() == ResourceType.TOPIC) {
+ AuthorizationResult.DENIED
+ } else if (action.operation() == AclOperation.CREATE && action.resourcePattern().resourceType() == ResourceType.CLUSTER) {
+ AuthorizationResult.DENIED
+ } else {
+ AuthorizationResult.ALLOWED
+ }
+ }.asJava
+ })
kafkaApis = createKafkaApis(
authorizer = Some(authorizer),
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams"),
@@ -11378,8 +11480,14 @@ class KafkaApisTest extends Logging {
)
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+ val missingTopics = Map("test" -> new CreatableTopic())
+ val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
+ .setMemberId("member")
+
+ future.complete(new StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics.asJava))
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
- assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.errorCode)
+ assertEquals(Errors.TOPIC_AUTHORIZATION_FAILED.code, response.data.errorCode)
+ assertEquals("Unauthorized to CREATE on topics test.", response.data.errorMessage())
}
@ParameterizedTest
diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
index 154814e35c7..862e0249ba0 100644
--- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
+++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala
@@ -745,9 +745,6 @@ class RequestQuotaTest extends BaseRequestTest {
case ApiKeys.STREAMS_GROUP_HEARTBEAT =>
new StreamsGroupHeartbeatRequest.Builder(new StreamsGroupHeartbeatRequestData(), true)
- case ApiKeys.STREAMS_GROUP_INITIALIZE =>
- new StreamsGroupInitializeRequest.Builder(new StreamsGroupInitializeRequestData(), true)
-
case _ =>
throw new IllegalArgumentException("Unsupported API key " + apiKey)
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
index 07d5d7daaee..b5a1cacbdca 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
@@ -41,8 +41,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
-import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -50,7 +48,7 @@ import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.TransactionResult;
import org.apache.kafka.common.utils.BufferSupplier;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
@@ -87,30 +85,16 @@ public interface GroupCoordinator {
ConsumerGroupHeartbeatRequestData request
);
- /**
- * Initialize a Streams Group.
- *
- * @param context The request context.
- * @param request The StreamsGroupInitializeRequest data.
- *
- * @return A future yielding the result, which contains the response and all topics to be created.
- * The error code(s) of the response are set to indicate the error(s) occurred during the execution.
- */
- CompletableFuture<StreamsGroupInitializeResult> streamsGroupInitialize(
- RequestContext context,
- StreamsGroupInitializeRequestData request
- );
-
/**
* Heartbeat to a Streams Group.
*
* @param context The request context.
* @param request The StreamsGroupHeartbeatResponseData data.
*
- * @return A future yielding the response.
+ * @return A future yielding the response together with internal topics to create.
* The error code(s) of the response are set to indicate the error(s) occurred during the execution.
*/
- CompletableFuture<StreamsGroupHeartbeatResponseData> streamsGroupHeartbeat(
+ CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
RequestContext context,
StreamsGroupHeartbeatRequestData request
);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 3f2f4920162..e775a977b6a 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -47,8 +47,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -78,7 +76,7 @@ import org.apache.kafka.coordinator.common.runtime.CoordinatorShardBuilderSuppli
import org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor;
import org.apache.kafka.coordinator.common.runtime.PartitionWriter;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.record.BrokerCompressionType;
@@ -351,48 +349,18 @@ public class GroupCoordinatorService implements GroupCoordinator {
));
}
- /**
- * See {@link GroupCoordinator#streamsGroupInitialize(RequestContext, org.apache.kafka.common.message.StreamsGroupInitializeRequestData)}.
- */
- @Override
- public CompletableFuture<StreamsGroupInitializeResult> streamsGroupInitialize(
- RequestContext context,
- StreamsGroupInitializeRequestData request
- ) {
- if (!isActive.get()) {
- return CompletableFuture.completedFuture(new StreamsGroupInitializeResult(new StreamsGroupInitializeResponseData()
- .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
- ));
- }
-
- return runtime.scheduleWriteOperation(
- "streams-group-initialize",
- topicPartitionFor(request.groupId()),
- Duration.ofMillis(config.offsetCommitTimeoutMs()),
- coordinator -> coordinator.streamsGroupInitialize(context, request)
- ).exceptionally(exception -> handleOperationException(
- "streams-group-initialize",
- request,
- exception,
- (error, message) -> new StreamsGroupInitializeResult(new StreamsGroupInitializeResponseData()
- .setErrorCode(error.code())
- .setErrorMessage(message)),
- log
- ));
- }
-
/**
* See {@link GroupCoordinator#streamsGroupHeartbeat(RequestContext, org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData)}.
*/
@Override
- public CompletableFuture<StreamsGroupHeartbeatResponseData> streamsGroupHeartbeat(
+ public CompletableFuture<StreamsGroupHeartbeatResult> streamsGroupHeartbeat(
RequestContext context,
StreamsGroupHeartbeatRequestData request
) {
if (!isActive.get()) {
- return CompletableFuture.completedFuture(new StreamsGroupHeartbeatResponseData()
+ return CompletableFuture.completedFuture(new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData()
.setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
- );
+ ));
}
return runtime.scheduleWriteOperation(
@@ -404,9 +372,12 @@ public class GroupCoordinatorService implements GroupCoordinator {
"streams-heartbeat",
request,
exception,
- (error, message) -> new StreamsGroupHeartbeatResponseData()
- .setErrorCode(error.code())
- .setErrorMessage(message),
+ (error, message) ->
+ new StreamsGroupHeartbeatResult(
+ new StreamsGroupHeartbeatResponseData()
+ .setErrorCode(error.code())
+ .setErrorMessage(message)
+ ),
log
));
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index 1e67ca0f882..895a7508696 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -41,8 +41,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
-import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -106,13 +104,12 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.taskassignor.StickyTaskAssignor;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.timeline.SnapshotRegistry;
-
import org.slf4j.Logger;
import java.util.ArrayList;
@@ -389,32 +386,16 @@ public class GroupCoordinatorShard implements CoordinatorShard<CoordinatorRecord
return groupMetadataManager.consumerGroupHeartbeat(context, request);
}
- /**
- * Handles a StreamsGroupInitialize request.
- *
- * @param context The request context.
- * @param request The actual StreamsGroupInitialize request.
- *
- * @return A Result containing the StreamsGroupInitialize response and
- * a list of records to update the state machine.
- */
- public CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> streamsGroupInitialize(
- RequestContext context,
- StreamsGroupInitializeRequestData request
- ) {
- return groupMetadataManager.streamsGroupInitialize(context, request);
- }
-
/**
* Handles a StreamsGroupHeartbeat request.
*
* @param context The request context.
* @param request The actual StreamsGroupHeartbeat request.
*
- * @return A Result containing the StreamsGroupHeartbeat response and
+ * @return A Result containing the StreamsGroupHeartbeat response, a list of internal topics to be created and
* a list of records to update the state machine.
*/
- public CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupHeartbeat(
+ public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
RequestContext context,
StreamsGroupHeartbeatRequestData request
) {
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 78e433203a4..9b9602d085f 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -61,8 +61,6 @@ import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.KeyValue
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TaskIds;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TaskOffset;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.protocol.Errors;
@@ -141,11 +139,9 @@ import org.apache.kafka.coordinator.group.modern.share.ShareGroup;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupAssignmentBuilder;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.streams.StreamsGroup;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import org.apache.kafka.coordinator.group.streams.StreamsTopology;
-import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager;
-import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
import org.apache.kafka.coordinator.group.taskassignor.TaskAssignor;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
@@ -1017,7 +1013,7 @@ public class GroupMetadataManager {
if (group == null || (createIfNotExists && maybeDeleteEmptyClassicGroup(group, records))) {
log.info("Creating streams group {}", groupId);
- return new StreamsGroup(snapshotRegistry, groupId, metrics);
+ return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
} else {
if (group.type() == STREAMS) {
return (StreamsGroup) group;
@@ -1163,7 +1159,7 @@ public class GroupMetadataManager {
if (group == null) {
log.info("Creating persisted streams group {}", groupId);
- StreamsGroup streamsGroup = new StreamsGroup(snapshotRegistry, groupId, metrics);
+ StreamsGroup streamsGroup = new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
groups.put(groupId, streamsGroup);
metrics.onStreamsGroupStateTransition(null, streamsGroup.state());
return streamsGroup;
@@ -1667,27 +1663,6 @@ public class GroupMetadataManager {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
}
-
- /**
- * Validates the request.
- *
- * @param request The request to validate.
- *
- * @throws InvalidRequestException if the request is not valid.
- * @throws UnsupportedAssignorException if the assignor is not supported.
- */
- private void throwIfStreamsGroupInitializeRequestIsInvalid(
- StreamsGroupInitializeRequestData request
- ) throws InvalidRequestException, UnsupportedAssignorException {
- throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
-
- if (request.topology().isEmpty()) {
- throw new InvalidRequestException("There must be at least one subtopology.");
- }
-
- // TODO: Check invariants
- }
-
/**
/**
@@ -1728,12 +1703,8 @@ public class GroupMetadataManager {
throw new InvalidRequestException("MemberEpoch is invalid.");
}
- // TODO: Check that task assignor exists.
-// if (request.assignor() != null && !assignors.containsKey(request.assignor())) {
-// throw new UnsupportedAssignorException("Assignor " + request.assignor()
-// + " is not supported. Supported assignors: " + String.join(", ", assignors.keySet())
-// + ".");
-// }
+ // TODO: Check supplied topology for invariants
+
}
/**
@@ -2249,7 +2220,7 @@ public class GroupMetadataManager {
* @param clientTags
* @return A Result containing the StreamsGroupHeartbeat response and a list of records to update the state machine.
*/
- private CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupHeartbeat(
+ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
String groupId,
String memberId,
int memberEpoch,
@@ -2257,6 +2228,7 @@ public class GroupMetadataManager {
String rackId,
int rebalanceTimeoutMs,
final String topologyId,
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> clientTopology,
String clientId,
String clientHost,
List<TaskIds> ownedActiveTasks,
@@ -2340,7 +2312,7 @@ public class GroupMetadataManager {
.build();
int groupEpoch = group.groupEpoch();
- Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata = group.subscriptionMetadata();
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> partitionMetadata = group.partitionMetadata();
boolean bumpGroupEpoch = hasStreamsMemberMetadataChanged(
groupId,
@@ -2349,18 +2321,40 @@ public class GroupMetadataManager {
records
);
- if (bumpGroupEpoch || group.hasMetadataExpired(currentTimeMs)) {
- // The subscription metadata is updated when the refresh deadline has been reached.
- subscriptionMetadata = group.computeSubscriptionMetadata(
+ // 2. Initialize/Update the group topology.
+ // TODO: Topology updates will come later
+ if (clientTopology != null && group.topology() == null) {
+
+ log.info("[GroupId {}] Member {} initialized the topology {}", groupId, memberId, topologyId);
+
+ StreamsGroupTopologyValue recordValue = convertToStreamsGroupTopologyRecord(clientTopology);
+
+ records.add(newStreamsGroupTopologyRecord(groupId, recordValue));
+
+ final Map<String, StreamsGroupTopologyValue.Subtopology> subtopologyMap = recordValue.topology().stream()
+ .collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
+ final StreamsTopology topology = new StreamsTopology(topologyId, subtopologyMap);
+
+ group.setTopology(topology);
+
+ bumpGroupEpoch = true;
+ }
+
+ // 3. Configure the partition metadata and create internal topics if needed.
+ if (bumpGroupEpoch || group.hasMetadataExpired(currentTimeMs) || group.configuredTopology() == null || !group.configuredTopology().isReady()) {
+
+ // The partition metadata is updated when the refresh deadline has been reached.
+ partitionMetadata = group.computePartitionMetadata(
metadataImage.topics(),
metadataImage.cluster()
);
- if (!subscriptionMetadata.equals(group.subscriptionMetadata())) {
+ if (!partitionMetadata.equals(group.partitionMetadata())) {
log.info("[GroupId {}] Computed new partition metadata: {}.",
- groupId, subscriptionMetadata);
+ groupId, partitionMetadata);
bumpGroupEpoch = true;
- records.add(newStreamsGroupPartitionMetadataRecord(groupId, subscriptionMetadata));
+ records.add(newStreamsGroupPartitionMetadataRecord(groupId, partitionMetadata));
+ group.setPartitionMetadata(partitionMetadata);
}
if (bumpGroupEpoch) {
@@ -2383,7 +2377,7 @@ public class GroupMetadataManager {
groupEpoch,
member,
updatedMember,
- subscriptionMetadata,
+ partitionMetadata,
staticMemberReplaced,
records
);
@@ -2408,18 +2402,10 @@ public class GroupMetadataManager {
scheduleStreamsGroupSessionTimeout(groupId, memberId);
- boolean requestTopologyInitialization = false;
- if (group.topology() == null && !isTopologyInitializationScheduled(groupId, topologyId)) {
- requestTopologyInitialization = true;
- scheduleStreamsGroupTopologyInitializationTimeout(groupId, topologyId, memberId, rebalanceTimeoutMs);
- log.info("Asking member {} at {} to initialize the topology", memberId, clientHost);
- }
-
// Prepare the response.
StreamsGroupHeartbeatResponseData response = new StreamsGroupHeartbeatResponseData()
.setMemberId(updatedMember.memberId())
.setMemberEpoch(updatedMember.memberEpoch())
- .setShouldInitializeTopology(requestTopologyInitialization)
.setHeartbeatIntervalMs(streamsGroupHeartbeatIntervalMs);
// The assignment is only provided in the following cases:
@@ -2436,7 +2422,14 @@ public class GroupMetadataManager {
response.setWarmupTasks(createStreamsGroupHeartbeatResponseTaskIds(updatedMember.assignedWarmupTasks()));
}
- return new CoordinatorResult<>(records, response);
+
+ Map<String, CreatableTopic> internalTopicsToBeCreated = Collections.emptyMap();
+ if (group.configuredTopology() != null) {
+ internalTopicsToBeCreated = group.configuredTopology().internalTopicsToBeCreated();
+ // TODO: Internal topic validation status
+ }
+
+ return new CoordinatorResult<>(records, new StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated));
}
private List<StreamsGroupHeartbeatResponseData.TaskIds> createStreamsGroupHeartbeatResponseTaskIds(final Map<String, Set<Integer>> taskIds) {
@@ -2658,108 +2651,6 @@ public class GroupMetadataManager {
return new CoordinatorResult<>(records, response);
}
-
- /**
- * Handles the initialization of the topology information on the broker side, that will be reused by all members of the group.
- *
- * @param groupId The group id from the request.
- * @param topologyId The topology ID.
- * @param subtopologies The list of subtopologies.
- * @return A Result containing the StreamsGroupInitialize response and a list of records to update the state machine.
- */
- private CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> streamsGroupInitialize(String groupId,
- String topologyId,
- List<StreamsGroupInitializeRequestData.Subtopology> subtopologies)
- throws ApiException {
-
- final List<CoordinatorRecord> records = new ArrayList<>();
-
- log.info("Initializing topology {} for group {} to {}", topologyId, groupId, subtopologies);
-
- final StreamsGroup group = getOrMaybeCreateStreamsGroup(groupId, false, records);
- throwIfNull(group, "group does not exist");
-
- if (!isTopologyInitializationScheduled(groupId, topologyId)) {
- log.warn("No topology to initialize for group ID {} and topology ID {} found.", groupId, topologyId);
- StreamsGroupInitializeResponseData response = new StreamsGroupInitializeResponseData();
- return new CoordinatorResult<>(records, new StreamsGroupInitializeResult(response));
- }
-
- StreamsGroupTopologyValue recordValue = convertToStreamsGroupTopologyRecord(subtopologies);
-
- final Map<String, ConfiguredSubtopology> configuredTopics =
- InternalTopicManager.configureTopics(logContext, recordValue.topology(), metadataImage);
-
- cancelStreamsGroupTopologyInitializationTimeout(groupId, topologyId);
-
- records.add(newStreamsGroupTopologyRecord(groupId, recordValue));
-
- final Map<String, StreamsGroupTopologyValue.Subtopology> subtopologyMap = recordValue.topology().stream()
- .collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
- final StreamsTopology topology = new StreamsTopology(topologyId, subtopologyMap);
-
- final Map<String, CreatableTopic> missingTopics = InternalTopicManager.missingTopics(configuredTopics, metadataImage);
- // TODO: This needs to be sorted out, once we merge heartbeat and initialization
- if (missingTopics.isEmpty()) {
- computeFirstTargetAssignmentAfterTopologyInitialization(group, records, topology);
- }
-
- StreamsGroupInitializeResponseData response = new StreamsGroupInitializeResponseData();
-
- return new CoordinatorResult<>(records, new StreamsGroupInitializeResult(response, missingTopics));
-
- }
-
- // ToDo: verify how much code we can share with Streams heartbeat handler
- private void computeFirstTargetAssignmentAfterTopologyInitialization(StreamsGroup group,
- List<CoordinatorRecord> records,
- StreamsTopology topology) {
- String groupId = group.groupId();
- final long currentTimeMs = time.milliseconds();
- int groupEpoch = group.groupEpoch();
-
- Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata =
- group.computeSubscriptionMetadata(
- metadataImage.topics(),
- metadataImage.cluster(),
- topology
- );
- if (!subscriptionMetadata.equals(group.subscriptionMetadata())) {
- log.info("[GroupId {}] Computed new partition metadata: {}.", groupId, subscriptionMetadata);
- records.add(newStreamsGroupPartitionMetadataRecord(groupId, subscriptionMetadata));
- }
- groupEpoch += 1;
- records.add(newStreamsGroupEpochRecord(groupId, groupEpoch));
- log.info("[GroupId {}] Bumped streams group epoch to {}.", groupId, groupEpoch);
- metrics.record(STREAMS_GROUP_REBALANCES_SENSOR_NAME);
- group.setMetadataRefreshDeadline(currentTimeMs + streamsGroupMetadataRefreshIntervalMs, groupEpoch);
-
- // TODO: Read the preferred server assignor from the group configuration
- String preferredServerAssignor = defaultTaskAssignor.name();
- try {
- org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder assignmentResultBuilder =
- new org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder(group.groupId(), groupEpoch, taskAssignors.get(preferredServerAssignor))
- .withMembers(group.members())
- .withTopology(topology)
- .withStaticMembers(group.staticMembers())
- .withSubscriptionMetadata(subscriptionMetadata)
- .withTargetAssignment(group.targetAssignment());
- org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult assignmentResult =
- assignmentResultBuilder
- .build();
-
- log.info("[GroupId {}] Computed a new target assignment for epoch {} with '{}' assignor: {}.",
- group.groupId(), groupEpoch, preferredServerAssignor, assignmentResult.targetAssignment());
-
- records.addAll(assignmentResult.records());
- } catch (PartitionAssignorException ex) {
- String msg = String.format("Failed to compute a new target assignment for epoch %d: %s",
- groupEpoch, ex.getMessage());
- log.error("[GroupId {}] {}.", group.groupId(), msg);
- throw new UnknownServerException(msg, ex);
- }
- }
-
/**
* Handle a JoinGroupRequest to a ConsumerGroup.
*
@@ -3945,7 +3836,7 @@ public class GroupMetadataManager {
.withMembers(group.members())
.withTopology(group.topology())
.withStaticMembers(group.staticMembers())
- .withSubscriptionMetadata(subscriptionMetadata)
+ .withPartitionMetadata(subscriptionMetadata)
.withTargetAssignment(group.targetAssignment())
.addOrUpdateMember(updatedMember.memberId(), updatedMember);
org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult assignmentResult;
@@ -4023,7 +3914,7 @@ public class GroupMetadataManager {
* @return A Result containing the StreamsGroupHeartbeat response and
* a list of records to update the state machine.
*/
- private CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupLeave(
+ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupLeave(
String groupId,
String instanceId,
String memberId,
@@ -4037,7 +3928,7 @@ public class GroupMetadataManager {
if (instanceId == null) {
StreamsGroupMember member = group.getOrMaybeCreateMember(memberId, false);
log.info("[GroupId {}] Member {} left the streams group.", groupId, memberId);
- return streamsGroupFenceMember(group, member, response);
+ return streamsGroupFenceMember(group, member, new StreamsGroupHeartbeatResult(response));
} else {
StreamsGroupMember member = group.staticMember(instanceId);
throwIfStaticMemberIsUnknown(member, instanceId);
@@ -4049,7 +3940,7 @@ public class GroupMetadataManager {
} else {
log.info("[GroupId {}] Static Member {} with instance id {} left the streams group.",
group.groupId(), memberId, instanceId);
- return streamsGroupFenceMember(group, member, response);
+ return streamsGroupFenceMember(group, member, new StreamsGroupHeartbeatResult(response));
}
}
}
@@ -4121,7 +4012,7 @@ public class GroupMetadataManager {
*
* @return A CoordinatorResult with a single record signifying that the static member is leaving.
*/
- private CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupStaticMemberGroupLeave(
+ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupStaticMemberGroupLeave(
StreamsGroup group,
StreamsGroupMember member
) {
@@ -4133,9 +4024,11 @@ public class GroupMetadataManager {
return new CoordinatorResult<>(
Collections.singletonList(newStreamsGroupCurrentAssignmentRecord(group.groupId(), leavingStaticMember)),
+ new StreamsGroupHeartbeatResult(
new StreamsGroupHeartbeatResponseData()
.setMemberId(member.memberId())
.setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ )
);
}
@@ -4527,50 +4420,6 @@ public class GroupMetadataManager {
() -> streamsGroupFenceMemberOperation(groupId, memberId, "the member session expired.")
);
}
-
- /**
- * Schedules (or reschedules) the topology initialisation timeout for the member.
- *
- * @param groupId The group id.
- * @param topologyId The topology id.
- * @param memberId The member id.
- * @param topologyIntializationTimeout The topology initialization timeout.
- */
- private void scheduleStreamsGroupTopologyInitializationTimeout(
- String groupId,
- String topologyId,
- String memberId,
- int topologyIntializationTimeout
- ) {
- timer.schedule(
- streamsTopologyInitializationTimeoutKey(groupId, topologyId),
- topologyIntializationTimeout,
- TimeUnit.MILLISECONDS,
- true,
- () -> streamsGroupFenceMemberOperation(groupId, memberId, "the timeout for the topology intialization expired.")
- );
- }
-
- /**
- * Verifies if the topology initialization timeout is scheduled.
- *
- * @param groupId The group id.
- * @param topologyId The topology id.
- * @return {@code true} if the topology initialization timeout is scheduled, {@code false} otherwise.
- */
- private boolean isTopologyInitializationScheduled(String groupId, String topologyId) {
- return timer.isScheduled(streamsTopologyInitializationTimeoutKey(groupId, topologyId));
- }
-
- /**
- * Cancels the topology initialization timeout.
- *
- * @param groupId The group id.
- * @param topologyId The topology id.
- */
- private void cancelStreamsGroupTopologyInitializationTimeout(String groupId, String topologyId) {
- timer.cancel(streamsTopologyInitializationTimeoutKey(groupId, topologyId));
- }
/**
* Cancels the session timeout of the member.
@@ -4889,38 +4738,16 @@ public class GroupMetadataManager {
}
}
- /**
- * Handles a StreamsGroupInitialize request.
- *
- * @param context The request context.
- * @param request The actual StreamsGroupInitialize request.
- *
- * @return A Result containing the StreamsGroupInitialize response and
- * a list of records to update the state machine.
- */
- public CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> streamsGroupInitialize(
- RequestContext context,
- StreamsGroupInitializeRequestData request
- ) throws ApiException {
- throwIfStreamsGroupInitializeRequestIsInvalid(request);
-
- return streamsGroupInitialize(
- request.groupId(),
- request.topologyId(),
- request.topology()
- );
- }
-
/**
* Handles a StreamsGroupHeartbeat request.
*
* @param context The request context.
* @param request The actual StreamsGroupHeartbeat request.
*
- * @return A Result containing the StreamsGroupHeartbeat response and
+ * @return A Result containing the StreamsGroupHeartbeat response, a list of internalt topics to create and
* a list of records to update the state machine.
*/
- public CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupHeartbeat(
+ public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
RequestContext context,
StreamsGroupHeartbeatRequestData request
) throws ApiException {
@@ -4943,6 +4770,7 @@ public class GroupMetadataManager {
request.rackId(),
request.rebalanceTimeoutMs(),
request.topologyId(),
+ request.topology(),
context.clientId(),
context.clientAddress.toString(),
request.activeTasks(),
@@ -4971,18 +4799,16 @@ public class GroupMetadataManager {
) {
String groupId = key.groupId();
StreamsGroup streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, value != null);
- Set<String> oldSubscribedTopicNames = new HashSet<>(streamsGroup.subscriptionMetadata().keySet());
+ Set<String> oldSubscribedTopicNames;
+ if (streamsGroup.topology() != null) {
+ oldSubscribedTopicNames = streamsGroup.topology().requiredTopics();
+ } else {
+ oldSubscribedTopicNames = Collections.emptySet();
+ }
if (value != null) {
StreamsTopology topology = StreamsTopology.fromRecord(value);
streamsGroup.setTopology(topology);
- Set<String> newSubscribedTopicNames = new HashSet<>();
- topology.subtopologies().forEach(
- (subtopologyId, subtopology) -> {
- newSubscribedTopicNames.addAll(subtopology.sourceTopics());
- newSubscribedTopicNames.addAll(subtopology.repartitionSourceTopics().stream()
- .map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet()));
- }
- );
+ Set<String> newSubscribedTopicNames = topology.requiredTopics();
updateGroupsByTopics(groupId, oldSubscribedTopicNames, newSubscribedTopicNames);
} else {
updateGroupsByTopics(groupId, oldSubscribedTopicNames, Collections.emptySet());
@@ -5422,13 +5248,13 @@ public class GroupMetadataManager {
StreamsGroup streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, false);
if (value != null) {
- Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata = new HashMap<>();
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> partitionMetadata = new HashMap<>();
value.topics().forEach(topicMetadata -> {
- subscriptionMetadata.put(topicMetadata.topicName(), org.apache.kafka.coordinator.group.streams.TopicMetadata.fromRecord(topicMetadata));
+ partitionMetadata.put(topicMetadata.topicName(), org.apache.kafka.coordinator.group.streams.TopicMetadata.fromRecord(topicMetadata));
});
- streamsGroup.setSubscriptionMetadata(subscriptionMetadata);
+ streamsGroup.setPartitionMetadata(partitionMetadata);
} else {
- streamsGroup.setSubscriptionMetadata(Collections.emptyMap());
+ streamsGroup.setPartitionMetadata(Collections.emptyMap());
}
}
@@ -5523,7 +5349,6 @@ public class GroupMetadataManager {
String memberId = key.memberId();
StreamsGroup streamsGroup = getOrMaybeCreatePersistedStreamsGroup(groupId, value != null);
- Set<String> oldSubscribedTopicNames = new HashSet<>(streamsGroup.subscriptionMetadata().keySet());
if (value != null) {
StreamsGroupMember oldMember = streamsGroup.getOrMaybeCreateMember(memberId, true);
@@ -5542,10 +5367,6 @@ public class GroupMetadataManager {
}
streamsGroup.removeMember(memberId);
}
-
- // ToDo: this is probably a no-op for a streams group since the subscribed topics do not change
- // between group members.
- updateGroupsByTopics(groupId, oldSubscribedTopicNames, streamsGroup.subscriptionMetadata().keySet());
}
/**
@@ -5780,6 +5601,9 @@ public class GroupMetadataManager {
if (group != null && (group.type() == CONSUMER || group.type() == SHARE)) {
((ModernGroup<?>) group).requestMetadataRefresh();
}
+ if (group != null && (group.type() == STREAMS)) {
+ ((StreamsGroup) group).requestMetadataRefresh();
+ }
});
}
@@ -5940,10 +5764,6 @@ public class GroupMetadataManager {
return "rebalance-timeout-" + groupId + "-" + memberId;
}
- public static String streamsTopologyInitializationTimeoutKey(String groupId, String topologyId) {
- return "topology-initialization-timeout-" + groupId + "-" + topologyId;
- }
-
/**
* Replays GroupMetadataKey/Value to update the soft state of
* the classic group.
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
index a2de5ea0d2c..5a28d83a98b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorMetricsShard.java
@@ -508,8 +508,8 @@ public class GroupCoordinatorMetricsShard implements CoordinatorMetricsShard {
case EMPTY:
incrementNumStreamsGroups(StreamsGroupState.EMPTY);
break;
- case INITIALIZING:
- incrementNumStreamsGroups(StreamsGroupState.INITIALIZING);
+ case NOT_READY:
+ incrementNumStreamsGroups(StreamsGroupState.NOT_READY);
break;
case ASSIGNING:
incrementNumStreamsGroups(StreamsGroupState.ASSIGNING);
@@ -533,8 +533,8 @@ public class GroupCoordinatorMetricsShard implements CoordinatorMetricsShard {
case EMPTY:
decrementNumStreamsGroups(StreamsGroupState.EMPTY);
break;
- case INITIALIZING:
- decrementNumStreamsGroups(StreamsGroupState.INITIALIZING);
+ case NOT_READY:
+ decrementNumStreamsGroups(StreamsGroupState.NOT_READY);
break;
case ASSIGNING:
decrementNumStreamsGroups(StreamsGroupState.ASSIGNING);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
index 532955f87aa..6e834eb2ba0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpers.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.coordinator.group.streams;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
@@ -105,16 +105,16 @@ public class CoordinatorStreamsRecordHelpers {
/**
* Creates a StreamsGroupPartitionMetadata record.
*
- * @param groupId The streams group id.
- * @param newSubscriptionMetadata The subscription metadata.
+ * @param groupId The streams group id.
+ * @param newPartitionMetadata The partition metadata.
* @return The record.
*/
public static CoordinatorRecord newStreamsGroupPartitionMetadataRecord(
String groupId,
- Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> newSubscriptionMetadata
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> newPartitionMetadata
) {
StreamsGroupPartitionMetadataValue value = new StreamsGroupPartitionMetadataValue();
- newSubscriptionMetadata.forEach((topicName, topicMetadata) -> {
+ newPartitionMetadata.forEach((topicName, topicMetadata) -> {
List<StreamsGroupPartitionMetadataValue.PartitionMetadata> partitionMetadata = new ArrayList<>();
// If the partition rack information map is empty, store an empty list in the record.
if (!topicMetadata.partitionRacks().isEmpty()) {
@@ -380,7 +380,7 @@ public class CoordinatorStreamsRecordHelpers {
* @return The record.
*/
public static CoordinatorRecord newStreamsGroupTopologyRecord(String groupId,
- List<StreamsGroupInitializeRequestData.Subtopology> subtopologies) {
+ List<StreamsGroupHeartbeatRequestData.Subtopology> subtopologies) {
return newStreamsGroupTopologyRecord(groupId, convertToStreamsGroupTopologyRecord(subtopologies));
}
@@ -401,12 +401,12 @@ public class CoordinatorStreamsRecordHelpers {
}
/**
- * Encodes subtopologies from the initialize RPC to a StreamsTopology record value.
+ * Encodes subtopologies from the Heartbeat RPC to a StreamsTopology record value.
*
* @param subtopologies The subtopologies in the new topology.
* @return The record value.
*/
- public static StreamsGroupTopologyValue convertToStreamsGroupTopologyRecord(List<StreamsGroupInitializeRequestData.Subtopology> subtopologies) {
+ public static StreamsGroupTopologyValue convertToStreamsGroupTopologyRecord(List<StreamsGroupHeartbeatRequestData.Subtopology> subtopologies) {
StreamsGroupTopologyValue value = new StreamsGroupTopologyValue();
subtopologies.forEach(subtopology -> {
List<StreamsGroupTopologyValue.TopicInfo> repartitionSourceTopics =
@@ -442,7 +442,7 @@ public class CoordinatorStreamsRecordHelpers {
return value;
}
- private static StreamsGroupTopologyValue.TopicInfo convertToTopicInfo(StreamsGroupInitializeRequestData.TopicInfo topicInfo) {
+ private static StreamsGroupTopologyValue.TopicInfo convertToTopicInfo(StreamsGroupHeartbeatRequestData.TopicInfo topicInfo) {
List<StreamsGroupTopologyValue.TopicConfig> topicConfigs = topicInfo.topicConfigs() != null ? topicInfo.topicConfigs().stream()
.map(config -> new StreamsGroupTopologyValue.TopicConfig().setKey(config.key()).setValue(config.value()))
.collect(Collectors.toList()) : null;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopology.java
similarity index 50%
copy from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
copy to group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopology.java
index 408f0bfcbb4..0848576d1e2 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopology.java
@@ -16,53 +16,52 @@
*/
package org.apache.kafka.coordinator.group.streams;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
-import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
-import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
-import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.TopicInfo;
+import org.apache.kafka.coordinator.group.streams.topics.ConfiguredInternalTopic;
+import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
/**
- * Immutable topology metadata.
+ * Immutable configured topology metadata.
*/
-public class StreamsTopology {
+public class StreamsConfiguredTopology {
private final String topologyId;
- private final Map<String, Subtopology> subtopologies;
+ private final Map<String, ConfiguredSubtopology> subtopologies;
- public StreamsTopology(final String topologyId,
- final Map<String, Subtopology> subtopologies) {
+ private final Map<String, CreatableTopic> internalTopicsToBeCreated;
+
+ public StreamsConfiguredTopology(final String topologyId,
+ final Map<String, ConfiguredSubtopology> subtopologies,
+ final Map<String, CreatableTopic> internalTopicsToBeCreated
+ ) {
this.topologyId = topologyId;
this.subtopologies = subtopologies;
+ this.internalTopicsToBeCreated = internalTopicsToBeCreated;
}
public String topologyId() {
return topologyId;
}
- public Map<String, Subtopology> subtopologies() {
+ public Map<String, ConfiguredSubtopology> subtopologies() {
return subtopologies;
}
- public Set<String> topicSubscription() {
- return subtopologies.values().stream()
- .flatMap(x -> Stream.concat(x.sourceTopics().stream(), x.repartitionSourceTopics().stream().map(
- TopicInfo::name))).collect(
- Collectors.toSet());
+ public Map<String, CreatableTopic> internalTopicsToBeCreated() {
+ return internalTopicsToBeCreated;
}
- public static StreamsTopology fromRecord(StreamsGroupTopologyValue record) {
- return new StreamsTopology(
- record.topologyId(),
- record.topology().stream().collect(Collectors.toMap(Subtopology::subtopologyId, x -> x))
- );
+ public boolean isReady() {
+ // TODO: Check for internal topics
+ return internalTopicsToBeCreated.isEmpty();
}
@Override
@@ -73,52 +72,53 @@ public class StreamsTopology {
if (o == null || getClass() != o.getClass()) {
return false;
}
- final StreamsTopology that = (StreamsTopology) o;
- return Objects.deepEquals(topologyId, that.topologyId) && Objects.equals(subtopologies, that.subtopologies);
+ final StreamsConfiguredTopology that = (StreamsConfiguredTopology) o;
+ return Objects.equals(topologyId, that.topologyId) && Objects.equals(subtopologies, that.subtopologies)
+ && Objects.equals(internalTopicsToBeCreated, that.internalTopicsToBeCreated);
}
@Override
public int hashCode() {
- return Objects.hash(topologyId, subtopologies);
+ return Objects.hash(topologyId, subtopologies, internalTopicsToBeCreated);
}
@Override
public String toString() {
- return "StreamsTopology{" +
- "topologyId=" + topologyId +
+ return "StreamsConfiguredTopology{" +
+ "topologyId='" + topologyId + '\'' +
", subtopologies=" + subtopologies +
+ ", internalTopicsToBeCreated=" + internalTopicsToBeCreated +
'}';
}
public List<StreamsGroupDescribeResponseData.Subtopology> asStreamsGroupDescribeTopology() {
- return subtopologies.values().stream().map(
+ return subtopologies.entrySet().stream().map(
subtopology -> new StreamsGroupDescribeResponseData.Subtopology()
- .setSourceTopicRegex(subtopology.sourceTopicRegex())
- .setSubtopologyId(subtopology.subtopologyId())
- .setSourceTopics(subtopology.sourceTopics())
- .setRepartitionSinkTopics(subtopology.repartitionSinkTopics())
+ .setSubtopologyId(subtopology.getKey())
+ .setSourceTopics(subtopology.getValue().sourceTopics().stream().sorted().collect(Collectors.toList()))
+ .setRepartitionSinkTopics(subtopology.getValue().repartitionSinkTopics().stream().sorted().collect(Collectors.toList()))
.setRepartitionSourceTopics(
- asStreamsGroupDescribeTopicInfo(subtopology.repartitionSourceTopics()))
+ asStreamsGroupDescribeTopicInfo(subtopology.getValue().repartitionSourceTopics().values()))
.setStateChangelogTopics(
- asStreamsGroupDescribeTopicInfo(subtopology.stateChangelogTopics()))
+ asStreamsGroupDescribeTopicInfo(subtopology.getValue().stateChangelogTopics().values()))
).collect(Collectors.toList());
}
private static List<StreamsGroupDescribeResponseData.TopicInfo> asStreamsGroupDescribeTopicInfo(
- final List<TopicInfo> topicInfos) {
+ final Collection<ConfiguredInternalTopic> topicInfos) {
return topicInfos.stream().map(x ->
new StreamsGroupDescribeResponseData.TopicInfo()
.setName(x.name())
- .setPartitions(x.partitions())
- .setReplicationFactor(x.replicationFactor())
+ .setPartitions(x.numberOfPartitions().orElse(0))
+ .setReplicationFactor(x.replicationFactor().orElse((short) 0))
.setTopicConfigs(
x.topicConfigs() != null ?
- x.topicConfigs().stream().map(
+ x.topicConfigs().entrySet().stream().map(
y -> new StreamsGroupDescribeResponseData.KeyValue()
- .setKey(y.key())
- .setValue(y.value())
+ .setKey(y.getKey())
+ .setValue(y.getValue())
).collect(Collectors.toList()) : null
)
- ).collect(Collectors.toList());
+ ).sorted().collect(Collectors.toList());
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
index f9891e62953..fac219c5ed4 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java
@@ -22,14 +22,18 @@ import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
import org.apache.kafka.common.message.ListGroupsResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.group.Group;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
+import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
+import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager;
import org.apache.kafka.image.ClusterImage;
import org.apache.kafka.image.TopicImage;
import org.apache.kafka.image.TopicsImage;
@@ -37,6 +41,7 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.apache.kafka.timeline.TimelineHashMap;
import org.apache.kafka.timeline.TimelineInteger;
import org.apache.kafka.timeline.TimelineObject;
+import org.slf4j.Logger;
import java.util.Collections;
import java.util.HashMap;
@@ -51,7 +56,7 @@ import java.util.Set;
import static java.util.Collections.emptyMap;
import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.ASSIGNING;
import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.EMPTY;
-import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.INITIALIZING;
+import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.NOT_READY;
import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.RECONCILING;
import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.STABLE;
@@ -62,7 +67,7 @@ public class StreamsGroup implements Group {
public enum StreamsGroupState {
EMPTY("Empty"),
- INITIALIZING("Initializing"),
+ NOT_READY("Not Ready"),
ASSIGNING("Assigning"),
RECONCILING("Reconciling"),
STABLE("Stable"),
@@ -74,7 +79,7 @@ public class StreamsGroup implements Group {
StreamsGroupState(String name) {
this.name = name;
- this.lowerCaseName = name.toLowerCase(Locale.ROOT);
+ this.lowerCaseName = name.toLowerCase(Locale.ROOT).replace(" ", "_");
}
@Override
@@ -100,6 +105,9 @@ public class StreamsGroup implements Group {
}
}
+ private final LogContext logContext;
+ private final Logger log;
+
/**
* The snapshot registry.
*/
@@ -116,8 +124,8 @@ public class StreamsGroup implements Group {
private final TimelineObject<StreamsGroupState> state;
/**
- * The group epoch. The epoch is incremented whenever the subscriptions are updated and it will trigger the computation of a new
- * assignment for the group.
+ * The group epoch. The epoch is incremented whenever the topology, topic metadata or the set of members changes and it will trigger
+ * the computation of a new assignment for the group.
*/
private final TimelineInteger groupEpoch;
@@ -134,7 +142,7 @@ public class StreamsGroup implements Group {
/**
* The metadata associated with each subscribed topic name.
*/
- private final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+ private final TimelineHashMap<String, TopicMetadata> partitionMetadata;
/**
* The target assignment epoch. An assignment epoch smaller than the group epoch means that a new assignment is required. The assignment
@@ -172,6 +180,11 @@ public class StreamsGroup implements Group {
*/
private final TimelineObject<Optional<StreamsTopology>> topology;
+ /**
+ * The configured topology including resolved regular expressions.
+ */
+ private final TimelineObject<Optional<StreamsConfiguredTopology>> configuredTopology;
+
/**
* The metadata refresh deadline. It consists of a timestamp in milliseconds together with the group epoch at the time of setting it.
* The metadata refresh time is considered as a soft state (read that it is not stored in a timeline data structure). It is like this
@@ -182,17 +195,20 @@ public class StreamsGroup implements Group {
private DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
public StreamsGroup(
+ LogContext logContext,
SnapshotRegistry snapshotRegistry,
String groupId,
GroupCoordinatorMetricsShard metrics
) {
+ this.log = logContext.logger(StreamsGroup.class);
+ this.logContext = logContext;
this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
this.groupId = Objects.requireNonNull(groupId);
- this.state = new TimelineObject<>(snapshotRegistry, INITIALIZING);
+ this.state = new TimelineObject<>(snapshotRegistry, EMPTY);
this.groupEpoch = new TimelineInteger(snapshotRegistry);
this.members = new TimelineHashMap<>(snapshotRegistry, 0);
this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
- this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.partitionMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
this.invertedTargetActiveTasksAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
@@ -203,6 +219,7 @@ public class StreamsGroup implements Group {
this.currentWarmupTaskProcessIds = new TimelineHashMap<>(snapshotRegistry, 0);
this.metrics = Objects.requireNonNull(metrics);
this.topology = new TimelineObject<>(snapshotRegistry, Optional.empty());
+ this.configuredTopology = new TimelineObject<>(snapshotRegistry, Optional.empty());
}
/**
@@ -239,12 +256,17 @@ public class StreamsGroup implements Group {
.setGroupType(type().toString());
}
+ public StreamsConfiguredTopology configuredTopology() {
+ return configuredTopology.get().orElse(null);
+ }
+
public StreamsTopology topology() {
return topology.get().orElse(null);
}
public void setTopology(StreamsTopology topology) {
this.topology.set(Optional.of(topology));
+ maybeUpdateConfiguredTopology();
maybeUpdateGroupState();
}
@@ -657,42 +679,44 @@ public class StreamsGroup implements Group {
}
/**
- * @return An immutable Map of subscription metadata for each topic that the consumer group is subscribed to.
+ * @return An immutable Map of partition metadata for each topic that the consumer group is subscribed to.
*/
- public Map<String, TopicMetadata> subscriptionMetadata() {
- return Collections.unmodifiableMap(subscribedTopicMetadata);
+ public Map<String, TopicMetadata> partitionMetadata() {
+ return Collections.unmodifiableMap(partitionMetadata);
}
/**
- * Updates the subscription metadata. This replaces the previous one.
+ * Updates the partition metadata. This replaces the previous one.
*
- * @param subscriptionMetadata The new subscription metadata.
+ * @param partitionMetadata The new partition metadata.
*/
- public void setSubscriptionMetadata(
- Map<String, TopicMetadata> subscriptionMetadata
+ public void setPartitionMetadata(
+ Map<String, TopicMetadata> partitionMetadata
) {
- this.subscribedTopicMetadata.clear();
- this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ this.partitionMetadata.clear();
+ this.partitionMetadata.putAll(partitionMetadata);
+ maybeUpdateConfiguredTopology();
+ maybeUpdateGroupState();
}
/**
- * Computes the subscription metadata based on the current subscription info.
+ * Computes the partition metadata based on the current topology and the current topics/cluster image.
*
* @param topicsImage The current metadata for all available topics.
* @param clusterImage The current metadata for the Kafka cluster.
- * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ * @return An immutable map of partition metadata for each topic that the streams topology is using (besides non-repartition sink topics)
*/
- public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ public Map<String, TopicMetadata> computePartitionMetadata(
TopicsImage topicsImage,
ClusterImage clusterImage,
StreamsTopology topology
) {
- Set<String> subscribedTopicNames = topology.topicSubscription();
+ Set<String> requiredTopicNames = topology.requiredTopics();
// Create the topic metadata for each subscribed topic.
- Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(requiredTopicNames.size());
- subscribedTopicNames.forEach(topicName -> {
+ requiredTopicNames.forEach(topicName -> {
TopicImage topicImage = topicsImage.getTopic(topicName);
if (topicImage != null) {
Map<Integer, Set<String>> partitionRacks = new HashMap<>();
@@ -723,19 +747,19 @@ public class StreamsGroup implements Group {
}
/**
- * Computes the subscription metadata based on the current subscription info.
+ * Computes the partition metadata based on the current topology and the current topics/cluster image.
*
* @param topicsImage The current metadata for all available topics.
* @param clusterImage The current metadata for the Kafka cluster.
- * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ * @return An immutable map of partition metadata for each topic that the streams topology is using (besides non-repartition sink topics)
*/
- public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ public Map<String, TopicMetadata> computePartitionMetadata(
TopicsImage topicsImage,
ClusterImage clusterImage
) {
Optional<StreamsTopology> topology = this.topology.get();
if (topology.isPresent()) {
- return computeSubscriptionMetadata(topicsImage, clusterImage, topology.get());
+ return computePartitionMetadata(topicsImage, clusterImage, topology.get());
}
return Collections.emptyMap();
}
@@ -846,7 +870,7 @@ public class StreamsGroup implements Group {
*/
@Override
public void validateDeleteGroup() throws ApiException {
- if (state() != StreamsGroupState.EMPTY && state() != StreamsGroupState.INITIALIZING) {
+ if (state() != StreamsGroupState.EMPTY) {
throw Errors.NON_EMPTY_GROUP.exception();
}
}
@@ -920,10 +944,10 @@ public class StreamsGroup implements Group {
private void maybeUpdateGroupState() {
StreamsGroupState previousState = state.get();
StreamsGroupState newState = STABLE;
- if (topology() == null) {
- newState = INITIALIZING;
- } else if (members.isEmpty()) {
+ if (members.isEmpty()) {
newState = EMPTY;
+ } else if (topology() == null || configuredTopology() == null || !configuredTopology().isReady()) {
+ newState = NOT_READY;
} else if (groupEpoch.get() > targetAssignmentEpoch.get()) {
newState = ASSIGNING;
} else {
@@ -939,6 +963,35 @@ public class StreamsGroup implements Group {
metrics.onStreamsGroupStateTransition(previousState, newState);
}
+ private void maybeUpdateConfiguredTopology() {
+ if (topology.get().isPresent()) {
+ final StreamsTopology streamsTopology = topology.get().get();
+
+ log.info("[GroupId {}] Attempting to configure the topology {}", groupId, streamsTopology);
+
+ try {
+ final Map<String, ConfiguredSubtopology> configuredTopics =
+ InternalTopicManager.configureTopics(logContext, streamsTopology.subtopologies().values(), partitionMetadata);
+ final Map<String, CreatableTopic> internalTopicsToBeCreated = InternalTopicManager.missingTopics(
+ configuredTopics, partitionMetadata);
+ if (internalTopicsToBeCreated.isEmpty()) {
+ log.info("[GroupId {}] Valid topic configuration found, topology {} is now initialized.", groupId, streamsTopology.topologyId());
+ } else {
+ log.info("[GroupId {}] Internal topics for topology {} need to be created: {}.", groupId, streamsTopology.topologyId(), internalTopicsToBeCreated.keySet());
+ }
+
+ this.configuredTopology.set(Optional.of(new StreamsConfiguredTopology(streamsTopology.topologyId(), configuredTopics, internalTopicsToBeCreated)));
+ } catch (Exception e) {
+ log.warn("[GroupId {}] Failed to configure internal topics for topology {}.", groupId, streamsTopology.topologyId(), e);
+ this.configuredTopology.set(Optional.empty());
+ // TODO: Set validation error, so it can be propagated to all members.
+ }
+
+ } else {
+ configuredTopology.set(Optional.empty());
+ }
+ }
+
/**
* Updates the tasks process IDs based on the old and the new member.
*
@@ -1145,7 +1198,7 @@ public class StreamsGroup implements Group {
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset))
- .setTopology(topology.get(committedOffset).map(StreamsTopology::asStreamsGroupDescribeTopology).orElse(null));
+ .setTopology(configuredTopology.get(committedOffset).map(StreamsConfiguredTopology::asStreamsGroupDescribeTopology).orElse(null));
members.entrySet(committedOffset).forEach(
entry -> describedGroup.members().add(
entry.getValue().asStreamsGroupDescribeMember(
@@ -1181,7 +1234,7 @@ public class StreamsGroup implements Group {
);
records.add(CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(groupId(), groupEpoch()));
- records.add(CoordinatorStreamsRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId(), subscriptionMetadata()));
+ records.add(CoordinatorStreamsRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId(), partitionMetadata()));
members().forEach((__, streamsGroupMember) ->
records.add(CoordinatorStreamsRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId(), streamsGroupMember))
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupInitializeResult.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
similarity index 76%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupInitializeResult.java
rename to group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
index 7a4845993eb..cd4060e183c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupInitializeResult.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
@@ -17,28 +17,28 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopic;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
-public class StreamsGroupInitializeResult {
+public class StreamsGroupHeartbeatResult {
- private final StreamsGroupInitializeResponseData data;
+ private final StreamsGroupHeartbeatResponseData data;
private final Map<String, CreatableTopic> creatableTopics;
- public StreamsGroupInitializeResult(StreamsGroupInitializeResponseData data, Map<String, CreatableTopic> creatableTopics) {
+ public StreamsGroupHeartbeatResult(StreamsGroupHeartbeatResponseData data, Map<String, CreatableTopic> creatableTopics) {
this.data = data;
this.creatableTopics = creatableTopics;
}
- public StreamsGroupInitializeResult(StreamsGroupInitializeResponseData data) {
+ public StreamsGroupHeartbeatResult(StreamsGroupHeartbeatResponseData data) {
this.data = data;
this.creatableTopics = Collections.emptyMap();
}
- public StreamsGroupInitializeResponseData responseData() {
+ public StreamsGroupHeartbeatResponseData responseData() {
return data;
}
@@ -54,7 +54,7 @@ public class StreamsGroupInitializeResult {
if (o == null || getClass() != o.getClass()) {
return false;
}
- final StreamsGroupInitializeResult that = (StreamsGroupInitializeResult) o;
+ final StreamsGroupHeartbeatResult that = (StreamsGroupHeartbeatResult) o;
return Objects.equals(data, that.data) && Objects.equals(creatableTopics,
that.creatableTopics);
}
@@ -66,7 +66,7 @@ public class StreamsGroupInitializeResult {
@Override
public String toString() {
- return "StreamsGroupInitializeResult{" +
+ return "StreamsGroupHeartbeatResult{" +
"data=" + data +
", creatableTopics=" + creatableTopics +
'}';
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
index 408f0bfcbb4..c40f896713c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
@@ -16,12 +16,10 @@
*/
package org.apache.kafka.coordinator.group.streams;
-import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.TopicInfo;
-import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -51,11 +49,17 @@ public class StreamsTopology {
return subtopologies;
}
- public Set<String> topicSubscription() {
+ public Set<String> requiredTopics() {
return subtopologies.values().stream()
- .flatMap(x -> Stream.concat(x.sourceTopics().stream(), x.repartitionSourceTopics().stream().map(
- TopicInfo::name))).collect(
- Collectors.toSet());
+ .flatMap(x ->
+ Stream.concat(
+ Stream.concat(
+ x.sourceTopics().stream(),
+ x.repartitionSourceTopics().stream().map(TopicInfo::name)
+ ),
+ x.stateChangelogTopics().stream().map(TopicInfo::name)
+ )
+ ).collect(Collectors.toSet());
}
public static StreamsTopology fromRecord(StreamsGroupTopologyValue record) {
@@ -89,36 +93,4 @@ public class StreamsTopology {
", subtopologies=" + subtopologies +
'}';
}
-
- public List<StreamsGroupDescribeResponseData.Subtopology> asStreamsGroupDescribeTopology() {
- return subtopologies.values().stream().map(
- subtopology -> new StreamsGroupDescribeResponseData.Subtopology()
- .setSourceTopicRegex(subtopology.sourceTopicRegex())
- .setSubtopologyId(subtopology.subtopologyId())
- .setSourceTopics(subtopology.sourceTopics())
- .setRepartitionSinkTopics(subtopology.repartitionSinkTopics())
- .setRepartitionSourceTopics(
- asStreamsGroupDescribeTopicInfo(subtopology.repartitionSourceTopics()))
- .setStateChangelogTopics(
- asStreamsGroupDescribeTopicInfo(subtopology.stateChangelogTopics()))
- ).collect(Collectors.toList());
- }
-
- private static List<StreamsGroupDescribeResponseData.TopicInfo> asStreamsGroupDescribeTopicInfo(
- final List<TopicInfo> topicInfos) {
- return topicInfos.stream().map(x ->
- new StreamsGroupDescribeResponseData.TopicInfo()
- .setName(x.name())
- .setPartitions(x.partitions())
- .setReplicationFactor(x.replicationFactor())
- .setTopicConfigs(
- x.topicConfigs() != null ?
- x.topicConfigs().stream().map(
- y -> new StreamsGroupDescribeResponseData.KeyValue()
- .setKey(y.key())
- .setValue(y.value())
- ).collect(Collectors.toList()) : null
- )
- ).collect(Collectors.toList());
- }
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
index fcb2e99ff55..b0a1d6e1874 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
@@ -175,13 +175,13 @@ public class TargetAssignmentBuilder {
/**
* Adds the subscription metadata to use.
*
- * @param subscriptionMetadata The subscription metadata.
+ * @param partitionMetadata The subscription metadata.
* @return This object.
*/
- public TargetAssignmentBuilder withSubscriptionMetadata(
- Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata
+ public TargetAssignmentBuilder withPartitionMetadata(
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> partitionMetadata
) {
- this.subscriptionMetadata = subscriptionMetadata;
+ this.subscriptionMetadata = partitionMetadata;
return this;
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java
index 9485617261f..cc2383fae42 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java
@@ -21,16 +21,13 @@ import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCon
import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicConfigCollection;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
-import org.apache.kafka.image.MetadataImage;
-import org.apache.kafka.image.TopicImage;
-
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
import org.slf4j.Logger;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -41,7 +38,7 @@ import java.util.stream.Stream;
public class InternalTopicManager {
public static Map<String, CreatableTopic> missingTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
- MetadataImage metadataImage) {
+ Map<String, TopicMetadata> topicMetadata) {
final Map<String, CreatableTopic> topicsToCreate = new HashMap<>();
for (ConfiguredSubtopology subtopology : subtopologyMap.values()) {
@@ -50,29 +47,27 @@ public class InternalTopicManager {
subtopology.stateChangelogTopics().values()
.forEach(x -> topicsToCreate.put(x.name(), toCreatableTopic(x)));
}
- // TODO: Validate if existing topics are compatible with the new topics
- for (String topic : metadataImage.topics().topicsByName().keySet()) {
+ // TODO: Validate if existing topics are compatible with the required topics
+ for (String topic : topicMetadata.keySet()) {
topicsToCreate.remove(topic);
}
return topicsToCreate;
}
-
public static Map<String, ConfiguredSubtopology> configureTopics(LogContext logContext,
- List<StreamsGroupTopologyValue.Subtopology> subtopologyList,
- MetadataImage metadataImage) {
-
+ Collection<StreamsGroupTopologyValue.Subtopology> subtopologies,
+ Map<String, TopicMetadata> topicMetadata) {
final Logger log = logContext.logger(InternalTopicManager.class);
final Map<String, ConfiguredSubtopology> configuredSubtopologies =
- subtopologyList.stream()
+ subtopologies.stream()
.collect(Collectors.toMap(
StreamsGroupTopologyValue.Subtopology::subtopologyId,
InternalTopicManager::fromPersistedSubtopology)
);
final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology =
- subtopologyList.stream()
+ subtopologies.stream()
.collect(Collectors.toMap(
StreamsGroupTopologyValue.Subtopology::subtopologyId,
InternalTopicManager::copartitionGroupsFromPersistedSubtopology)
@@ -87,7 +82,7 @@ public class InternalTopicManager {
).collect(Collectors.toMap(ConfiguredInternalTopic::name, x -> x));
final Function<String, Integer> topicPartitionCountProvider =
- topic -> getPartitionCount(metadataImage, topic, configuredInternalTopics);
+ topic -> getPartitionCount(topicMetadata, topic, configuredInternalTopics);
configureRepartitionTopics(logContext, configuredSubtopologies, topicPartitionCountProvider);
enforceCopartitioning(logContext, configuredSubtopologies, copartitionGroupsBySubtopology, topicPartitionCountProvider, log);
@@ -141,18 +136,18 @@ public class InternalTopicManager {
changelogTopics.setup();
}
- private static Integer getPartitionCount(MetadataImage metadataImage,
+ private static Integer getPartitionCount(Map<String, TopicMetadata> topicMetadata,
String topic,
Map<String, ConfiguredInternalTopic> configuredInternalTopics) {
- final TopicImage topicImage = metadataImage.topics().getTopic(topic);
- if (topicImage == null) {
+ final TopicMetadata metadata = topicMetadata.get(topic);
+ if (metadata == null) {
if (configuredInternalTopics.containsKey(topic) && configuredInternalTopics.get(topic).numberOfPartitions().isPresent()) {
return configuredInternalTopics.get(topic).numberOfPartitions().get();
} else {
return null;
}
} else {
- return topicImage.partitions().size();
+ return metadata.numPartitions();
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index 6ce91186adf..0bc81517647 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -57,8 +57,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
@@ -78,7 +76,7 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.server.record.BrokerCompressionType;
import org.apache.kafka.server.util.FutureUtils;
@@ -267,122 +265,6 @@ public class GroupCoordinatorServiceTest {
);
}
- @Test
- public void testStreamsGroupInitializeWhenNotStarted() throws ExecutionException, InterruptedException {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorService(
- new LogContext(),
- createConfig(),
- runtime,
- new GroupCoordinatorMetrics(),
- createConfigManager()
- );
-
- StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
- .setGroupId("foo");
-
- CompletableFuture<StreamsGroupInitializeResult> future = service.streamsGroupInitialize(
- requestContext(ApiKeys.STREAMS_GROUP_INITIALIZE),
- request
- );
-
- assertEquals(
- new StreamsGroupInitializeResult(
- new StreamsGroupInitializeResponseData()
- .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())),
- future.get()
- );
- }
-
- @Test
- public void testStreamsGroupInitialize() throws ExecutionException, InterruptedException, TimeoutException {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorService(
- new LogContext(),
- createConfig(),
- runtime,
- new GroupCoordinatorMetrics(),
- createConfigManager()
- );
-
- StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
- .setGroupId("foo");
-
- service.startup(() -> 1);
-
- when(runtime.scheduleWriteOperation(
- ArgumentMatchers.eq("streams-group-initialize"),
- ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
- ArgumentMatchers.eq(Duration.ofMillis(5000)),
- ArgumentMatchers.any()
- )).thenReturn(CompletableFuture.completedFuture(
- new StreamsGroupInitializeResponseData()
- ));
-
- CompletableFuture<StreamsGroupInitializeResult> future = service.streamsGroupInitialize(
- requestContext(ApiKeys.STREAMS_GROUP_INITIALIZE),
- request
- );
-
- assertEquals(new StreamsGroupInitializeResponseData(), future.get(5, TimeUnit.SECONDS));
- }
-
- private static Stream<Arguments> testStreamsGroupInitializeWithExceptionSource() {
- return Stream.of(
- Arguments.arguments(new UnknownTopicOrPartitionException(), Errors.COORDINATOR_NOT_AVAILABLE.code(), null),
- Arguments.arguments(new NotEnoughReplicasException(), Errors.COORDINATOR_NOT_AVAILABLE.code(), null),
- Arguments.arguments(new org.apache.kafka.common.errors.TimeoutException(), Errors.COORDINATOR_NOT_AVAILABLE.code(), null),
- Arguments.arguments(new NotLeaderOrFollowerException(), Errors.NOT_COORDINATOR.code(), null),
- Arguments.arguments(new KafkaStorageException(), Errors.NOT_COORDINATOR.code(), null),
- Arguments.arguments(new RecordTooLargeException(), Errors.UNKNOWN_SERVER_ERROR.code(), null),
- Arguments.arguments(new RecordBatchTooLargeException(), Errors.UNKNOWN_SERVER_ERROR.code(), null),
- Arguments.arguments(new InvalidFetchSizeException(""), Errors.UNKNOWN_SERVER_ERROR.code(), null),
- Arguments.arguments(new InvalidRequestException("Invalid"), Errors.INVALID_REQUEST.code(), "Invalid"),
- Arguments.arguments(new StreamsInvalidTopologyException("Invalid"), Errors.STREAMS_INVALID_TOPOLOGY.code(), "Invalid")
- );
- }
-
- @ParameterizedTest
- @MethodSource("testStreamsGroupInitializeWithExceptionSource")
- public void testStreamsGroupInitializeWithException(
- Throwable exception,
- short expectedErrorCode,
- String expectedErrorMessage
- ) throws ExecutionException, InterruptedException, TimeoutException {
- CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
- GroupCoordinatorService service = new GroupCoordinatorService(
- new LogContext(),
- createConfig(),
- runtime,
- new GroupCoordinatorMetrics(),
- createConfigManager()
- );
-
- StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData()
- .setGroupId("foo");
-
- service.startup(() -> 1);
-
- when(runtime.scheduleWriteOperation(
- ArgumentMatchers.eq("streams-group-initialize"),
- ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
- ArgumentMatchers.eq(Duration.ofMillis(5000)),
- ArgumentMatchers.any()
- )).thenReturn(FutureUtils.failedFuture(exception));
-
- CompletableFuture<StreamsGroupInitializeResult> future = service.streamsGroupInitialize(
- requestContext(ApiKeys.STREAMS_GROUP_INITIALIZE),
- request
- );
-
- assertEquals(
- new StreamsGroupInitializeResponseData()
- .setErrorCode(expectedErrorCode)
- .setErrorMessage(expectedErrorMessage),
- future.get(5, TimeUnit.SECONDS).responseData()
- );
- }
-
@Test
public void testStreamsGroupHeartbeatWhenNotStarted() throws ExecutionException, InterruptedException {
CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
@@ -397,14 +279,14 @@ public class GroupCoordinatorServiceTest {
StreamsGroupHeartbeatRequestData request = new StreamsGroupHeartbeatRequestData()
.setGroupId("foo");
- CompletableFuture<StreamsGroupHeartbeatResponseData> future = service.streamsGroupHeartbeat(
+ CompletableFuture<StreamsGroupHeartbeatResult> future = service.streamsGroupHeartbeat(
requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
request
);
assertEquals(
- new StreamsGroupHeartbeatResponseData()
- .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()),
+ new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData()
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())),
future.get()
);
}
@@ -431,15 +313,15 @@ public class GroupCoordinatorServiceTest {
ArgumentMatchers.eq(Duration.ofMillis(5000)),
ArgumentMatchers.any()
)).thenReturn(CompletableFuture.completedFuture(
- new StreamsGroupHeartbeatResponseData()
+ new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData())
));
- CompletableFuture<StreamsGroupHeartbeatResponseData> future = service.streamsGroupHeartbeat(
+ CompletableFuture<StreamsGroupHeartbeatResult> future = service.streamsGroupHeartbeat(
requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
request
);
- assertEquals(new StreamsGroupHeartbeatResponseData(), future.get(5, TimeUnit.SECONDS));
+ assertEquals(new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData()), future.get(5, TimeUnit.SECONDS));
}
private static Stream<Arguments> testStreamsGroupHeartbeatWithExceptionSource() {
@@ -452,7 +334,9 @@ public class GroupCoordinatorServiceTest {
Arguments.arguments(new RecordTooLargeException(), Errors.UNKNOWN_SERVER_ERROR.code(), null),
Arguments.arguments(new RecordBatchTooLargeException(), Errors.UNKNOWN_SERVER_ERROR.code(), null),
Arguments.arguments(new InvalidFetchSizeException(""), Errors.UNKNOWN_SERVER_ERROR.code(), null),
- Arguments.arguments(new InvalidRequestException("Invalid"), Errors.INVALID_REQUEST.code(), "Invalid")
+ Arguments.arguments(new InvalidRequestException("Invalid"), Errors.INVALID_REQUEST.code(), "Invalid"),
+ Arguments.arguments(new StreamsInvalidTopologyException("Invalid"), Errors.STREAMS_INVALID_TOPOLOGY.code(), "Invalid")
+ // TODO: Add other exceptions
);
}
@@ -484,19 +368,19 @@ public class GroupCoordinatorServiceTest {
ArgumentMatchers.any()
)).thenReturn(FutureUtils.failedFuture(exception));
- CompletableFuture<StreamsGroupHeartbeatResponseData> future = service.streamsGroupHeartbeat(
+ CompletableFuture<StreamsGroupHeartbeatResult> future = service.streamsGroupHeartbeat(
requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
request
);
assertEquals(
- new StreamsGroupHeartbeatResponseData()
+ new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData()
.setErrorCode(expectedErrorCode)
- .setErrorMessage(expectedErrorMessage),
+ .setErrorMessage(expectedErrorMessage)),
future.get(5, TimeUnit.SECONDS)
);
}
-
+
@Test
public void testPartitionFor() {
CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
@@ -1920,7 +1804,7 @@ public class GroupCoordinatorServiceTest {
future.get()
);
}
-
+
@Test
public void testDeleteOffsets() throws Exception {
CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
index b836131e667..c828c2b1dcd 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
@@ -26,8 +26,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -65,7 +63,7 @@ import org.apache.kafka.coordinator.group.generated.ShareGroupMemberMetadataKey;
import org.apache.kafka.coordinator.group.generated.ShareGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.generated.ShareGroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.ShareGroupMetadataValue;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.common.ApiMessageAndVersion;
@@ -136,38 +134,6 @@ public class GroupCoordinatorShardTest {
assertEquals(result, coordinator.consumerGroupHeartbeat(context, request));
}
- @Test
- public void testStreamsGroupInitialize() {
- GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
- OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
- CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
- CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
- GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
- new LogContext(),
- groupMetadataManager,
- offsetMetadataManager,
- Time.SYSTEM,
- new MockCoordinatorTimer<>(Time.SYSTEM),
- mock(GroupCoordinatorConfig.class),
- coordinatorMetrics,
- metricsShard
- );
-
- RequestContext context = requestContext(ApiKeys.STREAMS_GROUP_INITIALIZE);
- StreamsGroupInitializeRequestData request = new StreamsGroupInitializeRequestData();
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result = new CoordinatorResult<>(
- Collections.emptyList(),
- new StreamsGroupInitializeResult(new StreamsGroupInitializeResponseData())
- );
-
- when(groupMetadataManager.streamsGroupInitialize(
- context,
- request
- )).thenReturn(result);
-
- assertEquals(result, coordinator.streamsGroupInitialize(context, request));
- }
-
@Test
public void testStreamsGroupHeartbeat() {
GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
@@ -187,9 +153,9 @@ public class GroupCoordinatorShardTest {
RequestContext context = requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT);
StreamsGroupHeartbeatRequestData request = new StreamsGroupHeartbeatRequestData();
- CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> result = new CoordinatorResult<>(
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = new CoordinatorResult<>(
Collections.emptyList(),
- new StreamsGroupHeartbeatResponseData()
+ new StreamsGroupHeartbeatResult(new StreamsGroupHeartbeatResponseData())
);
when(groupMetadataManager.streamsGroupHeartbeat(
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index afdbe390021..4cf7f099d26 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -59,9 +59,8 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData.TopicInfo;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
-import org.apache.kafka.common.message.StreamsGroupInitializeResponseData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment;
import org.apache.kafka.common.message.SyncGroupResponseData;
@@ -99,8 +98,9 @@ import org.apache.kafka.coordinator.group.modern.share.ShareGroupBuilder;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupMember;
import org.apache.kafka.coordinator.group.streams.CoordinatorStreamsRecordHelpers;
import org.apache.kafka.coordinator.group.streams.StreamsGroup;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState;
import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil;
import org.apache.kafka.image.MetadataDelta;
@@ -275,9 +275,7 @@ public class GroupMetadataManagerTest {
@Test
public void testStreamsRequestValidation() {
-// TODO MockTaskAssignor assignor = new MockTaskAssignor("range");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
-// .withAssignors(Collections.singletonList(assignor))
.build();
Exception ex;
@@ -355,14 +353,7 @@ public class GroupMetadataManagerTest {
.setRackId("")));
assertEquals("RackId can't be empty.", ex.getMessage());
-// TODO: // ServerAssignor must exist if provided in all requests.
-// ex = assertThrows(UnsupportedAssignorException.class, () -> context.streamsGroupHeartbeat(
-// new StreamsGroupHeartbeatRequestData()
-// .setGroupId("foo")
-// .setMemberId(Uuid.randomUuid().toString())
-// .setMemberEpoch(1)
-// .setServerAssignor("bar")));
-// assertEquals("ServerAssignor bar is not supported. Supported assignors: range.", ex.getMessage());
+ // TODO: Test supplied topology
}
@Test
@@ -421,28 +412,44 @@ public class GroupMetadataManagerTest {
}
@Test
- public void testJoiningNonExistingStreamsGroup() {
+ public void testJoiningNonExistingStreamsGroupNoMissingTopics() {
String groupId = "group-id";
int rebalanceTimeoutMs = 300000;
String topologyId = "topology-id";
String processId = "process-id";
+ String subtopologyId = "subtopology-id";
+ MockTaskAssignor assignor = new MockTaskAssignor("mock");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(Uuid.randomUuid(), "input-topic", 3)
+ .addRacks()
+ .build())
+ .withTaskAssignors(Collections.singletonList(assignor))
.build();
- StreamsGroupHeartbeatRequestData heartbeat = buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> topology = new ArrayList<>();
+ topology.add(new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setRepartitionSinkTopics(Collections.emptyList())
+ .setRepartitionSourceTopics(Collections.emptyList())
+ .setSourceTopics(Collections.singletonList("input-topic"))
+ .setStateChangelogTopics(Collections.emptyList())
+ );
+ StreamsGroupHeartbeatRequestData heartbeat =
+ buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, topology, processId, rebalanceTimeoutMs);
+ prepareStreamsGroupAssignment(assignor, heartbeat.memberId(), "subtopology-id");
- CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> result = context.streamsGroupHeartbeat(heartbeat);
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(heartbeat);
assertNotNull(result.response());
- StreamsGroupHeartbeatResponseData response = result.response();
+ StreamsGroupHeartbeatResponseData response = result.response().responseData();
assertEquals(Errors.NONE.code(), response.errorCode());
assertFalse(response.memberId().isEmpty());
assertEquals(1, response.memberEpoch());
- assertTrue(response.shouldInitializeTopology());
assertTrue(response.activeTasks().isEmpty());
assertTrue(response.standbyTasks().isEmpty());
assertTrue(response.warmupTasks().isEmpty());
List<CoordinatorRecord> coordinatorRecords = result.records();
- assertEquals(5, coordinatorRecords.size());
+ assertEquals(7, coordinatorRecords.size());
assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupEpochRecord(groupId, 1)));
StreamsGroupMember member = new StreamsGroupMember.Builder(response.memberId())
.setClientId("client")
@@ -462,6 +469,12 @@ public class GroupMetadataManagerTest {
Collections.emptyMap()
)
));
+ assertTrue(coordinatorRecords.contains(
+ CoordinatorStreamsRecordHelpers.newStreamsGroupTopologyRecord(
+ groupId,
+ topology
+ )
+ ));
StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
.withTargetAssignment(
1,
@@ -471,40 +484,55 @@ public class GroupMetadataManagerTest {
.withOwnedStandbyTasks(Collections.emptyList())
.withOwnedWarmupTasks(Collections.emptyList())
.build();
- CoordinatorStreamsRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, updatedMember);
- assertEquals(StreamsGroup.StreamsGroupState.INITIALIZING, context.streamsGroupState("group-id"));
- assertTrue(context.timer.isScheduled("topology-initialization-timeout-group-id-topology-id"));
- assertEquals(rebalanceTimeoutMs, context.timer.timeout("topology-initialization-timeout-group-id-topology-id").deadlineMs - context.time.milliseconds());
+ assertTrue(coordinatorRecords.contains(
+ CoordinatorStreamsRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, updatedMember)
+ ));
+ assertEquals(StreamsGroup.StreamsGroupState.STABLE, context.streamsGroupState("group-id"));
+
+ final Map<String, CreatableTopic> creatableTopic = result.response().creatableTopics();
+ assertEquals(Collections.emptyMap(), creatableTopic);
}
@Test
- public void testJoiningExistingInitializingStreamsGroup() {
+ public void testJoiningNonExistingStreamsGroupMissingTopics() {
String groupId = "group-id";
int rebalanceTimeoutMs = 300000;
String topologyId = "topology-id";
String processId = "process-id";
+ String subtopologyId = "subtopology-id";
+ MockTaskAssignor assignor = new MockTaskAssignor("mock");
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(Uuid.randomUuid(), "input-topic", 3)
+ .addRacks()
+ .build())
+ .withTaskAssignors(Collections.singletonList(assignor))
.build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> topology = new ArrayList<>();
+ topology.add(new StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setRepartitionSinkTopics(Collections.emptyList())
+ .setRepartitionSourceTopics(Collections.emptyList())
+ .setSourceTopics(Collections.singletonList("input-topic"))
+ .setStateChangelogTopics(Collections.singletonList(new TopicInfo().setName("changelog-topic")))
+ );
StreamsGroupHeartbeatRequestData heartbeat =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
+ buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, topology, processId, rebalanceTimeoutMs);
+ prepareStreamsGroupAssignment(assignor, heartbeat.memberId(), "subtopology-id");
- CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> result = context.streamsGroupHeartbeat(heartbeat);
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(heartbeat);
assertNotNull(result.response());
- StreamsGroupHeartbeatResponseData response = result.response();
+ StreamsGroupHeartbeatResponseData response = result.response().responseData();
assertEquals(Errors.NONE.code(), response.errorCode());
assertFalse(response.memberId().isEmpty());
- assertEquals(2, response.memberEpoch());
- assertFalse(response.shouldInitializeTopology());
+ assertEquals(1, response.memberEpoch());
assertTrue(response.activeTasks().isEmpty());
assertTrue(response.standbyTasks().isEmpty());
assertTrue(response.warmupTasks().isEmpty());
List<CoordinatorRecord> coordinatorRecords = result.records();
- assertEquals(5, coordinatorRecords.size());
- assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupEpochRecord(groupId, 2)));
+ assertEquals(7, coordinatorRecords.size());
+ assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupEpochRecord(groupId, 1)));
StreamsGroupMember member = new StreamsGroupMember.Builder(response.memberId())
.setClientId("client")
.setClientHost("localhost/127.0.0.1")
@@ -513,7 +541,7 @@ public class GroupMetadataManagerTest {
.setProcessId(processId)
.build();
assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupMemberRecord(groupId, member)));
- assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(groupId, 2)));
+ assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(groupId, 1)));
assertTrue(coordinatorRecords.contains(
CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentRecord(
groupId,
@@ -523,6 +551,12 @@ public class GroupMetadataManagerTest {
Collections.emptyMap()
)
));
+ assertTrue(coordinatorRecords.contains(
+ CoordinatorStreamsRecordHelpers.newStreamsGroupTopologyRecord(
+ groupId,
+ topology
+ )
+ ));
StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
.withTargetAssignment(
1,
@@ -532,248 +566,43 @@ public class GroupMetadataManagerTest {
.withOwnedStandbyTasks(Collections.emptyList())
.withOwnedWarmupTasks(Collections.emptyList())
.build();
- CoordinatorStreamsRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, updatedMember);
- assertEquals(StreamsGroup.StreamsGroupState.INITIALIZING, context.streamsGroupState("group-id"));
- }
-
- @Test
- public void testInitTopologyExistingInitializingStreamsGroup() {
- String groupId = "group-id";
- int rebalanceTimeoutMs = 300000;
- String topologyId = "topology-id";
- String processId = "process-id";
- String inputTopicName = "input-topic";
- String subtopologyId = "subtopology-id";
- Uuid inputTopicId = Uuid.randomUuid();
- MetadataImage metadataImage = new MetadataImageBuilder()
- .addTopic(inputTopicId, inputTopicName, 3)
- .addRacks()
- .build();
- MockTaskAssignor assignor = new MockTaskAssignor("mock");
- GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withMetadataImage(metadataImage)
- .withTaskAssignors(Collections.singletonList(assignor))
- .build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- final CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> heartbeatResult =
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
- final List<StreamsGroupInitializeRequestData.Subtopology> subtopologies = new ArrayList<>();
- subtopologies.add(new StreamsGroupInitializeRequestData.Subtopology()
- .setSubtopologyId(subtopologyId)
- .setRepartitionSinkTopics(Collections.emptyList())
- .setRepartitionSourceTopics(Collections.emptyList())
- .setSourceTopics(Collections.singletonList("input-topic"))
- );
- prepareStreamsGroupAssignment(assignor, heartbeatResult.response().memberId(), subtopologyId);
- StreamsGroupInitializeRequestData initialize = new StreamsGroupInitializeRequestData()
- .setGroupId(groupId)
- .setTopologyId(topologyId)
- .setTopology(subtopologies);
-
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result = context.streamsGroupInitialize(initialize);
-
- assertNotNull(result.response());
- StreamsGroupInitializeResponseData response = result.response().responseData();
- assertEquals(Errors.NONE.code(), response.errorCode());
- List<CoordinatorRecord> coordinatorRecords = result.records();
- assertEquals(5, coordinatorRecords.size());
- assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupEpochRecord(groupId, 2)));
- assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupTopologyRecord(groupId, subtopologies)));
-
- org.apache.kafka.coordinator.group.streams.TopicMetadata topicMetadata = new org.apache.kafka.coordinator.group.streams.TopicMetadata(
- inputTopicId,
- inputTopicName,
- 3,
- mkMap(
- mkEntry(0, new HashSet<>(List.of("rack1", "rack0"))),
- mkEntry(1, new HashSet<>(List.of("rack2", "rack1"))),
- mkEntry(2, new HashSet<>(List.of("rack2", "rack3")))
- )
- );
assertTrue(coordinatorRecords.contains(
- CoordinatorStreamsRecordHelpers.newStreamsGroupPartitionMetadataRecord(groupId, mkMap(mkEntry(inputTopicName, topicMetadata)))
+ CoordinatorStreamsRecordHelpers.newStreamsGroupCurrentAssignmentRecord(groupId, updatedMember)
));
- assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(groupId, 2)));
- assertTrue(coordinatorRecords.contains(
- CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentRecord(
- groupId,
- heartbeatResult.response().memberId(),
- TaskAssignmentTestUtil.mkTasksPerSubtopology(TaskAssignmentTestUtil.mkTasks(subtopologyId, 0, 1, 2)),
- Collections.emptyMap(),
- Collections.emptyMap()
- )
- ));
- assertFalse(context.timer.isScheduled("topology-initialization-timeout-group-id-topology-id"));
- assertEquals(StreamsGroup.StreamsGroupState.RECONCILING, context.streamsGroupState("group-id"));
- }
-
- @Test
- public void testInitTopologyWithWrongGroupId() {
- String groupId = "group-id";
- int rebalanceTimeoutMs = 300000;
- String topologyId = "topology-id";
- String processId = "process-id";
- String inputTopicName = "input-topic";
- String subtopologyId = "subtopology-id";
- Uuid inputTopicId = Uuid.randomUuid();
- MetadataImage metadataImage = new MetadataImageBuilder()
- .addTopic(inputTopicId, inputTopicName, 3)
- .addRacks()
- .build();
- MockTaskAssignor assignor = new MockTaskAssignor("mock");
- GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withMetadataImage(metadataImage)
- .withTaskAssignors(Collections.singletonList(assignor))
- .build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- final CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> heartbeatResult =
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
- final List<StreamsGroupInitializeRequestData.Subtopology> subtopologies = new ArrayList<>();
- subtopologies.add(new StreamsGroupInitializeRequestData.Subtopology()
- .setSubtopologyId(subtopologyId)
- .setRepartitionSinkTopics(Collections.emptyList())
- .setRepartitionSourceTopics(Collections.emptyList())
- .setSourceTopics(Collections.singletonList("input-topic"))
- );
- prepareStreamsGroupAssignment(assignor, heartbeatResult.response().memberId(), subtopologyId);
- final String wrongGroupId = "wrong-" + groupId;
- StreamsGroupInitializeRequestData initialize = new StreamsGroupInitializeRequestData()
- .setGroupId(wrongGroupId)
- .setTopologyId(topologyId)
- .setTopology(subtopologies);
-
- final GroupIdNotFoundException groupIdNotFoundException =
- assertThrows(GroupIdNotFoundException.class, () -> context.streamsGroupInitialize(initialize));
+ assertEquals(StreamsGroupState.NOT_READY, context.streamsGroupState("group-id"));
+ final Map<String, CreatableTopic> creatableTopic = result.response().creatableTopics();
assertEquals(
- "Streams group " + wrongGroupId + " not found.",
- groupIdNotFoundException.getMessage()
+ Map.of("changelog-topic", new CreatableTopic().setName("changelog-topic").setNumPartitions(3).setReplicationFactor((short) -1)),
+ creatableTopic
);
}
@Test
- public void testInitTopologyWithWrongTopologyId() {
+ public void testJoiningExistingNotReadyStreamsGroupMissingTopics() {
String groupId = "group-id";
int rebalanceTimeoutMs = 300000;
String topologyId = "topology-id";
String processId = "process-id";
- String inputTopicName = "input-topic";
- String subtopologyId = "subtopology-id";
- Uuid inputTopicId = Uuid.randomUuid();
- MetadataImage metadataImage = new MetadataImageBuilder()
- .addTopic(inputTopicId, inputTopicName, 3)
- .addRacks()
- .build();
- MockTaskAssignor assignor = new MockTaskAssignor("mock");
- GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withMetadataImage(metadataImage)
- .withTaskAssignors(Collections.singletonList(assignor))
- .build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- final CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> heartbeatResult =
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
- final List<StreamsGroupInitializeRequestData.Subtopology> subtopologies = new ArrayList<>();
- subtopologies.add(new StreamsGroupInitializeRequestData.Subtopology()
- .setSubtopologyId(subtopologyId)
- .setRepartitionSinkTopics(Collections.emptyList())
- .setRepartitionSourceTopics(Collections.emptyList())
- .setSourceTopics(Collections.singletonList("input-topic"))
- );
- prepareStreamsGroupAssignment(assignor, heartbeatResult.response().memberId(), subtopologyId);
- final String wrongTopologyId = "wrong-" + topologyId;
- StreamsGroupInitializeRequestData initialize = new StreamsGroupInitializeRequestData()
- .setGroupId(groupId)
- .setTopologyId(wrongTopologyId)
- .setTopology(subtopologies);
-
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result = context.streamsGroupInitialize(initialize);
-
- assertNotNull(result.response());
- StreamsGroupInitializeResponseData response = result.response().responseData();
- assertEquals(Errors.NONE.code(), response.errorCode());
- assertTrue(result.records().isEmpty());
- }
-
- @Test
- public void testInitTopologyForInitializedStreamsGroup() {
- String groupId = "group-id";
- int rebalanceTimeoutMs = 300000;
- String topologyId = "topology-id";
- String processId = "process-id";
- String inputTopicName = "input-topic";
- String subtopologyId = "subtopology-id";
- Uuid inputTopicId = Uuid.randomUuid();
- MetadataImage metadataImage = new MetadataImageBuilder()
- .addTopic(inputTopicId, inputTopicName, 3)
- .addRacks()
- .build();
MockTaskAssignor assignor = new MockTaskAssignor("mock");
- GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
- .withMetadataImage(metadataImage)
- .withTaskAssignors(Collections.singletonList(assignor))
- .build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- final CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> heartbeatResult =
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
- final List<StreamsGroupInitializeRequestData.Subtopology> subtopologies = new ArrayList<>();
- subtopologies.add(new StreamsGroupInitializeRequestData.Subtopology()
- .setSubtopologyId(subtopologyId)
- .setRepartitionSinkTopics(Collections.emptyList())
- .setRepartitionSourceTopics(Collections.emptyList())
- .setSourceTopics(Collections.singletonList("input-topic"))
- );
- prepareStreamsGroupAssignment(assignor, heartbeatResult.response().memberId(), subtopologyId);
- StreamsGroupInitializeRequestData initialize = new StreamsGroupInitializeRequestData()
- .setGroupId(groupId)
- .setTopologyId(topologyId)
- .setTopology(subtopologies);
-
- context.streamsGroupInitialize(initialize);
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result = context.streamsGroupInitialize(initialize);
-
- assertNotNull(result.response());
- StreamsGroupInitializeResponseData response = result.response().responseData();
- assertEquals(Errors.NONE.code(), response.errorCode());
- assertTrue(result.records().isEmpty());
- }
-
- @Test
- public void testTopologyInitializationMissingInternalTopics() {
- String groupId = "group-id";
- int rebalanceTimeoutMs = 300000;
- String topologyId = "topology-id";
- String processId = "process-id";
- Uuid fooTopicId = Uuid.randomUuid();
- String fooTopicName = "repartition";
- Uuid barTopicId = Uuid.randomUuid();
- String barTopicName = "bar";
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
.withMetadataImage(new MetadataImageBuilder()
- .addTopic(fooTopicId, fooTopicName, 4)
- .addTopic(barTopicId, barTopicName, 4)
+ .addTopic(Uuid.randomUuid(), "bar", 3)
.addRacks()
.build())
+ .withTaskAssignors(Collections.singletonList(assignor))
.build();
- StreamsGroupHeartbeatRequestData heartbeatToCreateAndRequestInitGroup =
- buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, processId, rebalanceTimeoutMs);
- context.streamsGroupHeartbeat(heartbeatToCreateAndRequestInitGroup);
- assertThrows(GroupIdNotFoundException.class, () ->
- context.groupMetadataManager.consumerGroup(groupId));
- final List<StreamsGroupInitializeRequestData.Subtopology> topology = Collections.singletonList(
- new StreamsGroupInitializeRequestData.Subtopology()
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> topology = Collections.singletonList(
+ new StreamsGroupHeartbeatRequestData.Subtopology()
.setSubtopologyId("subtopology-id")
.setSourceTopics(Collections.singletonList("bar"))
.setRepartitionSourceTopics(
Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicInfo()
+ new StreamsGroupHeartbeatRequestData.TopicInfo()
.setName("repartition")
.setPartitions(4)
.setTopicConfigs(Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicConfig()
+ new StreamsGroupHeartbeatRequestData.KeyValue()
.setKey("config-name1")
.setValue("config-value1")
))
@@ -781,55 +610,106 @@ public class GroupMetadataManagerTest {
)
.setStateChangelogTopics(
Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicInfo()
+ new StreamsGroupHeartbeatRequestData.TopicInfo()
.setName("changelog")
- .setReplicationFactor((short) 3)
+ .setReplicationFactor((short) 1)
.setTopicConfigs(Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicConfig()
+ new StreamsGroupHeartbeatRequestData.KeyValue()
.setKey("config-name2")
.setValue("config-value2")
))
)
)
);
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result =
- context.streamsGroupInitialize(
- new StreamsGroupInitializeRequestData()
- .setGroupId(groupId)
- .setTopologyId(topologyId)
- .setTopology(topology)
- );
+ StreamsGroupHeartbeatRequestData heartbeatToCreateGroup =
+ buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, topology, processId, rebalanceTimeoutMs);
+ prepareStreamsGroupAssignment(assignor, heartbeatToCreateGroup.memberId(), "subtopology-id");
+ context.streamsGroupHeartbeat(heartbeatToCreateGroup);
+
+ StreamsGroupHeartbeatRequestData heartbeat =
+ buildFirstStreamsGroupHeartbeatRequest(groupId, topologyId, topology, processId, rebalanceTimeoutMs);
+
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = context.streamsGroupHeartbeat(heartbeat);
- CreatableTopicConfigCollection expectedConfig = new CreatableTopicConfigCollection();
- expectedConfig.add(
+ assertNotNull(result.response());
+ Map<String, CreatableTopic> creatableTopics = result.response().creatableTopics();
+ CreatableTopicConfigCollection expectedConfig1 = new CreatableTopicConfigCollection();
+ expectedConfig1.add(
+ new CreatableTopicConfig()
+ .setName("config-name1")
+ .setValue("config-value1")
+ );
+ CreatableTopicConfigCollection expectedConfig2 = new CreatableTopicConfigCollection();
+ expectedConfig2.add(
new CreatableTopicConfig()
.setName("config-name2")
.setValue("config-value2")
);
- CreatableTopic expected =
+ CreatableTopic expected1 =
new CreatableTopic()
- .setName("changelog")
+ .setName("repartition")
.setNumPartitions(4)
- .setReplicationFactor((short) 3)
- .setConfigs(expectedConfig);
+ .setReplicationFactor((short) -1) // default
+ .setConfigs(expectedConfig1);
+ CreatableTopic expected2 =
+ new CreatableTopic()
+ .setName("changelog")
+ .setNumPartitions(3)
+ .setReplicationFactor((short) 1)
+ .setConfigs(expectedConfig2);
assertEquals(
- new StreamsGroupInitializeResult(
- new StreamsGroupInitializeResponseData(),
- Collections.singletonMap(
- "changelog", expected
- )
+ Map.of(
+ "repartition", expected1,
+ "changelog", expected2
),
- result.response()
+ creatableTopics
);
- // TODO: Need to check the generated records. Adapt this unit test after merging of
- // initilization & heartbeat
+ StreamsGroupHeartbeatResponseData response = result.response().responseData();
+ assertEquals(Errors.NONE.code(), response.errorCode());
+ assertFalse(response.memberId().isEmpty());
+ assertEquals(2, response.memberEpoch());
+ assertTrue(response.activeTasks().isEmpty());
+ assertTrue(response.standbyTasks().isEmpty());
+ assertTrue(response.warmupTasks().isEmpty());
+ List<CoordinatorRecord> coordinatorRecords = result.records();
+ assertEquals(5, coordinatorRecords.size());
+ assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupEpochRecord(groupId, 2)));
+ StreamsGroupMember member = new StreamsGroupMember.Builder(response.memberId())
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setRebalanceTimeoutMs(rebalanceTimeoutMs)
+ .setTopologyId(topologyId)
+ .setProcessId(processId)
+ .build();
+ assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupMemberRecord(groupId, member)));
+ assertTrue(coordinatorRecords.contains(CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentEpochRecord(groupId, 2)));
+ assertTrue(coordinatorRecords.contains(
+ CoordinatorStreamsRecordHelpers.newStreamsGroupTargetAssignmentRecord(
+ groupId,
+ member.memberId(),
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ )
+ ));
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(
+ 1,
+ new org.apache.kafka.coordinator.group.streams.Assignment(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap())
+ )
+ .withOwnedActiveTasks(Collections.emptyList())
+ .withOwnedStandbyTasks(Collections.emptyList())
+ .withOwnedWarmupTasks(Collections.emptyList())
+ .build();
+ assertEquals(StreamsGroup.StreamsGroupState.NOT_READY, context.streamsGroupState("group-id"));
}
private StreamsGroupHeartbeatRequestData buildFirstStreamsGroupHeartbeatRequest(
final String groupId,
final String topologyId,
+ final List<StreamsGroupHeartbeatRequestData.Subtopology> topology,
final String processId,
final int rebalanceTimeoutMs) {
@@ -841,6 +721,7 @@ public class GroupMetadataManagerTest {
.setRackId(null)
.setRebalanceTimeoutMs(rebalanceTimeoutMs)
.setTopologyId(topologyId)
+ .setTopology(topology)
.setActiveTasks(Collections.emptyList())
.setStandbyTasks(Collections.emptyList())
.setWarmupTasks(Collections.emptyList())
@@ -9329,7 +9210,7 @@ public class GroupMetadataManagerTest {
new StreamsGroupDescribeResponseData.DescribedGroup()
.setGroupEpoch(epoch)
.setGroupId(streamsGroupIds.get(0))
- .setGroupState(StreamsGroup.StreamsGroupState.INITIALIZING.toString())
+ .setGroupState(StreamsGroupState.EMPTY.toString())
.setAssignmentEpoch(0),
new StreamsGroupDescribeResponseData.DescribedGroup()
.setGroupEpoch(epoch)
@@ -9339,7 +9220,7 @@ public class GroupMetadataManagerTest {
TaskAssignmentTestUtil.mkAssignment(Collections.emptyMap())
)
))
- .setGroupState(StreamsGroup.StreamsGroupState.INITIALIZING.toString())
+ .setGroupState(StreamsGroupState.NOT_READY.toString())
);
List<StreamsGroupDescribeResponseData.DescribedGroup> actual = context.sendStreamsGroupDescribe(streamsGroupIds);
@@ -9409,7 +9290,7 @@ public class GroupMetadataManagerTest {
memberBuilder1.build().asStreamsGroupDescribeMember(TaskAssignmentTestUtil.mkAssignment(Collections.emptyMap())),
memberBuilder2.build().asStreamsGroupDescribeMember(TaskAssignmentTestUtil.mkAssignment(assignmentMap, assignmentMap, assignmentMap))
))
- .setGroupState(StreamsGroup.StreamsGroupState.INITIALIZING.toString())
+ .setGroupState(StreamsGroup.StreamsGroupState.NOT_READY.toString())
.setGroupEpoch(epoch + 2);
assertEquals(1, actual.size());
assertEquals(describedGroup, actual.get(0));
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
index 08990d9c077..08ef0327abe 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
@@ -38,8 +38,6 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseData;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
-import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
import org.apache.kafka.common.message.SyncGroupRequestData;
import org.apache.kafka.common.message.SyncGroupResponseData;
import org.apache.kafka.common.network.ClientInformation;
@@ -110,7 +108,7 @@ import org.apache.kafka.coordinator.group.modern.share.ShareGroup;
import org.apache.kafka.coordinator.group.modern.share.ShareGroupBuilder;
import org.apache.kafka.coordinator.group.streams.StreamsGroup;
import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupInitializeResult;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.taskassignor.TaskAssignor;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.server.common.ApiMessageAndVersion;
@@ -735,38 +733,7 @@ public class GroupMetadataManagerTestContext {
return result;
}
-
- public CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> streamsGroupInitialize(
- StreamsGroupInitializeRequestData request
- ) {
- RequestContext context = new RequestContext(
- new RequestHeader(
- ApiKeys.STREAMS_GROUP_INITIALIZE,
- ApiKeys.STREAMS_GROUP_INITIALIZE.latestVersion(),
- "client",
- 0
- ),
- "1",
- InetAddress.getLoopbackAddress(),
- KafkaPrincipal.ANONYMOUS,
- ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
- SecurityProtocol.PLAINTEXT,
- ClientInformation.EMPTY,
- false
- );
-
- CoordinatorResult<StreamsGroupInitializeResult, CoordinatorRecord> result = groupMetadataManager.streamsGroupInitialize(
- context,
- request
- );
-
- if (result.replayRecords()) {
- result.records().forEach(this::replay);
- }
- return result;
- }
-
- public CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> streamsGroupHeartbeat(
+ public CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> streamsGroupHeartbeat(
StreamsGroupHeartbeatRequestData request
) {
RequestContext context = new RequestContext(
@@ -785,7 +752,7 @@ public class GroupMetadataManagerTestContext {
false
);
- CoordinatorResult<StreamsGroupHeartbeatResponseData, CoordinatorRecord> result = groupMetadataManager.streamsGroupHeartbeat(
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> result = groupMetadataManager.streamsGroupHeartbeat(
context,
request
);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
index 65ad8d43fa8..029c3b8025f 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CoordinatorStreamsRecordHelpersTest.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.coordinator.group.streams;
-import org.apache.kafka.common.message.StreamsGroupInitializeRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
@@ -34,20 +34,20 @@ class CoordinatorStreamsRecordHelpersTest {
@Test
public void testNewStreamsGroupTopologyRecord() {
- List<StreamsGroupInitializeRequestData.Subtopology> topology =
- Collections.singletonList(new StreamsGroupInitializeRequestData.Subtopology()
+ List<StreamsGroupHeartbeatRequestData.Subtopology> topology =
+ Collections.singletonList(new StreamsGroupHeartbeatRequestData.Subtopology()
.setSubtopologyId("subtopology-id")
.setRepartitionSinkTopics(Collections.singletonList("foo"))
.setSourceTopics(Collections.singletonList("bar"))
.setSourceTopicRegex(Collections.singletonList("regex"))
.setRepartitionSourceTopics(
Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicInfo()
+ new StreamsGroupHeartbeatRequestData.TopicInfo()
.setName("repartition")
.setPartitions(4)
.setReplicationFactor((short) 3)
.setTopicConfigs(Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicConfig()
+ new StreamsGroupHeartbeatRequestData.KeyValue()
.setKey("config-name1")
.setValue("config-value1")
))
@@ -55,21 +55,21 @@ class CoordinatorStreamsRecordHelpersTest {
)
.setStateChangelogTopics(
Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicInfo()
+ new StreamsGroupHeartbeatRequestData.TopicInfo()
.setName("changelog")
.setReplicationFactor((short) 2)
.setTopicConfigs(Collections.singletonList(
- new StreamsGroupInitializeRequestData.TopicConfig()
+ new StreamsGroupHeartbeatRequestData.KeyValue()
.setKey("config-name2")
.setValue("config-value2")
))
)
)
.setCopartitionGroups(Arrays.asList(
- new StreamsGroupInitializeRequestData.CopartitionGroup()
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
.setSourceTopics(Collections.singletonList((short) 0))
.setRepartitionSourceTopics(Collections.singletonList((short) 0)),
- new StreamsGroupInitializeRequestData.CopartitionGroup()
+ new StreamsGroupHeartbeatRequestData.CopartitionGroup()
.setSourceTopicRegex(Collections.singletonList((short) 0))
))
);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopologyTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopologyTest.java
new file mode 100644
index 00000000000..9e91b36f0b2
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsConfiguredTopologyTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+import org.apache.kafka.coordinator.group.streams.topics.ConfiguredInternalTopic;
+import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class StreamsConfiguredTopologyTest {
+
+ @Test
+ public void streamsTopologyIdShouldBeCorrect() {
+ StreamsConfiguredTopology topology = new StreamsConfiguredTopology("topology-id", Collections.emptyMap(), Collections.emptyMap());
+ assertEquals("topology-id", topology.topologyId());
+ }
+
+ @Test
+ public void subtopologiesShouldBeCorrect() {
+ Map<String, ConfiguredSubtopology> subtopologies = mkMap(
+ mkEntry("subtopology-1", new ConfiguredSubtopology()),
+ mkEntry("subtopology-2", new ConfiguredSubtopology())
+ );
+ StreamsConfiguredTopology topology = new StreamsConfiguredTopology("topology-id", subtopologies, Collections.emptyMap());
+ assertEquals(subtopologies, topology.subtopologies());
+ }
+
+ @Test
+ public void asStreamsGroupDescribeTopologyShouldReturnCorrectSubtopologies() {
+ Map<String, ConfiguredSubtopology> subtopologies = mkMap(
+ mkEntry("subtopology-1", new ConfiguredSubtopology()
+ .setSourceTopics(Collections.singleton("source-topic-1"))
+ .setRepartitionSinkTopics(Collections.singleton("sink-topic-1"))
+ .setRepartitionSourceTopics(
+ Collections.singletonMap("repartition-topic-1", new ConfiguredInternalTopic("repartition-topic-1", Collections.emptyMap(), Optional.of(2), Optional.of((short) 3))))
+ .setStateChangelogTopics(
+ Collections.singletonMap("changelog-topic-1", new ConfiguredInternalTopic("changelog-topic-1", Collections.emptyMap(), Optional.of(1), Optional.of((short) 2))))
+ ),
+ mkEntry("subtopology-2", new ConfiguredSubtopology()
+ .setSourceTopics(Collections.singleton("source-topic-2"))
+ .setRepartitionSinkTopics(Collections.singleton("sink-topic-2"))
+ .setRepartitionSourceTopics(
+ Collections.singletonMap("repartition-topic-2", new ConfiguredInternalTopic("repartition-topic-2", Collections.emptyMap(), Optional.of(2), Optional.of((short) 3))))
+ .setStateChangelogTopics(
+ Collections.singletonMap("changelog-topic-2", new ConfiguredInternalTopic("changelog-topic-2", Collections.emptyMap(), Optional.of(1), Optional.of((short) 2))))
+ )
+ );
+ StreamsConfiguredTopology topology = new StreamsConfiguredTopology("topology-id", subtopologies, Collections.emptyMap());
+ List<StreamsGroupDescribeResponseData.Subtopology> result = topology.asStreamsGroupDescribeTopology();
+ assertEquals(2, result.size());
+ assertEquals("subtopology-1", result.get(0).subtopologyId());
+ assertEquals(Collections.singletonList("source-topic-1"), result.get(0).sourceTopics());
+ assertEquals(Collections.singletonList("sink-topic-1"), result.get(0).repartitionSinkTopics());
+ assertEquals("repartition-topic-1", result.get(0).repartitionSourceTopics().get(0).name());
+ assertEquals((short) 3, result.get(0).repartitionSourceTopics().get(0).replicationFactor());
+ assertEquals(2, result.get(0).repartitionSourceTopics().get(0).partitions());
+ assertEquals("changelog-topic-1", result.get(0).stateChangelogTopics().get(0).name());
+ assertEquals((short) 2, result.get(0).stateChangelogTopics().get(0).replicationFactor());
+ assertEquals(1, result.get(0).stateChangelogTopics().get(0).partitions());
+ assertEquals("subtopology-2", result.get(1).subtopologyId());
+ assertEquals(Collections.singletonList("source-topic-2"), result.get(1).sourceTopics());
+ assertEquals(Collections.singletonList("sink-topic-2"), result.get(1).repartitionSinkTopics());
+ assertEquals("repartition-topic-2", result.get(1).repartitionSourceTopics().get(0).name());
+ assertEquals((short) 3, result.get(1).repartitionSourceTopics().get(0).replicationFactor());
+ assertEquals(2, result.get(1).repartitionSourceTopics().get(0).partitions());
+ assertEquals("changelog-topic-2", result.get(1).stateChangelogTopics().get(0).name());
+ assertEquals((short) 2, result.get(1).stateChangelogTopics().get(0).replicationFactor());
+ assertEquals(1, result.get(1).stateChangelogTopics().get(0).partitions());
+ }
+}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
index fbe9ca1821e..1269587d6c4 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
@@ -28,8 +28,8 @@ import org.apache.kafka.coordinator.group.OffsetAndMetadata;
import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
+import org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState;
import org.apache.kafka.timeline.SnapshotRegistry;
-
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -53,9 +53,12 @@ import static org.mockito.Mockito.mock;
public class StreamsGroupTest {
+ private static final LogContext LOG_CONTEXT = new LogContext();
+
private StreamsGroup createStreamsGroup(String groupId) {
- SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(LOG_CONTEXT);
return new StreamsGroup(
+ LOG_CONTEXT,
snapshotRegistry,
groupId,
mock(GroupCoordinatorMetricsShard.class)
@@ -421,7 +424,7 @@ public class StreamsGroupTest {
@Test
public void testGroupState() {
StreamsGroup streamsGroup = createStreamsGroup("foo");
- assertEquals(StreamsGroup.StreamsGroupState.INITIALIZING, streamsGroup.state());
+ assertEquals(StreamsGroupState.EMPTY, streamsGroup.state());
StreamsGroupMember member1 = new StreamsGroupMember.Builder("member1")
.setState(MemberState.STABLE)
@@ -433,7 +436,7 @@ public class StreamsGroupTest {
streamsGroup.setGroupEpoch(1);
assertEquals(MemberState.STABLE, member1.state());
- assertEquals(StreamsGroup.StreamsGroupState.INITIALIZING, streamsGroup.state());
+ assertEquals(StreamsGroup.StreamsGroupState.NOT_READY, streamsGroup.state());
streamsGroup.setTopology(new StreamsTopology("topology-id", Collections.emptyMap()));
@@ -498,9 +501,9 @@ public class StreamsGroupTest {
@Test
public void testUpdateInvertedAssignment() {
- SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(LOG_CONTEXT);
GroupCoordinatorMetricsShard metricsShard = mock(GroupCoordinatorMetricsShard.class);
- StreamsGroup streamsGroup = new StreamsGroup(snapshotRegistry, "test-group", metricsShard);
+ StreamsGroup streamsGroup = new StreamsGroup(LOG_CONTEXT, snapshotRegistry, "test-group", metricsShard);
String subtopologyId = "foo-sub";
String memberId1 = "member1";
String memberId2 = "member2";
@@ -543,7 +546,7 @@ public class StreamsGroupTest {
emptyMap(),
emptyMap()
);
- streamsGroup.updateTargetAssignment(memberId2, newAssignment);
+ streamsGroup.updateTargetAssignment(memberId2, newAssignment2);
// Verify that partition 1 is assigned to member2
assertEquals(
@@ -664,8 +667,9 @@ public class StreamsGroupTest {
@Test
public void testValidateOffsetFetch() {
- SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(LOG_CONTEXT);
StreamsGroup group = new StreamsGroup(
+ LOG_CONTEXT,
snapshotRegistry,
"group-foo",
mock(GroupCoordinatorMetricsShard.class)
@@ -698,7 +702,7 @@ public class StreamsGroupTest {
public void testValidateDeleteGroup() {
StreamsGroup streamsGroup = createStreamsGroup("foo");
- assertEquals(StreamsGroup.StreamsGroupState.INITIALIZING, streamsGroup.state());
+ assertEquals(StreamsGroupState.EMPTY, streamsGroup.state());
assertDoesNotThrow(streamsGroup::validateDeleteGroup);
StreamsGroupMember member1 = new StreamsGroupMember.Builder("member1")
@@ -732,7 +736,7 @@ public class StreamsGroupTest {
long commitTimestamp = 20000L;
long offsetsRetentionMs = 10000L;
OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(15000L, OptionalInt.empty(), "", commitTimestamp, OptionalLong.empty());
- StreamsGroup group = new StreamsGroup(new SnapshotRegistry(new LogContext()), "group-id", mock(GroupCoordinatorMetricsShard.class));
+ StreamsGroup group = new StreamsGroup(LOG_CONTEXT, new SnapshotRegistry(LOG_CONTEXT), "group-id", mock(GroupCoordinatorMetricsShard.class));
Optional<OffsetExpirationCondition> offsetExpirationCondition = group.offsetExpirationCondition();
assertTrue(offsetExpirationCondition.isPresent());
@@ -743,22 +747,23 @@ public class StreamsGroupTest {
}
@Test
- public void testIsInStatesCaseInsensitive() {
- SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ public void testIsInStatesCaseInsensitiveAndUnderscored() {
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(LOG_CONTEXT);
GroupCoordinatorMetricsShard metricsShard = new GroupCoordinatorMetricsShard(
snapshotRegistry,
emptyMap(),
new TopicPartition("__consumer_offsets", 0)
);
- StreamsGroup group = new StreamsGroup(snapshotRegistry, "group-foo", metricsShard);
+ StreamsGroup group = new StreamsGroup(LOG_CONTEXT, snapshotRegistry, "group-foo", metricsShard);
snapshotRegistry.idempotentCreateSnapshot(0);
- assertTrue(group.isInStates(Collections.singleton("initializing"), 0));
- assertFalse(group.isInStates(Collections.singleton("Initializing"), 0));
+ assertTrue(group.isInStates(Collections.singleton("empty"), 0));
+ assertFalse(group.isInStates(Collections.singleton("Empty"), 0));
- group.setTopology(new StreamsTopology("topology-id", Collections.emptyMap()));
+ group.updateMember(new StreamsGroupMember.Builder("member1")
+ .build());
snapshotRegistry.idempotentCreateSnapshot(1);
- assertTrue(group.isInStates(Collections.singleton("initializing"), 0));
- assertTrue(group.isInStates(Collections.singleton("empty"), 1));
- assertFalse(group.isInStates(Collections.singleton("initializing"), 1));
+ assertTrue(group.isInStates(Collections.singleton("empty"), 0));
+ assertTrue(group.isInStates(Collections.singleton("not_ready"), 1));
+ assertFalse(group.isInStates(Collections.singleton("empty"), 1));
}
}
\ No newline at end of file
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java
index 7aa303603ab..54f8de1a3da 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsTopologyTest.java
@@ -16,17 +16,14 @@
*/
package org.apache.kafka.coordinator.group.streams;
-import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.TopicInfo;
-
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -55,7 +52,7 @@ public class StreamsTopologyTest {
}
@Test
- public void topicSubscriptionShouldBeCorrect() {
+ public void requiredTopicsShouldBeCorrect() {
Map<String, Subtopology> subtopologies = mkMap(
mkEntry("subtopology-1", new Subtopology()
.setSourceTopics(Arrays.asList("source-topic-1", "source-topic-2"))
@@ -63,6 +60,10 @@ public class StreamsTopologyTest {
new TopicInfo().setName("repartition-topic-1"),
new TopicInfo().setName("repartition-topic-2")
))
+ .setStateChangelogTopics(Arrays.asList(
+ new TopicInfo().setName("changelog-topic-1"),
+ new TopicInfo().setName("changelog-topic-2")
+ ))
),
mkEntry("subtopology-2", new Subtopology()
.setSourceTopics(Arrays.asList("source-topic-3", "source-topic-4"))
@@ -75,9 +76,10 @@ public class StreamsTopologyTest {
StreamsTopology topology = new StreamsTopology("topology-id", subtopologies);
Set<String> expectedTopics = new HashSet<>(Arrays.asList(
"source-topic-1", "source-topic-2", "repartition-topic-1", "repartition-topic-2",
- "source-topic-3", "source-topic-4", "repartition-topic-3", "repartition-topic-4"
+ "source-topic-3", "source-topic-4", "repartition-topic-3", "repartition-topic-4",
+ "changelog-topic-1", "changelog-topic-2"
));
- assertEquals(expectedTopics, topology.topicSubscription());
+ assertEquals(expectedTopics, topology.requiredTopics());
}
@Test
@@ -140,65 +142,4 @@ public class StreamsTopologyTest {
String expectedString = "StreamsTopology{topologyId=topology-id, subtopologies=" + subtopologies + "}";
assertEquals(expectedString, topology.toString());
}
-
- @Test
- public void asStreamsGroupDescribeTopologyShouldReturnCorrectSubtopologies() {
- Map<String, Subtopology> subtopologies = mkMap(
- mkEntry("subtopology-1", new Subtopology()
- .setSourceTopicRegex(Collections.singletonList("regex-1"))
- .setSubtopologyId("subtopology-1")
- .setSourceTopics(Collections.singletonList("source-topic-1"))
- .setRepartitionSinkTopics(Collections.singletonList("sink-topic-1"))
- .setRepartitionSourceTopics(
- Collections.singletonList(new TopicInfo()
- .setName("repartition-topic-1")
- .setReplicationFactor((short) 3)
- .setPartitions(2)))
- .setStateChangelogTopics(
- Collections.singletonList(new TopicInfo()
- .setName("changelog-topic-1")
- .setReplicationFactor((short) 2)
- .setPartitions(1)))
- ),
- mkEntry("subtopology-2", new Subtopology()
- .setSourceTopicRegex(Collections.singletonList("regex-2"))
- .setSubtopologyId("subtopology-2")
- .setSourceTopics(Collections.singletonList("source-topic-2"))
- .setRepartitionSinkTopics(Collections.singletonList("sink-topic-2"))
- .setRepartitionSourceTopics(
- Collections.singletonList(new TopicInfo()
- .setName("repartition-topic-2")
- .setReplicationFactor((short) 3)
- .setPartitions(2)))
- .setStateChangelogTopics(
- Collections.singletonList(new TopicInfo()
- .setName("changelog-topic-2")
- .setReplicationFactor((short) 2)
- .setPartitions(1)))
- )
- );
- StreamsTopology topology = new StreamsTopology("topology-id", subtopologies);
- List<StreamsGroupDescribeResponseData.Subtopology> result = topology.asStreamsGroupDescribeTopology();
- assertEquals(2, result.size());
- assertEquals(Collections.singletonList("regex-1"), result.get(0).sourceTopicRegex());
- assertEquals("subtopology-1", result.get(0).subtopologyId());
- assertEquals(Collections.singletonList("source-topic-1"), result.get(0).sourceTopics());
- assertEquals(Collections.singletonList("sink-topic-1"), result.get(0).repartitionSinkTopics());
- assertEquals("repartition-topic-1", result.get(0).repartitionSourceTopics().get(0).name());
- assertEquals((short) 3, result.get(0).repartitionSourceTopics().get(0).replicationFactor());
- assertEquals(2, result.get(0).repartitionSourceTopics().get(0).partitions());
- assertEquals("changelog-topic-1", result.get(0).stateChangelogTopics().get(0).name());
- assertEquals((short) 2, result.get(0).stateChangelogTopics().get(0).replicationFactor());
- assertEquals(1, result.get(0).stateChangelogTopics().get(0).partitions());
- assertEquals(Collections.singletonList("regex-2"), result.get(1).sourceTopicRegex());
- assertEquals("subtopology-2", result.get(1).subtopologyId());
- assertEquals(Collections.singletonList("source-topic-2"), result.get(1).sourceTopics());
- assertEquals(Collections.singletonList("sink-topic-2"), result.get(1).repartitionSinkTopics());
- assertEquals("repartition-topic-2", result.get(1).repartitionSourceTopics().get(0).name());
- assertEquals((short) 3, result.get(1).repartitionSourceTopics().get(0).replicationFactor());
- assertEquals(2, result.get(1).repartitionSourceTopics().get(0).partitions());
- assertEquals("changelog-topic-2", result.get(1).stateChangelogTopics().get(0).name());
- assertEquals((short) 2, result.get(1).stateChangelogTopics().get(0).replicationFactor());
- assertEquals(1, result.get(1).stateChangelogTopics().get(0).partitions());
- }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
index e9324bcf696..773bafd6f1d 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
@@ -226,7 +226,7 @@ public class TargetAssignmentBuilderTest {
.withMembers(members)
.withTopology(topology)
.withStaticMembers(staticMembers)
- .withSubscriptionMetadata(subscriptionMetadata)
+ .withPartitionMetadata(subscriptionMetadata)
.withTargetAssignment(targetAssignment);
// Add the updated members or delete the deleted members.
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManagerTest.java
index e6ef6fee5d3..ecde45d96d5 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManagerTest.java
@@ -23,14 +23,12 @@ import org.apache.kafka.common.message.CreateTopicsRequestData.CreatableTopicCon
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
-import org.apache.kafka.image.MetadataImage;
-import org.apache.kafka.image.TopicImage;
-import org.apache.kafka.image.TopicsImage;
-import org.apache.kafka.server.immutable.ImmutableMap;
+import org.apache.kafka.coordinator.group.streams.TopicMetadata;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -39,34 +37,19 @@ import java.util.Set;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
class InternalTopicManagerTest {
@Test
void testMissingTopics() {
- MetadataImage metadataImage = mock(MetadataImage.class);
- TopicsImage topicsImage = mock(TopicsImage.class);
- when(metadataImage.topics()).thenReturn(topicsImage);
- final TopicImage sourceTopic1 =
- new TopicImage("source_topic1", Uuid.randomUuid(), mkMap(mkEntry(0, null), mkEntry(1, null)));
- final TopicImage sourceTopic2 =
- new TopicImage("source_topic2", Uuid.randomUuid(), mkMap(mkEntry(0, null), mkEntry(1, null)));
- final TopicImage stateChangelogTopic2 = new TopicImage("state_changelog_topic2", Uuid.randomUuid(),
- mkMap(mkEntry(0, null), mkEntry(1, null)));
- when(topicsImage.topicsByName()).thenReturn(
- ImmutableMap.singleton("source_topic1", sourceTopic1)
- .updated("source_topic2", sourceTopic2)
- .updated("state_changelog_topic2", stateChangelogTopic2)
- );
- when(topicsImage.getTopic(eq("source_topic1"))).thenReturn(sourceTopic1);
- when(topicsImage.getTopic(eq("source_topic2"))).thenReturn(sourceTopic2);
- when(topicsImage.getTopic(eq("state_changelog_topic2"))).thenReturn(stateChangelogTopic2);
+ Map<String, TopicMetadata> topicMetadata = new HashMap<>();
+ topicMetadata.put("source_topic1", new TopicMetadata(Uuid.randomUuid(), "source_topic1", 2, Collections.emptyMap()));
+ topicMetadata.put("source_topic2", new TopicMetadata(Uuid.randomUuid(), "source_topic2", 2, Collections.emptyMap()));
+ topicMetadata.put("state_changelog_topic2",
+ new TopicMetadata(Uuid.randomUuid(), "state_changelog_topic2", 2, Collections.emptyMap()));
Map<String, ConfiguredSubtopology> subtopologyMap = makeExpectedConfiguredTopology();
- Map<String, CreatableTopic> missingTopics = InternalTopicManager.missingTopics(subtopologyMap, metadataImage);
+ Map<String, CreatableTopic> missingTopics = InternalTopicManager.missingTopics(subtopologyMap, topicMetadata);
assertEquals(2, missingTopics.size());
assertEquals(
@@ -90,17 +73,13 @@ class InternalTopicManagerTest {
@Test
void testConfigureTopics() {
- MetadataImage metadataImage = mock(MetadataImage.class);
- TopicsImage topicsImage = mock(TopicsImage.class);
- when(metadataImage.topics()).thenReturn(topicsImage);
- when(topicsImage.getTopic(eq("source_topic1"))).thenReturn(
- new TopicImage("source_topic1", Uuid.randomUuid(), mkMap(mkEntry(0, null), mkEntry(1, null))));
- when(topicsImage.getTopic(eq("source_topic2"))).thenReturn(
- new TopicImage("source_topic2", Uuid.randomUuid(), mkMap(mkEntry(0, null), mkEntry(1, null))));
+ Map<String, TopicMetadata> topicMetadata = new HashMap<>();
+ topicMetadata.put("source_topic1", new TopicMetadata(Uuid.randomUuid(), "source_topic1", 2, Collections.emptyMap()));
+ topicMetadata.put("source_topic2", new TopicMetadata(Uuid.randomUuid(), "source_topic2", 2, Collections.emptyMap()));
List<Subtopology> subtopologyList = makeTestTopology();
Map<String, ConfiguredSubtopology> configuredSubtopologies =
- InternalTopicManager.configureTopics(new LogContext(), subtopologyList, metadataImage);
+ InternalTopicManager.configureTopics(new LogContext(), subtopologyList, topicMetadata);
Map<String, ConfiguredSubtopology> expectedConfiguredSubtopologyMap = makeExpectedConfiguredTopology();
assertEquals(expectedConfiguredSubtopologyMap, configuredSubtopologies);
diff --git a/server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java b/server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java
index ac744ef7bac..f7b26601f7c 100644
--- a/server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java
+++ b/server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java
@@ -175,6 +175,10 @@ import org.apache.kafka.common.message.ShareGroupHeartbeatRequestDataJsonConvert
import org.apache.kafka.common.message.ShareGroupHeartbeatResponseDataJsonConverter;
import org.apache.kafka.common.message.StopReplicaRequestDataJsonConverter;
import org.apache.kafka.common.message.StopReplicaResponseDataJsonConverter;
+import org.apache.kafka.common.message.StreamsGroupDescribeRequestDataJsonConverter;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseDataJsonConverter;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestDataJsonConverter;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseDataJsonConverter;
import org.apache.kafka.common.message.SyncGroupRequestDataJsonConverter;
import org.apache.kafka.common.message.SyncGroupResponseDataJsonConverter;
import org.apache.kafka.common.message.TxnOffsetCommitRequestDataJsonConverter;
@@ -356,6 +360,10 @@ import org.apache.kafka.common.requests.ShareGroupHeartbeatRequest;
import org.apache.kafka.common.requests.ShareGroupHeartbeatResponse;
import org.apache.kafka.common.requests.StopReplicaRequest;
import org.apache.kafka.common.requests.StopReplicaResponse;
+import org.apache.kafka.common.requests.StreamsGroupDescribeRequest;
+import org.apache.kafka.common.requests.StreamsGroupDescribeResponse;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatRequest;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse;
import org.apache.kafka.common.requests.SyncGroupRequest;
import org.apache.kafka.common.requests.SyncGroupResponse;
import org.apache.kafka.common.requests.TxnOffsetCommitRequest;
@@ -541,6 +549,10 @@ public class RequestConvertToJson {
return ShareGroupDescribeRequestDataJsonConverter.write(((ShareGroupDescribeRequest) request).data(), request.version());
case SHARE_GROUP_HEARTBEAT:
return ShareGroupHeartbeatRequestDataJsonConverter.write(((ShareGroupHeartbeatRequest) request).data(), request.version());
+ case STREAMS_GROUP_DESCRIBE:
+ return StreamsGroupDescribeRequestDataJsonConverter.write(((StreamsGroupDescribeRequest) request).data(), request.version());
+ case STREAMS_GROUP_HEARTBEAT:
+ return StreamsGroupHeartbeatRequestDataJsonConverter.write(((StreamsGroupHeartbeatRequest) request).data(), request.version());
case STOP_REPLICA:
return StopReplicaRequestDataJsonConverter.write(((StopReplicaRequest) request).data(), request.version());
case SYNC_GROUP:
@@ -725,6 +737,10 @@ public class RequestConvertToJson {
return ShareGroupDescribeResponseDataJsonConverter.write(((ShareGroupDescribeResponse) response).data(), version);
case SHARE_GROUP_HEARTBEAT:
return ShareGroupHeartbeatResponseDataJsonConverter.write(((ShareGroupHeartbeatResponse) response).data(), version);
+ case STREAMS_GROUP_DESCRIBE:
+ return StreamsGroupDescribeResponseDataJsonConverter.write(((StreamsGroupDescribeResponse) response).data(), version);
+ case STREAMS_GROUP_HEARTBEAT:
+ return StreamsGroupHeartbeatResponseDataJsonConverter.write(((StreamsGroupHeartbeatResponse) response).data(), version);
case STOP_REPLICA:
return StopReplicaResponseDataJsonConverter.write(((StopReplicaResponse) response).data(), version);
case SYNC_GROUP:
diff --git a/streams/src/test/resources/log4j.properties b/streams/src/test/resources/log4j.properties
index 2f0f3833eb9..4dcab88183b 100644
--- a/streams/src/test/resources/log4j.properties
+++ b/streams/src/test/resources/log4j.properties
@@ -28,7 +28,6 @@ log4j.logger.org.apache.kafka.clients.consumer=INFO
log4j.logger.org.apache.kafka.clients.producer=INFO
log4j.logger.org.apache.kafka.streams=INFO
log4j.logger.org.apache.kafka.coordinator.group=INFO
-log4j.logger.org.apache.kafka.clients.consumer.internals.StreamsGroupInitializeRequestManager=INFO
log4j.logger.org.apache.kafka.clients.consumer.internals.StreamsGroupHeartbeatRequestManager=INFO
# printing out the configs takes up a huge amount of the allotted characters,