You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/09/25 16:02:46 UTC
(kafka) 15/28: Create assignment-related classes for streams groups
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch kip1071
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit b165df7a7bd00613c143ce1d8f8c871ff577364f
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Wed Jun 19 11:44:31 2024 +0200
Create assignment-related classes for streams groups
Translate from org.apache.kafka.coordinator.group.consumer package
CurrentAssignmentBuilder
TargetAssignmentBuilder
See https://github.com/lucasbru/kafka/pull/23
copy unit tests that can be preserved easily
---
.../group/GroupCoordinatorRecordHelpers.java | 335 +++++++--
.../coordinator/group/streams/Assignment.java | 7 +
.../group/streams/CurrentAssignmentBuilder.java | 374 ++++++++++
.../coordinator/group/streams/MemberState.java | 17 +-
.../coordinator/group/streams/StreamsGroup.java | 301 +++++---
.../group/streams/StreamsGroupMember.java | 63 +-
.../coordinator/group/streams/StreamsTopology.java | 82 ++
.../group/streams/SubscribedTopicMetadata.java | 98 ---
.../group/streams/TargetAssignmentBuilder.java | 390 ++++++++++
.../kafka/coordinator/group/streams/TopicIds.java | 176 -----
.../coordinator/group/streams/TopicMetadata.java | 29 +-
.../group/streams/TopologyMetadata.java | 105 +++
.../StreamsGroupCurrentMemberAssignmentKey.json | 4 +-
.../message/StreamsGroupMemberMetadataKey.json | 4 +-
.../common/message/StreamsGroupMetadataKey.json | 2 +-
.../message/StreamsGroupPartitionMetadataKey.json | 2 +-
.../StreamsGroupTargetAssignmentMemberKey.json | 4 +-
.../StreamsGroupTargetAssignmentMetadataKey.json | 2 +-
.../common/message/StreamsGroupTopologyKey.json | 2 +-
.../coordinator/group/AssignmentTestUtil.java | 19 -
.../coordinator/group/streams/AssignmentTest.java | 22 +-
.../streams/CurrentAssignmentBuilderTest.java | 576 +++++++++++++++
.../group/streams/StreamsGroupMemberTest.java | 117 +--
.../group/streams/StreamsGroupTest.java | 68 +-
.../group/streams/TargetAssignmentBuilderTest.java | 821 +++++++++++++++++++++
.../group/streams/TaskAssignmentTestUtil.java | 124 ++++
26 files changed, 3133 insertions(+), 611 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
index 03185177b43..063ef9366d8 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java
@@ -56,6 +56,8 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataKe
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMetadataValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataKey;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupPartitionMetadataValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMemberValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTargetAssignmentMetadataKey;
@@ -77,8 +79,7 @@ import java.util.Set;
import java.util.stream.Collectors;
/**
- * This class contains helper methods to create records stored in
- * the __consumer_offsets topic.
+ * This class contains helper methods to create records stored in the __consumer_offsets topic.
*/
@SuppressWarnings("ClassDataAbstractionCoupling")
public class GroupCoordinatorRecordHelpers {
@@ -87,8 +88,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupMemberMetadata record.
*
- * @param groupId The consumer group id.
- * @param member The consumer group member.
+ * @param groupId The consumer group id.
+ * @param member The consumer group member.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupMemberSubscriptionRecord(
@@ -120,6 +121,28 @@ public class GroupCoordinatorRecordHelpers {
);
}
+ /**
+ * Creates a ConsumerGroupMemberMetadata tombstone.
+ *
+ * @param groupId The consumer group id.
+ * @param memberId The consumer group member id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newConsumerGroupMemberSubscriptionTombstoneRecord(
+ String groupId,
+ String memberId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ConsumerGroupMemberMetadataKey()
+ .setGroupId(groupId)
+ .setMemberId(memberId),
+ (short) 5
+ ),
+ null // Tombstone.
+ );
+ }
+
public static CoordinatorRecord newStreamsGroupMemberRecord(
String groupId,
StreamsGroupMember member
@@ -141,31 +164,39 @@ public class GroupCoordinatorRecordHelpers {
.setTopologyHash(member.topologyHash())
.setProcessId(member.processId())
.setHostInfo(member.hostInfo())
- .setClientTags(member.clientTags())
+ .setClientTags(member.clientTags().entrySet().stream().map(e ->
+ new StreamsGroupMemberMetadataValue.KeyValue()
+ .setKey(e.getKey())
+ .setValue(e.getValue())
+ ).collect(Collectors.toList()))
.setUserData(member.userData())
- .setAssignmentConfigs(member.assignmentConfigs()),
- (short) 0
+ .setAssignmentConfigs(member.assignmentConfigs().entrySet().stream().map(e ->
+ new StreamsGroupMemberMetadataValue.KeyValue()
+ .setKey(e.getKey())
+ .setValue(e.getValue())
+ ).collect(Collectors.toList())),
+ (short) 0
)
);
}
/**
- * Creates a ConsumerGroupMemberMetadata tombstone.
+ * Creates a StreamsGroupMemberMetadata tombstone.
*
- * @param groupId The consumer group id.
- * @param memberId The consumer group member id.
+ * @param groupId The streams group id.
+ * @param memberId The streams group member id.
* @return The record.
*/
- public static CoordinatorRecord newConsumerGroupMemberSubscriptionTombstoneRecord(
+ public static CoordinatorRecord newStreamsMemberSubscriptionTombstoneRecord(
String groupId,
String memberId
) {
return new CoordinatorRecord(
new ApiMessageAndVersion(
- new ConsumerGroupMemberMetadataKey()
+ new StreamsGroupMemberMetadataKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 5
+ (short) 11
),
null // Tombstone.
);
@@ -174,8 +205,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupPartitionMetadata record.
*
- * @param groupId The consumer group id.
- * @param newSubscriptionMetadata The subscription metadata.
+ * @param groupId The consumer group id.
+ * @param newSubscriptionMetadata The subscription metadata.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupSubscriptionMetadataRecord(
@@ -207,7 +238,7 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupPartitionMetadata tombstone.
*
- * @param groupId The consumer group id.
+ * @param groupId The consumer group id.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupSubscriptionMetadataTombstoneRecord(
@@ -223,6 +254,69 @@ public class GroupCoordinatorRecordHelpers {
);
}
+ /**
+ * Creates a StreamsGroupPartitionMetadata record.
+ *
+ * @param groupId The streams group id.
+ * @param newSubscriptionMetadata The subscription metadata.
+ * @return The record.
+ */
+ public static CoordinatorRecord newStreamsGroupSubscriptionMetadataRecord(
+ String groupId,
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> newSubscriptionMetadata
+ ) {
+ StreamsGroupPartitionMetadataValue value = new StreamsGroupPartitionMetadataValue();
+ newSubscriptionMetadata.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()) {
+ topicMetadata.partitionRacks().forEach((partition, racks) ->
+ partitionMetadata.add(new StreamsGroupPartitionMetadataValue.PartitionMetadata()
+ .setPartition(partition)
+ .setRacks(new ArrayList<>(racks))
+ )
+ );
+ }
+ value.topics().add(new StreamsGroupPartitionMetadataValue.TopicMetadata()
+ .setTopicId(topicMetadata.id())
+ .setTopicName(topicMetadata.name())
+ .setNumPartitions(topicMetadata.numPartitions())
+ .setPartitionMetadata(partitionMetadata)
+ );
+ });
+
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new StreamsGroupPartitionMetadataKey()
+ .setGroupId(groupId),
+ (short) 10
+ ),
+ new ApiMessageAndVersion(
+ value,
+ (short) 0
+ )
+ );
+ }
+
+ /**
+ * Creates a StreamsGroupPartitionMetadata tombstone.
+ *
+ * @param groupId The streams group id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newStreamsGroupSubscriptionMetadataTombstoneRecord(
+ String groupId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new StreamsGroupPartitionMetadataKey()
+ .setGroupId(groupId),
+ (short) 10
+ ),
+ null // Tombstone.
+ );
+ }
+
/**
* Creates a ConsumerGroupMetadata record.
*
@@ -248,6 +342,25 @@ public class GroupCoordinatorRecordHelpers {
);
}
+ /**
+ * Creates a ConsumerGroupMetadata tombstone.
+ *
+ * @param groupId The consumer group id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newGroupEpochTombstoneRecord(
+ String groupId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ConsumerGroupMetadataKey()
+ .setGroupId(groupId),
+ (short) 3
+ ),
+ null // Tombstone.
+ );
+ }
+
public static CoordinatorRecord newStreamsGroupEpochRecord(
String groupId,
int newGroupEpoch
@@ -267,9 +380,28 @@ public class GroupCoordinatorRecordHelpers {
}
/**
- * Creates a ConsumerGroupMetadata tombstone.
+ * Creates a StreamsGroupMetadata tombstone.
*
- * @param groupId The consumer group id.
+ * @param groupId The streams group id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newStreamsGroupEpochTombstoneRecord(
+ String groupId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new StreamsGroupMetadataKey()
+ .setGroupId(groupId),
+ (short) 9
+ ),
+ null // Tombstone.
+ );
+ }
+
+ /**
+ * Creates a StreamsGroupMetadata tombstone.
+ *
+ * @param groupId The streams group id.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupEpochTombstoneRecord(
@@ -288,9 +420,9 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupTargetAssignmentMember record.
*
- * @param groupId The consumer group id.
- * @param memberId The consumer group member id.
- * @param partitions The target partitions of the member.
+ * @param groupId The consumer group id.
+ * @param memberId The consumer group member id.
+ * @param partitions The target partitions of the member.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupTargetAssignmentRecord(
@@ -324,7 +456,29 @@ public class GroupCoordinatorRecordHelpers {
);
}
- public static CoordinatorRecord newTargetAssignmentRecord(
+ /**
+ * Creates a ConsumerGroupTargetAssignmentMember tombstone.
+ *
+ * @param groupId The consumer group id.
+ * @param memberId The consumer group member id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newTargetAssignmentTombstoneRecord(
+ String groupId,
+ String memberId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ConsumerGroupTargetAssignmentMemberKey()
+ .setGroupId(groupId)
+ .setMemberId(memberId),
+ (short) 7
+ ),
+ null // Tombstone.
+ );
+ }
+
+ public static CoordinatorRecord newStreamsTargetAssignmentRecord(
String groupId,
String memberId,
Map<String, Set<Integer>> activeTasks,
@@ -374,22 +528,22 @@ public class GroupCoordinatorRecordHelpers {
}
/**
- * Creates a ConsumerGroupTargetAssignmentMember tombstone.
+ * Creates a StreamsGroupTargetAssignmentMember tombstone.
*
- * @param groupId The consumer group id.
- * @param memberId The consumer group member id.
+ * @param groupId The streams group id.
+ * @param memberId The streams group member id.
* @return The record.
*/
- public static CoordinatorRecord newConsumerGroupTargetAssignmentTombstoneRecord(
+ public static CoordinatorRecord newStreamsTargetAssignmentTombstoneRecord(
String groupId,
String memberId
) {
return new CoordinatorRecord(
new ApiMessageAndVersion(
- new ConsumerGroupTargetAssignmentMemberKey()
+ new StreamsGroupTargetAssignmentMemberKey()
.setGroupId(groupId)
.setMemberId(memberId),
- (short) 7
+ (short) 13
),
null // Tombstone.
);
@@ -398,8 +552,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupTargetAssignmentMetadata record.
*
- * @param groupId The consumer group id.
- * @param assignmentEpoch The consumer group epoch.
+ * @param groupId The consumer group id.
+ * @param assignmentEpoch The consumer group epoch.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupTargetAssignmentEpochRecord(
@@ -420,6 +574,26 @@ public class GroupCoordinatorRecordHelpers {
);
}
+ /**
+ * Creates a ConsumerGroupTargetAssignmentMetadata tombstone.
+ *
+ * @param groupId The consumer group id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newConsumerTargetAssignmentEpochTombstoneRecord(
+ String groupId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ConsumerGroupTargetAssignmentMetadataKey()
+ .setGroupId(groupId),
+ (short) 6
+ ),
+ null // Tombstone.
+ );
+ }
+
+
public static CoordinatorRecord newStreamsTargetAssignmentEpochRecord(
String groupId,
int assignmentEpoch
@@ -439,19 +613,19 @@ public class GroupCoordinatorRecordHelpers {
}
/**
- * Creates a ConsumerGroupTargetAssignmentMetadata tombstone.
+ * Creates a StreamsGroupTargetAssignmentMetadata tombstone.
*
- * @param groupId The consumer group id.
+ * @param groupId The streams group id.
* @return The record.
*/
- public static CoordinatorRecord newConsumerGroupTargetAssignmentEpochTombstoneRecord(
+ public static CoordinatorRecord newStreamsTargetAssignmentEpochTombstoneRecord(
String groupId
) {
return new CoordinatorRecord(
new ApiMessageAndVersion(
- new ConsumerGroupTargetAssignmentMetadataKey()
+ new StreamsGroupTargetAssignmentMetadataKey()
.setGroupId(groupId),
- (short) 6
+ (short) 12
),
null // Tombstone.
);
@@ -460,8 +634,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a ConsumerGroupCurrentMemberAssignment record.
*
- * @param groupId The consumer group id.
- * @param member The consumer group member.
+ * @param groupId The consumer group id.
+ * @param member The consumer group member.
* @return The record.
*/
public static CoordinatorRecord newConsumerGroupCurrentAssignmentRecord(
@@ -487,7 +661,29 @@ public class GroupCoordinatorRecordHelpers {
);
}
- public static CoordinatorRecord newCurrentAssignmentRecord(
+ /**
+ * Creates a ConsumerGroupCurrentMemberAssignment tombstone.
+ *
+ * @param groupId The consumer group id.
+ * @param memberId The consumer group member id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newCurrentAssignmentTombstoneRecord(
+ String groupId,
+ String memberId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new ConsumerGroupCurrentMemberAssignmentKey()
+ .setGroupId(groupId)
+ .setMemberId(memberId),
+ (short) 8
+ ),
+ null // Tombstone
+ );
+ }
+
+ public static CoordinatorRecord newStreamsCurrentAssignmentRecord(
String groupId,
StreamsGroupMember member
) {
@@ -541,6 +737,28 @@ public class GroupCoordinatorRecordHelpers {
);
}
+ /**
+ * Creates a StreamsGroupCurrentMemberAssignment tombstone.
+ *
+ * @param groupId The streams group id.
+ * @param memberId The streams group member id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newStreamsCurrentAssignmentTombstoneRecord(
+ String groupId,
+ String memberId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new StreamsGroupCurrentMemberAssignmentKey()
+ .setGroupId(groupId)
+ .setMemberId(memberId),
+ (short) 14
+ ),
+ null // Tombstone
+ );
+ }
+
/**
* Creates a ConsumerGroupCurrentMemberAssignment tombstone.
*
@@ -566,9 +784,9 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a GroupMetadata record.
*
- * @param group The classic group.
- * @param assignment The classic group assignment.
- * @param metadataVersion The metadata version.
+ * @param group The classic group.
+ * @param assignment The classic group assignment.
+ * @param metadataVersion The metadata version.
* @return The record.
*/
public static CoordinatorRecord newGroupMetadataRecord(
@@ -624,7 +842,7 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a GroupMetadata tombstone.
*
- * @param groupId The group id.
+ * @param groupId The group id.
* @return The record.
*/
public static CoordinatorRecord newGroupMetadataTombstoneRecord(
@@ -643,8 +861,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates an empty GroupMetadata record.
*
- * @param group The classic group.
- * @param metadataVersion The metadata version.
+ * @param group The classic group.
+ * @param metadataVersion The metadata version.
* @return The record.
*/
public static CoordinatorRecord newEmptyGroupMetadataRecord(
@@ -713,9 +931,9 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates an OffsetCommit tombstone record.
*
- * @param groupId The group id.
- * @param topic The topic name.
- * @param partitionId The partition id.
+ * @param groupId The group id.
+ * @param topic The topic name.
+ * @param partitionId The partition id.
* @return The record.
*/
public static CoordinatorRecord newOffsetCommitTombstoneRecord(
@@ -1079,8 +1297,8 @@ public class GroupCoordinatorRecordHelpers {
/**
* Creates a StreamsTopology record.
*
- * @param groupId The consumer group id.
- * @param subtopologies The subtopologies in the new topology.
+ * @param groupId The consumer group id.
+ * @param subtopologies The subtopologies in the new topology.
* @return The record.
*/
public static CoordinatorRecord newStreamsGroupTopologyRecord(String groupId,
@@ -1112,4 +1330,23 @@ public class GroupCoordinatorRecordHelpers {
new ApiMessageAndVersion(value, (short) 0));
}
+ /**
+ * Creates a StreamsGroupTopology tombstone.
+ *
+ * @param groupId The streams group id.
+ * @return The record.
+ */
+ public static CoordinatorRecord newStreamsGroupTopologyRecordTombstone(
+ String groupId
+ ) {
+ return new CoordinatorRecord(
+ new ApiMessageAndVersion(
+ new StreamsGroupTopologyKey()
+ .setGroupId(groupId),
+ (short) 15
+ ),
+ null // Tombstone
+ );
+ }
+
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/Assignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/Assignment.java
index b80e34cbfe5..17d94499d4c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/Assignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/Assignment.java
@@ -40,6 +40,13 @@ public class Assignment {
private final Map<String, Set<Integer>> standbyTasks;
private final Map<String, Set<Integer>> warmupTasks;
+ // TODO: Tests using this constructor probably should be extended to cover standby tasks
+ public Assignment(final Map<String, Set<Integer>> activeTasks) {
+ this.activeTasks = Collections.unmodifiableMap(Objects.requireNonNull(activeTasks));
+ this.standbyTasks = Collections.emptyMap();
+ this.warmupTasks = Collections.emptyMap();
+ }
+
public Assignment(final Map<String, Set<Integer>> activeTasks,
final Map<String, Set<Integer>> standbyTasks,
final Map<String, Set<Integer>> warmupTasks) {
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
new file mode 100644
index 00000000000..30813ba5c13
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java
@@ -0,0 +1,374 @@
+/*
+ * 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.errors.FencedMemberEpochException;
+import org.apache.kafka.common.message.StreamsHeartbeatRequestData;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.BiFunction;
+
+/**
+ * The CurrentAssignmentBuilder class encapsulates the reconciliation engine of the streams group protocol. Given the current state of a
+ * member and a desired or target assignment state, the state machine takes the necessary steps to converge them.
+ */
+public class CurrentAssignmentBuilder {
+
+ /**
+ * The streams group member which is reconciled.
+ */
+ private final StreamsGroupMember member;
+
+ /**
+ * The target assignment epoch.
+ */
+ private int targetAssignmentEpoch;
+
+ /**
+ * The target assignment.
+ */
+ private org.apache.kafka.coordinator.group.streams.Assignment targetAssignment;
+
+ /**
+ * A function which returns the current epoch of a topic-partition or -1 if the topic-partition is not assigned. The current epoch is
+ * the epoch of the current owner.
+ */
+ private BiFunction<String, Integer, Integer> currentPartitionEpoch;
+
+ /**
+ * The active tasks owned by the streams. This is directly provided by the member in the StreamsHeartbeat request.
+ */
+ private List<StreamsHeartbeatRequestData.TaskIds> ownedActiveTasks;
+
+ /**
+ * The standby tasks owned by the streams. This is directly provided by the member in the StreamsHeartbeat request.
+ */
+ private List<StreamsHeartbeatRequestData.TaskIds> ownedStandbyTasks;
+
+ /**
+ * The warmup tasks owned by the streams. This is directly provided by the member in the StreamsHeartbeat request.
+ */
+ private List<StreamsHeartbeatRequestData.TaskIds> ownedWarmupTasks;
+
+ /**
+ * Constructs the CurrentAssignmentBuilder based on the current state of the provided streams group member.
+ *
+ * @param member The streams group member that must be reconciled.
+ */
+ public CurrentAssignmentBuilder(StreamsGroupMember member) {
+ this.member = Objects.requireNonNull(member);
+ }
+
+ /**
+ * Sets the target assignment epoch and the target assignment that the streams group member must be reconciled to.
+ *
+ * @param targetAssignmentEpoch The target assignment epoch.
+ * @param targetAssignment The target assignment.
+ * @return This object.
+ */
+ public CurrentAssignmentBuilder withTargetAssignment(
+ int targetAssignmentEpoch,
+ Assignment targetAssignment
+ ) {
+ this.targetAssignmentEpoch = targetAssignmentEpoch;
+ this.targetAssignment = Objects.requireNonNull(targetAssignment);
+ return this;
+ }
+
+ /**
+ * Sets a BiFunction which allows to retrieve the current epoch of a partition. This is used by the state machine to determine if a
+ * partition is free or still used by another member.
+ *
+ * @param currentPartitionEpoch A BiFunction which gets the epoch of a topic id / tasks id pair.
+ * @return This object.
+ */
+ public CurrentAssignmentBuilder withCurrentPartitionEpoch(
+ BiFunction<String, Integer, Integer> currentPartitionEpoch
+ ) {
+ this.currentPartitionEpoch = Objects.requireNonNull(currentPartitionEpoch);
+ return this;
+ }
+
+ /**
+ * Sets the active tasks currently owned by the member. This comes directly from the last StreamsHeartbeat request. This is used to
+ * determine if the member has revoked the necessary tasks.
+ *
+ * @param ownedActiveTasks A list of topic-tasks.
+ * @return This object.
+ */
+ public CurrentAssignmentBuilder withOwnedActiveTasks(
+ List<StreamsHeartbeatRequestData.TaskIds> ownedActiveTasks
+ ) {
+ this.ownedActiveTasks = ownedActiveTasks;
+ return this;
+ }
+
+ /**
+ * Sets the standby tasks currently owned by the member. This comes directly from the last StreamsHeartbeat request. This is used to
+ * determine if the member has revoked the necessary tasks.
+ *
+ * @param ownedStandbyTasks A list of topic-tasks.
+ * @return This object.
+ */
+ public CurrentAssignmentBuilder withOwnedStandbyTasks(
+ List<StreamsHeartbeatRequestData.TaskIds> ownedStandbyTasks
+ ) {
+ this.ownedStandbyTasks = ownedStandbyTasks;
+ return this;
+ }
+
+ /**
+ * Sets the warmup tasks currently owned by the member. This comes directly from the last StreamsHeartbeat request. This is used to
+ * determine if the member has revoked the necessary tasks.
+ *
+ * @param ownedWarmupTasks A list of topic-tasks.
+ * @return This object.
+ */
+ public CurrentAssignmentBuilder withOwnedWarmupTasks(
+ List<StreamsHeartbeatRequestData.TaskIds> ownedWarmupTasks
+ ) {
+ this.ownedWarmupTasks = ownedWarmupTasks;
+ return this;
+ }
+
+
+ /**
+ * Builds the next state for the member or keep the current one if it is not possible to move forward with the current state.
+ *
+ * @return A new StreamsGroupMember or the current one.
+ */
+ public StreamsGroupMember build() {
+ switch (member.state()) {
+ case STABLE:
+ // When the member is in the STABLE state, we verify if a newer
+ // epoch (or target assignment) is available. If it is, we can
+ // reconcile the member towards it. Otherwise, we return.
+ if (member.memberEpoch() != targetAssignmentEpoch) {
+ return computeNextAssignment(
+ member.memberEpoch(),
+ member.assignedActiveTasks(),
+ member.assignedStandbyTasks(),
+ member.assignedWarmupTasks()
+ );
+ } else {
+ return member;
+ }
+
+ case UNREVOKED_TASKS:
+ // When the member is in the UNREVOKED_TASKS state, we wait
+ // until the member has revoked the necessary tasks. They are
+ // considered revoked when they are not anymore reported in the
+ // owned tasks set in the StreamsHeartbeat API.
+
+ // If the member provides its owned tasks. We verify if it still
+ // owns any of the revoked tasks. If it does, we cannot progress.
+ if (ownsRevokedTasks(member.activeTasksPendingRevocation())) {
+ return member;
+ }
+
+ // When the member has revoked all the pending tasks, it can
+ // transition to the next epoch (current + 1) and we can reconcile
+ // its state towards the latest target assignment.
+ return computeNextAssignment(
+ member.memberEpoch() + 1,
+ member.assignedActiveTasks(),
+ member.assignedStandbyTasks(),
+ member.assignedWarmupTasks()
+ );
+
+ case UNRELEASED_TASKS:
+ // When the member is in the UNRELEASED_TASKS, we reconcile the
+ // member towards the latest target assignment. This will assign any
+ // of the unreleased tasks when they become available.
+ return computeNextAssignment(
+ member.memberEpoch(),
+ member.assignedActiveTasks(),
+ member.assignedStandbyTasks(),
+ member.assignedWarmupTasks()
+ );
+
+ case UNKNOWN:
+ // We could only end up in this state if a new state is added in the
+ // future and the group coordinator is downgraded. In this case, the
+ // best option is to fence the member to force it to rejoin the group
+ // without any tasks and to reconcile it again from scratch.
+ if (ownedActiveTasks == null || !ownedActiveTasks.isEmpty()) {
+ throw new FencedMemberEpochException("The streams group member is in a unknown state. "
+ + "The member must abandon all its tasks and rejoin.");
+ }
+
+ return computeNextAssignment(
+ targetAssignmentEpoch,
+ member.assignedActiveTasks(),
+ member.assignedStandbyTasks(),
+ member.assignedWarmupTasks()
+ );
+ }
+
+ return member;
+ }
+
+ /**
+ * Decides whether the current ownedTopicTasks contains any partition that is pending revocation.
+ *
+ * @param assignment The assignment that has the tasks pending revocation.
+ * @return A boolean based on the condition mentioned above.
+ */
+ private boolean ownsRevokedTasks(
+ Map<String, Set<Integer>> assignment
+ ) {
+ if (ownedActiveTasks == null) {
+ return true;
+ }
+
+ for (StreamsHeartbeatRequestData.TaskIds activeTasks : ownedActiveTasks) {
+ Set<Integer> tasksPendingRevocation =
+ assignment.getOrDefault(activeTasks.subtopology(), Collections.emptySet());
+
+ for (Integer partitionId : activeTasks.partitions()) {
+ if (tasksPendingRevocation.contains(partitionId)) {
+ return true;
+ }
+ }
+ }
+
+ return false;
+ }
+
+ /**
+ * Computes the next assignment.
+ *
+ * @param memberEpoch The epoch of the member to use. This may be different from the epoch in
+ * {@link CurrentAssignmentBuilder#member}.
+ * @param memberAssignedActiveTasks The assigned active tasks of the member to use.
+ * @param memberAssignedStandbyTasks The assigned standby tasks of the member to use.
+ * @param memberAssignedWarmupTasks The assigned warmup tasks of the member to use.
+ * @return A new StreamsGroupMember.
+ */
+ private StreamsGroupMember computeNextAssignment(
+ int memberEpoch,
+ Map<String, Set<Integer>> memberAssignedActiveTasks,
+ Map<String, Set<Integer>> memberAssignedStandbyTasks,
+ Map<String, Set<Integer>> memberAssignedWarmupTasks
+ ) {
+ boolean hasUnreleasedTasks = false;
+ Map<String, Set<Integer>> newAssignedTasks = new HashMap<>();
+ Map<String, Set<Integer>> newTasksPendingRevocation = new HashMap<>();
+ Map<String, Set<Integer>> newTasksPendingAssignment = new HashMap<>();
+
+ Set<String> allSubtopologyIds = new HashSet<>(targetAssignment.activeTasks().keySet());
+ allSubtopologyIds.addAll(memberAssignedActiveTasks.keySet());
+
+ for (String subtopologyId : allSubtopologyIds) {
+ Set<Integer> target = targetAssignment.activeTasks()
+ .getOrDefault(subtopologyId, Collections.emptySet());
+ Set<Integer> currentAssignedTasks = memberAssignedActiveTasks
+ .getOrDefault(subtopologyId, Collections.emptySet());
+
+ // New Assigned Tasks = Previous Assigned Tasks ∩ Target
+ Set<Integer> assignedTasks = new HashSet<>(currentAssignedTasks);
+ assignedTasks.retainAll(target);
+
+ // Tasks Pending Revocation = Previous Assigned Tasks - New Assigned Tasks
+ Set<Integer> tasksPendingRevocation = new HashSet<>(currentAssignedTasks);
+ tasksPendingRevocation.removeAll(assignedTasks);
+
+ // Tasks Pending Assignment = Target - New Assigned Tasks - Unreleased Tasks
+ Set<Integer> tasksPendingAssignment = new HashSet<>(target);
+ tasksPendingAssignment.removeAll(assignedTasks);
+ hasUnreleasedTasks = tasksPendingAssignment.removeIf(partitionId ->
+ currentPartitionEpoch.apply(subtopologyId, partitionId) != -1
+ ) || hasUnreleasedTasks;
+
+ if (!assignedTasks.isEmpty()) {
+ newAssignedTasks.put(subtopologyId, assignedTasks);
+ }
+
+ if (!tasksPendingRevocation.isEmpty()) {
+ newTasksPendingRevocation.put(subtopologyId, tasksPendingRevocation);
+ }
+
+ if (!tasksPendingAssignment.isEmpty()) {
+ newTasksPendingAssignment.put(subtopologyId, tasksPendingAssignment);
+ }
+ }
+
+ if (!newTasksPendingRevocation.isEmpty() && ownsRevokedTasks(newTasksPendingRevocation)) {
+ // If there are tasks to be revoked, the member remains in its current
+ // epoch and requests the revocation of those tasks. It transitions to
+ // the UNREVOKED_TASKS state to wait until the client acknowledges the
+ // revocation of the tasks.
+ return new StreamsGroupMember.Builder(member)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .updateMemberEpoch(memberEpoch)
+ .setAssignedActiveTasks(newAssignedTasks)
+ .setActiveTasksPendingRevocation(newTasksPendingRevocation)
+ .setAssignedStandbyTasks(memberAssignedStandbyTasks)
+ .setAssignedWarmupTasks(memberAssignedWarmupTasks)
+ .build();
+ } else if (!newTasksPendingAssignment.isEmpty()) {
+ // If there are tasks to be assigned, the member transitions to the
+ // target epoch and requests the assignment of those tasks. Note that
+ // the tasks are directly added to the assigned tasks set. The
+ // member transitions to the STABLE state or to the UNRELEASED_TASKS
+ // state depending on whether there are unreleased tasks or not.
+ newTasksPendingAssignment.forEach((subtopologyId, tasks) -> newAssignedTasks
+ .computeIfAbsent(subtopologyId, __ -> new HashSet<>())
+ .addAll(tasks));
+ org.apache.kafka.coordinator.group.streams.MemberState newState =
+ hasUnreleasedTasks ? org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS
+ : org.apache.kafka.coordinator.group.streams.MemberState.STABLE;
+ return new StreamsGroupMember.Builder(member)
+ .setState(newState)
+ .updateMemberEpoch(targetAssignmentEpoch)
+ .setAssignedActiveTasks(newAssignedTasks)
+ .setActiveTasksPendingRevocation(Collections.emptyMap())
+ .setAssignedStandbyTasks(memberAssignedStandbyTasks)
+ .setAssignedWarmupTasks(memberAssignedWarmupTasks)
+ .build();
+ } else if (hasUnreleasedTasks) {
+ // If there are no tasks to be revoked nor to be assigned but some
+ // tasks are not available yet, the member transitions to the target
+ // epoch, to the UNRELEASED_TASKS state and waits.
+ return new StreamsGroupMember.Builder(member)
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .updateMemberEpoch(targetAssignmentEpoch)
+ .setAssignedActiveTasks(newAssignedTasks)
+ .setActiveTasksPendingRevocation(Collections.emptyMap())
+ .setAssignedStandbyTasks(memberAssignedStandbyTasks)
+ .setAssignedWarmupTasks(memberAssignedWarmupTasks)
+ .build();
+ } else {
+ // Otherwise, the member transitions to the target epoch and to the
+ // STABLE state.
+ return new StreamsGroupMember.Builder(member)
+ .setState(MemberState.STABLE)
+ .updateMemberEpoch(targetAssignmentEpoch)
+ .setAssignedActiveTasks(newAssignedTasks)
+ .setActiveTasksPendingRevocation(Collections.emptyMap())
+ .setAssignedStandbyTasks(memberAssignedStandbyTasks)
+ .setAssignedWarmupTasks(memberAssignedWarmupTasks)
+ .build();
+ }
+ }
+}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberState.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberState.java
index 53518e61934..f9ccb8f2ebc 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberState.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberState.java
@@ -20,8 +20,8 @@ import java.util.HashMap;
import java.util.Map;
/**
- * The various states that a member can be in. For their definition,
- * refer to the documentation of {{@link org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder}}.
+ * The various states that a member can be in. For their definition, refer to the documentation of
+ * {{@link org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder}}.
*/
public enum MemberState {
@@ -31,28 +31,25 @@ public enum MemberState {
STABLE((byte) 1),
/**
- * The member must revoke some tasks in order to be able to
- * transition to the next epoch.
+ * The member must revoke some tasks in order to be able to transition to the next epoch.
*/
UNREVOKED_TASKS((byte) 2),
/**
- * The member transitioned to the last epoch but waits on some
- * tasks which have not been revoked by their previous
- * owners yet.
+ * The member transitioned to the last epoch but waits on some tasks which have not been revoked by their previous owners yet.
*/
UNRELEASED_TASKS((byte) 3),
/**
- * The member is in an unknown state. This can only happen if a future
- * version of the software introduces a new state unknown by this version.
+ * The member is in an unknown state. This can only happen if a future version of the software introduces a new state unknown by this
+ * version.
*/
UNKNOWN((byte) 127);
private final static Map<Byte, MemberState> VALUES_TO_ENUMS = new HashMap<>();
static {
- for (MemberState state: MemberState.values()) {
+ for (MemberState state : MemberState.values()) {
VALUES_TO_ENUMS.put(state.value(), state);
}
}
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 d28e9218994..4ba97bc2288 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
@@ -29,7 +29,6 @@ import org.apache.kafka.coordinator.group.CoordinatorRecordHelpers;
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.assignor.SubscriptionType;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetricsShard;
import org.apache.kafka.image.ClusterImage;
import org.apache.kafka.image.TopicImage;
@@ -50,16 +49,13 @@ import java.util.Optional;
import java.util.Set;
import static java.util.Collections.emptyMap;
-import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
-import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
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.RECONCILING;
import static org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState.STABLE;
/**
- * A Streams Group. All the metadata in this class are backed by
- * records in the __consumer_offsets partitions.
+ * A Streams Group. All the metadata in this class are backed by records in the __consumer_offsets partitions.
*/
public class StreamsGroup implements Group {
@@ -90,6 +86,7 @@ public class StreamsGroup implements Group {
}
public static class DeadlineAndEpoch {
+
static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
public final long deadlineMs;
@@ -117,9 +114,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 subscriptions are updated and it will trigger the computation of a new
+ * assignment for the group.
*/
private final TimelineInteger groupEpoch;
@@ -139,9 +135,13 @@ public class StreamsGroup implements Group {
private final TimelineHashMap<String, Integer> assignors;
/**
- * The target assignment epoch. An assignment epoch smaller than the group epoch
- * means that a new assignment is required. The assignment epoch is updated when
- * a new assignment is installed.
+ * The metadata associated with each subscribed topic name.
+ */
+ private final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch means that a new assignment is required. The assignment
+ * epoch is updated when a new assignment is installed.
*/
private final TimelineInteger targetAssignmentEpoch;
@@ -151,17 +151,15 @@ public class StreamsGroup implements Group {
private final TimelineHashMap<String, Assignment> targetAssignment;
/**
- * Reverse lookup map representing tasks with
- * their current member assignments.
+ * Reverse lookup map representing tasks with their current member assignments.
*/
private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetActiveTasksAssignment;
private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetStandbyTasksAssignment;
private final TimelineHashMap<String, TimelineHashMap<Integer, String>> invertedTargetWarmupTasksAssignment;
/**
- * The current partition epoch maps each topic-partitions to their current epoch where
- * the epoch is the epoch of their owners. When a member revokes a partition, it removes
- * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ * The current partition epoch maps each topic-partitions to their current epoch where the epoch is the epoch of their owners. When a
+ * member revokes a partition, it removes its epochs from this map. When a member gets a partition, it adds its epochs to this map.
*/
private final TimelineHashMap<String, TimelineHashMap<Integer, Integer>> currentActiveTasksEpoch;
private final TimelineHashMap<String, TimelineHashMap<Integer, Integer>> currentStandbyTasksEpoch;
@@ -173,14 +171,16 @@ public class StreamsGroup implements Group {
private final GroupCoordinatorMetricsShard metrics;
/**
- * 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
- * because it is not persisted to the log. The group epoch is here to ensure that the
- * metadata refresh deadline is invalidated if the group epoch does not correspond to
- * the current group epoch. This can happen if the metadata refresh deadline is updated
- * after having refreshed the metadata but the write operation failed. In this case, the
- * time is not automatically rolled back.
+ * The Streams topology.
+ */
+ private final StreamsTopology topology;
+
+ /**
+ * 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
+ * because it is not persisted to the log. The group epoch is here to ensure that the metadata refresh deadline is invalidated if the
+ * group epoch does not correspond to the current group epoch. This can happen if the metadata refresh deadline is updated after having
+ * refreshed the metadata but the write operation failed. In this case, the time is not automatically rolled back.
*/
private DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
@@ -196,6 +196,7 @@ public class StreamsGroup implements Group {
this.members = new TimelineHashMap<>(snapshotRegistry, 0);
this.staticMembers = new TimelineHashMap<>(snapshotRegistry, 0);
this.assignors = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
this.invertedTargetActiveTasksAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
@@ -205,6 +206,7 @@ public class StreamsGroup implements Group {
this.currentStandbyTasksEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
this.currentWarmupTasksEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
this.metrics = Objects.requireNonNull(metrics);
+ this.topology = null; // TODO
}
/**
@@ -241,6 +243,10 @@ public class StreamsGroup implements Group {
.setGroupType(type().toString());
}
+ public StreamsTopology topology() {
+ return topology;
+ }
+
/**
* @return The group id.
*/
@@ -298,11 +304,9 @@ public class StreamsGroup implements Group {
}
/**
- * Get member id of a static member that matches the given group
- * instance id.
+ * Get member id of a static member that matches the given group instance id.
*
* @param groupInstanceId The group instance id.
- *
* @return The member id corresponding to the given instance id or null if it does not exist
*/
public String staticMemberId(String groupInstanceId) {
@@ -310,13 +314,11 @@ public class StreamsGroup implements Group {
}
/**
- * Gets or creates a new member but without adding it to the group. Adding a member
- * is done via the {@link StreamsGroup#updateMember(StreamsGroupMember)} method.
+ * Gets or creates a new member but without adding it to the group. Adding a member is done via the
+ * {@link StreamsGroup#updateMember(StreamsGroupMember)} method.
*
* @param memberId The member id.
- * @param createIfNotExists Booleans indicating whether the member must be
- * created if it does not exist.
- *
+ * @param createIfNotExists Booleans indicating whether the member must be created if it does not exist.
* @return A StreamsGroupMember.
*/
public StreamsGroupMember getOrMaybeCreateMember(
@@ -324,7 +326,9 @@ public class StreamsGroup implements Group {
boolean createIfNotExists
) {
StreamsGroupMember member = members.get(memberId);
- if (member != null) return member;
+ if (member != null) {
+ return member;
+ }
if (!createIfNotExists) {
throw new UnknownMemberIdException(
@@ -339,7 +343,6 @@ public class StreamsGroup implements Group {
* Gets a static member.
*
* @param instanceId The group instance id.
- *
* @return The member corresponding to the given instance id or null if it does not exist
*/
public StreamsGroupMember staticMember(String instanceId) {
@@ -401,7 +404,6 @@ public class StreamsGroup implements Group {
* Returns true if the member exists.
*
* @param memberId The member id.
- *
* @return A boolean indicating whether the member exists or not.
*/
public boolean hasMember(String memberId) {
@@ -432,16 +434,14 @@ public class StreamsGroup implements Group {
/**
* Returns the target assignment of the member.
*
- * @return The StreamsGroupMemberAssignment or an EMPTY one if it does not
- * exist.
+ * @return The StreamsGroupMemberAssignment or an EMPTY one if it does not exist.
*/
public Assignment targetAssignment(String memberId) {
return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
}
/**
- * @return An immutable map containing all the topic partitions
- * with their current member assignments.
+ * @return An immutable map containing all the topic partitions with their current member assignments.
*/
public Map<String, Map<Integer, String>> invertedTargetActiveTasksAssignment() {
return Collections.unmodifiableMap(invertedTargetActiveTasksAssignment);
@@ -488,8 +488,8 @@ public class StreamsGroup implements Group {
/**
* Updates the target assignment of a member.
*
- * @param memberId The member id.
- * @param newTargetAssignment The new target assignment.
+ * @param memberId The member id.
+ * @param newTargetAssignment The new target assignment.
*/
public void updateTargetAssignment(String memberId, Assignment newTargetAssignment) {
updateInvertedTargetActiveTasksAssignment(
@@ -553,9 +553,9 @@ public class StreamsGroup implements Group {
/**
* Updates the reverse lookup map of the target assignment.
*
- * @param memberId The member Id.
- * @param oldTargetAssignment The old target assignment.
- * @param newTargetAssignment The new target assignment.
+ * @param memberId The member Id.
+ * @param oldTargetAssignment The old target assignment.
+ * @param newTargetAssignment The new target assignment.
*/
private void updateInvertedTargetAssignment(
String memberId,
@@ -633,12 +633,10 @@ public class StreamsGroup implements Group {
}
/**
- * Returns the current epoch of a partition or -1 if the partition
- * does not have one.
- *
- * @param topicId The topic id.
- * @param partitionId The partition id.
+ * Returns the current epoch of a partition or -1 if the partition does not have one.
*
+ * @param topicId The topic id.
+ * @param partitionId The partition id.
* @return The epoch or -1.
*/
public int currentActiveTaskEpoch(
@@ -652,6 +650,108 @@ public class StreamsGroup implements Group {
}
}
+ /**
+ * Compute the preferred (server side) assignor for the group while taking into account the updated member. The computation relies on
+ * {{@link StreamsGroup#assignors}} persisted structure but it does not update it.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ * @return An Optional containing the preferred assignor.
+ */
+ public Optional<String> computePreferredServerAssignor(
+ StreamsGroupMember oldMember,
+ StreamsGroupMember newMember
+ ) {
+ // Copy the current count and update it.
+ Map<String, Integer> counts = new HashMap<>(this.assignors);
+ maybeUpdateAssignors(counts, oldMember, newMember);
+
+ return counts.entrySet().stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ }
+
+ /**
+ * @return The preferred assignor for the group.
+ */
+ public Optional<String> preferredServerAssignor() {
+ return preferredServerAssignor(Long.MAX_VALUE);
+ }
+
+ /**
+ * @return The preferred assignor for the group with given offset.
+ */
+ public Optional<String> preferredServerAssignor(long committedOffset) {
+ return assignors.entrySet(committedOffset).stream()
+ .max(Map.Entry.comparingByValue())
+ .map(Map.Entry::getKey);
+ }
+
+ /**
+ * @return An immutable Map of subscription metadata for each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> subscriptionMetadata() {
+ return Collections.unmodifiableMap(subscribedTopicMetadata);
+ }
+
+ /**
+ * Updates the subscription metadata. This replaces the previous one.
+ *
+ * @param subscriptionMetadata The new subscription metadata.
+ */
+ public void setSubscriptionMetadata(
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscribedTopicMetadata.clear();
+ this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ }
+
+ /**
+ * Computes the subscription metadata based on the current subscription info.
+ *
+ * @param subscribedTopicNames Map of topic names to the number of subscribers.
+ * @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.
+ */
+ public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ Map<String, Integer> subscribedTopicNames,
+ TopicsImage topicsImage,
+ ClusterImage clusterImage
+ ) {
+ // Create the topic metadata for each subscribed topic.
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+
+ subscribedTopicNames.forEach((topicName, count) -> {
+ TopicImage topicImage = topicsImage.getTopic(topicName);
+ if (topicImage != null) {
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ topicImage.partitions().forEach((partition, partitionRegistration) -> {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ Optional<String> rackOptional = clusterImage.broker(replica).rack();
+ // Only add the rack if it is available for the broker/replica.
+ rackOptional.ifPresent(racks::add);
+ }
+ // If rack information is unavailable for all replicas of this partition,
+ // no corresponding entry will be stored for it in the map.
+ if (!racks.isEmpty()) {
+ partitionRacks.put(partition, racks);
+ }
+ });
+
+ newSubscriptionMetadata.put(topicName, new TopicMetadata(
+ topicImage.id(),
+ topicImage.name(),
+ topicImage.partitions().size(),
+ partitionRacks)
+ );
+ }
+ });
+
+ return Collections.unmodifiableMap(newSubscriptionMetadata);
+ }
+
/**
* Updates the metadata refresh deadline.
*
@@ -673,11 +773,9 @@ public class StreamsGroup implements Group {
}
/**
- * Checks if a metadata refresh is required. A refresh is required in two cases:
- * 1) The deadline is smaller or equal to the current time;
- * 2) The group epoch associated with the deadline is larger than
- * the current group epoch. This means that the operations which updated
- * the deadline failed.
+ * Checks if a metadata refresh is required. A refresh is required in two cases: 1) The deadline is smaller or equal to the current
+ * time; 2) The group epoch associated with the deadline is larger than the current group epoch. This means that the operations which
+ * updated the deadline failed.
*
* @param currentTimeMs The current time in milliseconds.
* @return A boolean indicating whether a refresh is required or not.
@@ -696,11 +794,10 @@ public class StreamsGroup implements Group {
/**
* Validates the OffsetCommit request.
*
- * @param memberId The member id.
- * @param groupInstanceId The group instance id.
- * @param memberEpoch The member epoch.
- * @param isTransactional Whether the offset commit is transactional or not. It has no
- * impact when a consumer group is used.
+ * @param memberId The member id.
+ * @param groupInstanceId The group instance id.
+ * @param memberEpoch The member epoch.
+ * @param isTransactional Whether the offset commit is transactional or not. It has no impact when a consumer group is used.
*/
@Override
public void validateOffsetCommit(
@@ -712,7 +809,9 @@ public class StreamsGroup implements Group {
// When the member epoch is -1, the request comes from either the admin client
// or a consumer which does not use the group management facility. In this case,
// the request can commit offsets if the group is empty.
- if (memberEpoch < 0 && members().isEmpty()) return;
+ if (memberEpoch < 0 && members().isEmpty()) {
+ return;
+ }
final StreamsGroupMember member = getOrMaybeCreateMember(memberId, false);
validateMemberEpoch(memberEpoch, member.memberEpoch());
@@ -721,9 +820,9 @@ public class StreamsGroup implements Group {
/**
* Validates the OffsetFetch request.
*
- * @param memberId The member id for consumer groups.
- * @param memberEpoch The member epoch for consumer groups.
- * @param lastCommittedOffset The last committed offsets in the timeline.
+ * @param memberId The member id for consumer groups.
+ * @param memberEpoch The member epoch for consumer groups.
+ * @param lastCommittedOffset The last committed offsets in the timeline.
*/
@Override
public void validateOffsetFetch(
@@ -734,7 +833,9 @@ public class StreamsGroup implements Group {
// When the member id is null and the member epoch is -1, the request either comes
// from the admin client or from a client which does not provide them. In this case,
// the fetch request is accepted.
- if (memberId == null && memberEpoch < 0) return;
+ if (memberId == null && memberEpoch < 0) {
+ return;
+ }
final StreamsGroupMember member = members.get(memberId, lastCommittedOffset);
if (member == null) {
@@ -748,7 +849,8 @@ public class StreamsGroup implements Group {
* Validates the OffsetDelete request.
*/
@Override
- public void validateOffsetDelete() {}
+ public void validateOffsetDelete() {
+ }
/**
* Validates the DeleteGroups request.
@@ -773,20 +875,21 @@ public class StreamsGroup implements Group {
@Override
public void createGroupTombstoneRecords(List<CoordinatorRecord> records) {
members().forEach((memberId, member) ->
- records.add(CoordinatorRecordHelpers.newCurrentAssignmentTombstoneRecord(groupId(), memberId))
+ records.add(CoordinatorRecordHelpers.newStreamsCurrentAssignmentTombstoneRecord(groupId(), memberId))
);
members().forEach((memberId, member) ->
- records.add(CoordinatorRecordHelpers.newTargetAssignmentTombstoneRecord(groupId(), memberId))
+ records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentTombstoneRecord(groupId(), memberId))
);
- records.add(CoordinatorRecordHelpers.newTargetAssignmentEpochTombstoneRecord(groupId()));
+ records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentEpochTombstoneRecord(groupId()));
members().forEach((memberId, member) ->
- records.add(CoordinatorRecordHelpers.newMemberSubscriptionTombstoneRecord(groupId(), memberId))
+ records.add(CoordinatorRecordHelpers.newStreamsMemberSubscriptionTombstoneRecord(groupId(), memberId))
);
- records.add(CoordinatorRecordHelpers.newGroupSubscriptionMetadataTombstoneRecord(groupId()));
- records.add(CoordinatorRecordHelpers.newGroupEpochTombstoneRecord(groupId()));
+ records.add(CoordinatorRecordHelpers.newStreamsGroupSubscriptionMetadataTombstoneRecord(groupId()));
+ records.add(CoordinatorRecordHelpers.newStreamsGroupEpochTombstoneRecord(groupId()));
+ records.add(CoordinatorRecordHelpers.newStreamsGroupTopologyRecordTombstone(groupId()));
}
@Override
@@ -810,8 +913,7 @@ public class StreamsGroup implements Group {
}
/**
- * Throws a StaleMemberEpochException if the received member epoch does not match
- * the expected member epoch.
+ * Throws a StaleMemberEpochException if the received member epoch does not match the expected member epoch.
*/
private void validateMemberEpoch(
int receivedMemberEpoch,
@@ -845,30 +947,6 @@ public class StreamsGroup implements Group {
state.set(newState);
}
- /**
- * Compute the subscription type of the consumer group.
- *
- * @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
- *
- * @return {@link org.apache.kafka.coordinator.group.assignor.SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
- * otherwise, {@link org.apache.kafka.coordinator.group.assignor.SubscriptionType#HETEROGENEOUS}.
- */
- public static SubscriptionType subscriptionType(
- Map<String, Integer> subscribedTopicNames,
- int numberOfMembers
- ) {
- if (subscribedTopicNames.isEmpty()) {
- return HOMOGENEOUS;
- }
-
- for (int subscriberCount : subscribedTopicNames.values()) {
- if (subscriberCount != numberOfMembers) {
- return HETEROGENEOUS;
- }
- }
- return HOMOGENEOUS;
- }
-
/**
* Updates the tasks epochs based on the old and the new member.
*
@@ -936,8 +1014,7 @@ public class StreamsGroup implements Group {
*
* @param assignment The assignment.
* @param expectedEpoch The expected epoch.
- * @throws IllegalStateException if the epoch does not match the expected one.
- * package-private for testing.
+ * @throws IllegalStateException if the epoch does not match the expected one. package-private for testing.
*/
private void removeTaskEpochs(
Map<String, Set<Integer>> assignment,
@@ -972,12 +1049,11 @@ public class StreamsGroup implements Group {
/**
* Adds the partitions epoch based on the provided assignment.
*
- * @param activeTasks The assigned active tasks.
- * @param standbyTasks The assigned standby tasks.
- * @param warmupTasks The assigned warmup tasks.
- * @param epoch The new epoch.
- * @throws IllegalStateException if the partition already has an epoch assigned.
- * package-private for testing.
+ * @param activeTasks The assigned active tasks.
+ * @param standbyTasks The assigned standby tasks.
+ * @param warmupTasks The assigned warmup tasks.
+ * @param epoch The new epoch.
+ * @throws IllegalStateException if the partition already has an epoch assigned. package-private for testing.
*/
void addTaskEpochs(
Map<String, Set<Integer>> activeTasks,
@@ -1014,11 +1090,12 @@ public class StreamsGroup implements Group {
}
/**
- * Decrements value by 1; returns null when reaching zero. This helper is
- * meant to be used with Map#compute.
+ * Decrements value by 1; returns null when reaching zero. This helper is meant to be used with Map#compute.
*/
private static Integer decValue(String key, Integer value) {
- if (value == null) return null;
+ if (value == null) {
+ return null;
+ }
return value == 1 ? null : value - 1;
}
@@ -1044,7 +1121,7 @@ public class StreamsGroup implements Group {
records.add(CoordinatorRecordHelpers.newStreamsGroupEpochRecord(groupId(), groupEpoch()));
members().forEach((streamsGroupMemberId, streamsGroupMember) ->
- records.add(CoordinatorRecordHelpers.newTargetAssignmentRecord(
+ records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord(
groupId(),
streamsGroupMemberId,
targetAssignment(streamsGroupMemberId).activeTasks(),
@@ -1054,9 +1131,10 @@ public class StreamsGroup implements Group {
);
records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentEpochRecord(groupId(), groupEpoch()));
+ records.add(CoordinatorRecordHelpers.newStreamsGroupSubscriptionMetadataRecord(groupId(), subscriptionMetadata()));
members().forEach((__, streamsGroupMember) ->
- records.add(CoordinatorRecordHelpers.newCurrentAssignmentRecord(groupId(), streamsGroupMember))
+ records.add(CoordinatorRecordHelpers.newStreamsCurrentAssignmentRecord(groupId(), streamsGroupMember))
);
}
@@ -1083,8 +1161,7 @@ public class StreamsGroup implements Group {
* Checks whether the member has any unreleased tasks.
*
* @param member The member to check.
- * @return A boolean indicating whether the member has partitions in the target
- * assignment that hasn't been revoked by other members.
+ * @return A boolean indicating whether the member has partitions in the target assignment that hasn't been revoked by other members.
*/
public boolean waitingOnUnreleasedActiveTasks(StreamsGroupMember member) {
if (member.state() == MemberState.UNRELEASED_TASKS) {
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
index 0379a58bb2c..e24a4eb966e 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
@@ -18,7 +18,6 @@ package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
-import org.apache.kafka.coordinator.group.consumer.ConsumerGroupMember;
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
import org.apache.kafka.image.TopicImage;
@@ -37,20 +36,18 @@ import java.util.Set;
import java.util.stream.Collectors;
/**
- * StreamsGroupMember contains all the information related to a member
- * within a Streams group. This class is immutable and is fully backed
+ * StreamsGroupMember contains all the information related to a member within a Streams group. This class is immutable and is fully backed
* by records stored in the __consumer_offsets topic.
*/
public class StreamsGroupMember {
/**
- * A builder that facilitates the creation of a new member or the update of
- * an existing one.
- *
- * Please refer to the javadoc of {{@link StreamsGroupMember}} for the
- * definition of the fields.
+ * A builder that facilitates the creation of a new member or the update of an existing one.
+ * <p>
+ * Please refer to the javadoc of {{@link StreamsGroupMember}} for the definition of the fields.
*/
public static class Builder {
+
private final String memberId;
private int memberEpoch = 0;
private int previousMemberEpoch = -1;
@@ -64,9 +61,9 @@ public class StreamsGroupMember {
private String assignor;
private String processId;
private StreamsGroupMemberMetadataValue.HostInfo hostInfo;
- private List<StreamsGroupMemberMetadataValue.KeyValue> clientTags;
+ private Map<String, String> clientTags;
private byte[] userData;
- private List<StreamsGroupMemberMetadataValue.KeyValue> assignmentConfigs;
+ private Map<String, String> assignmentConfigs;
private Map<String, Set<Integer>> assignedActiveTasks = Collections.emptyMap();
private Map<String, Set<Integer>> assignedStandbyTasks = Collections.emptyMap();
private Map<String, Set<Integer>> assignedWarmupTasks = Collections.emptyMap();
@@ -188,7 +185,7 @@ public class StreamsGroupMember {
return this;
}
- public Builder setClientTags(List<StreamsGroupMemberMetadataValue.KeyValue> clientTags) {
+ public Builder setClientTags(Map<String, String> clientTags) {
this.clientTags = clientTags;
return this;
}
@@ -198,7 +195,7 @@ public class StreamsGroupMember {
return this;
}
- public Builder setAssignmentConfigs(List<StreamsGroupMemberMetadataValue.KeyValue> assignmentConfigs) {
+ public Builder setAssignmentConfigs(Map<String, String> assignmentConfigs) {
this.assignmentConfigs = assignmentConfigs;
return this;
}
@@ -233,9 +230,15 @@ public class StreamsGroupMember {
setAssignor(record.assignor());
setProcessId(record.processId());
setHostInfo(record.hostInfo());
- setClientTags(record.clientTags());
+ setClientTags(record.clientTags().stream().collect(Collectors.toMap(
+ StreamsGroupMemberMetadataValue.KeyValue::key,
+ StreamsGroupMemberMetadataValue.KeyValue::value
+ )));
setUserData(record.userData());
- setAssignmentConfigs(record.assignmentConfigs());
+ setAssignmentConfigs(record.assignmentConfigs().stream().collect(Collectors.toMap(
+ StreamsGroupMemberMetadataValue.KeyValue::key,
+ StreamsGroupMemberMetadataValue.KeyValue::value
+ )));
return this;
}
@@ -332,37 +335,37 @@ public class StreamsGroupMember {
/**
* The topology hash
*/
- private byte[] topologyHash;
+ private final byte[] topologyHash;
/**
* The assignor
*/
- private String assignor;
+ private final String assignor;
/**
* The process ID
*/
- private String processId;
+ private final String processId;
/**
* The host info
*/
- private StreamsGroupMemberMetadataValue.HostInfo hostInfo;
+ private final StreamsGroupMemberMetadataValue.HostInfo hostInfo;
/**
* The client tags
*/
- private List<StreamsGroupMemberMetadataValue.KeyValue> clientTags;
+ private final Map<String, String> clientTags;
/**
* The user data
*/
- private byte[] userData;
+ private final byte[] userData;
/**
* The assignment configs
*/
- private List<StreamsGroupMemberMetadataValue.KeyValue> assignmentConfigs;
+ private final Map<String, String> assignmentConfigs;
/**
* Active tasks assigned to this member.
@@ -384,6 +387,7 @@ public class StreamsGroupMember {
*/
private final Map<String, Set<Integer>> activeTasksPendingRevocation;
+ @SuppressWarnings("checkstyle:ParameterNumber")
private StreamsGroupMember(
String memberId,
int memberEpoch,
@@ -397,9 +401,9 @@ public class StreamsGroupMember {
String assignor,
String processId,
StreamsGroupMemberMetadataValue.HostInfo hostInfo,
- List<StreamsGroupMemberMetadataValue.KeyValue> clientTags,
+ Map<String, String> clientTags,
byte[] userData,
- List<StreamsGroupMemberMetadataValue.KeyValue> assignmentConfigs,
+ Map<String, String> assignmentConfigs,
MemberState state,
Map<String, Set<Integer>> assignedActiveTasks,
Map<String, Set<Integer>> assignedStandbyTasks,
@@ -515,7 +519,7 @@ public class StreamsGroupMember {
/**
* @return The client tags
*/
- public List<StreamsGroupMemberMetadataValue.KeyValue> clientTags() {
+ public Map<String, String> clientTags() {
return clientTags;
}
@@ -526,7 +530,7 @@ public class StreamsGroupMember {
/**
* @return The assignment configs
*/
- public List<StreamsGroupMemberMetadataValue.KeyValue> assignmentConfigs() {
+ public Map<String, String> assignmentConfigs() {
return assignmentConfigs;
}
@@ -601,10 +605,15 @@ public class StreamsGroupMember {
}
}
+ @SuppressWarnings("checkstyle:CyclomaticComplexity")
@Override
public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
StreamsGroupMember that = (StreamsGroupMember) o;
return memberEpoch == that.memberEpoch
&& previousMemberEpoch == that.previousMemberEpoch
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
new file mode 100644
index 00000000000..0af7e375af0
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java
@@ -0,0 +1,82 @@
+/*
+ * 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.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Immutable topology metadata.
+ */
+public class StreamsTopology {
+
+ private final byte[] topologyHash;
+
+ private final Map<String, Subtopology> subtopologies;
+
+ public StreamsTopology(final byte[] topologyHash, final Map<String, Subtopology> subtopologies) {
+ this.topologyHash = topologyHash;
+ this.subtopologies = subtopologies;
+ }
+
+ public byte[] topologyHash() {
+ return topologyHash;
+ }
+
+ public Map<String, Subtopology> subtopologies() {
+ return subtopologies;
+ }
+
+ public static StreamsTopology fromRecord(
+ StreamsGroupTopologyValue record
+ ) {
+ return new StreamsTopology(
+ record.topologyHash(),
+ record.topology().stream().collect(Collectors.toMap(Subtopology::subtopology, x -> x))
+ );
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final StreamsTopology that = (StreamsTopology) o;
+ return Objects.deepEquals(topologyHash, that.topologyHash) && Objects.equals(subtopologies, that.subtopologies);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(Arrays.hashCode(topologyHash), subtopologies);
+ }
+
+ @Override
+ public String toString() {
+ return "StreamsTopology{" +
+ "topologyHash=" + Arrays.toString(topologyHash) +
+ ", subtopologies=" + subtopologies +
+ '}';
+ }
+}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/SubscribedTopicMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/SubscribedTopicMetadata.java
deleted file mode 100644
index fb9b1133637..00000000000
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/SubscribedTopicMetadata.java
+++ /dev/null
@@ -1,98 +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.coordinator.group.streams;
-
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;
-import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
-
-import java.util.Collections;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * The subscribed topic metadata class is used by the {@link org.apache.kafka.coordinator.group.assignor.PartitionAssignor} to obtain
- * topic and partition metadata for the topics that the consumer group is subscribed to.
- */
-public class SubscribedTopicMetadata implements SubscribedTopicDescriber {
- /**
- * The topic Ids mapped to their corresponding {@link org.apache.kafka.coordinator.group.consumer.TopicMetadata}
- * object, which contains topic and partition metadata.
- */
- private final Map<Uuid, org.apache.kafka.coordinator.group.consumer.TopicMetadata> topicMetadata;
-
- public SubscribedTopicMetadata(Map<Uuid, org.apache.kafka.coordinator.group.consumer.TopicMetadata> topicMetadata) {
- this.topicMetadata = Objects.requireNonNull(topicMetadata);
- }
-
- /**
- * Map of topic Ids to topic metadata.
- *
- * @return The map of topic Ids to topic metadata.
- */
- public Map<Uuid, org.apache.kafka.coordinator.group.consumer.TopicMetadata> topicMetadata() {
- return this.topicMetadata;
- }
-
- /**
- * The number of partitions for the given topic Id.
- *
- * @param topicId Uuid corresponding to the topic.
- * @return The number of partitions corresponding to the given topic Id,
- * or -1 if the topic Id does not exist.
- */
- @Override
- public int numPartitions(Uuid topicId) {
- org.apache.kafka.coordinator.group.consumer.TopicMetadata topic = this.topicMetadata.get(topicId);
- return topic == null ? -1 : topic.numPartitions();
- }
-
- /**
- * Returns all the available racks associated with the replicas of the given partition.
- *
- * @param topicId Uuid corresponding to the partition's topic.
- * @param partition Partition Id within the topic.
- * @return The set of racks corresponding to the replicas of the topics partition.
- * If the topic Id does not exist or no partition rack information is available, an empty set is returned.
- */
- @Override
- public Set<String> racksForPartition(Uuid topicId, int partition) {
- TopicMetadata topic = this.topicMetadata.get(topicId);
- return topic == null ? Collections.emptySet() : topic.partitionRacks().getOrDefault(partition, Collections.emptySet());
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- SubscribedTopicMetadata that = (SubscribedTopicMetadata) o;
- return topicMetadata.equals(that.topicMetadata);
- }
-
- @Override
- public int hashCode() {
- return topicMetadata.hashCode();
- }
-
- @Override
- public String toString() {
- return "SubscribedTopicMetadata(" +
- "topicMetadata=" + topicMetadata +
- ')';
- }
-}
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
new file mode 100644
index 00000000000..26bc409cb7b
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
@@ -0,0 +1,390 @@
+/*
+ * 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.coordinator.group.CoordinatorRecord;
+import org.apache.kafka.coordinator.group.CoordinatorRecordHelpers;
+import org.apache.kafka.coordinator.group.taskassignor.AssignmentMemberSpec;
+import org.apache.kafka.coordinator.group.taskassignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.taskassignor.GroupSpecImpl;
+import org.apache.kafka.coordinator.group.taskassignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.taskassignor.TaskAssignor;
+import org.apache.kafka.coordinator.group.taskassignor.TaskAssignorException;
+import org.apache.kafka.image.TopicsImage;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpers.newTargetAssignmentEpochRecord;
+
+/**
+ * Build a new Target Assignment based on the provided parameters. As a result, it yields the records that must be persisted to the log and
+ * the new member assignments as a map.
+ * <p>
+ * Records are only created for members which have a new target assignment. If their assignment did not change, no new record is needed.
+ * <p>
+ * When a member is deleted, it is assumed that its target assignment record is deleted as part of the member deletion process. In other
+ * words, this class does not yield a tombstone for removed members.
+ */
+public class TargetAssignmentBuilder {
+
+ /**
+ * The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
+ */
+ public static class TargetAssignmentResult {
+
+ /**
+ * The records that must be applied to the __streams_offsets topics to persist the new target assignment.
+ */
+ private final List<CoordinatorRecord> records;
+
+ /**
+ * The new target assignment for the group.
+ */
+ private final Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment;
+
+ TargetAssignmentResult(
+ List<CoordinatorRecord> records,
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment
+ ) {
+ Objects.requireNonNull(records);
+ Objects.requireNonNull(targetAssignment);
+ this.records = records;
+ this.targetAssignment = targetAssignment;
+ }
+
+ /**
+ * @return The records.
+ */
+ public List<CoordinatorRecord> records() {
+ return records;
+ }
+
+ /**
+ * @return The target assignment.
+ */
+ public Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment() {
+ return targetAssignment;
+ }
+ }
+
+ /**
+ * The group id.
+ */
+ private final String groupId;
+
+ /**
+ * The group epoch.
+ */
+ private final int groupEpoch;
+
+ /**
+ * The partition assignor used to compute the assignment.
+ */
+ private final TaskAssignor assignor;
+
+ /**
+ * The members in the group.
+ */
+ private Map<String, StreamsGroupMember> members = Collections.emptyMap();
+
+ /**
+ * The subscription metadata.
+ */
+ private Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata = Collections.emptyMap();
+
+ /**
+ * The existing target assignment.
+ */
+ private Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment = Collections.emptyMap();
+
+ /**
+ * Reverse lookup map representing tasks with their current member assignments.
+ */
+ private Map<String, Map<Integer, String>> invertedTargetAssignment = Collections.emptyMap();
+
+ /**
+ * The topics image.
+ */
+ private TopicsImage topicsImage = TopicsImage.EMPTY;
+
+ /**
+ * The topology.
+ */
+ private StreamsTopology topology;
+
+ /**
+ * The members which have been updated or deleted. Deleted members are signaled by a null value.
+ */
+ private final Map<String, StreamsGroupMember> updatedMembers = new HashMap<>();
+
+ /**
+ * The static members in the group.
+ */
+ private Map<String, String> staticMembers = new HashMap<>();
+
+ /**
+ * Constructs the object.
+ *
+ * @param groupId The group id.
+ * @param groupEpoch The group epoch to compute a target assignment for.
+ * @param assignor The assignor to use to compute the target assignment.
+ */
+ public TargetAssignmentBuilder(
+ String groupId,
+ int groupEpoch,
+ TaskAssignor assignor,
+ StreamsTopology topology
+ ) {
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = groupEpoch;
+ this.assignor = Objects.requireNonNull(assignor);
+ this.topology = Objects.requireNonNull(topology);
+ }
+
+ /**
+ * Adds all the existing members.
+ *
+ * @param members The existing members in the streams group.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withMembers(
+ Map<String, StreamsGroupMember> members
+ ) {
+ this.members = members;
+ return this;
+ }
+
+ /**
+ * Adds all the existing static members.
+ *
+ * @param staticMembers The existing static members in the streams group.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withStaticMembers(
+ Map<String, String> staticMembers
+ ) {
+ this.staticMembers = staticMembers;
+ return this;
+ }
+
+ /**
+ * Adds the subscription metadata to use.
+ *
+ * @param subscriptionMetadata The subscription metadata.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withSubscriptionMetadata(
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscriptionMetadata = subscriptionMetadata;
+ return this;
+ }
+
+ /**
+ * Adds the existing target assignment.
+ *
+ * @param targetAssignment The existing target assignment.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withTargetAssignment(
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment
+ ) {
+ this.targetAssignment = targetAssignment;
+ return this;
+ }
+
+ /**
+ * Adds the existing topic partition assignments.
+ *
+ * @param invertedTargetAssignment The reverse lookup map of the current target assignment.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withInvertedTargetAssignment(
+ Map<String, Map<Integer, String>> invertedTargetAssignment
+ ) {
+ this.invertedTargetAssignment = invertedTargetAssignment;
+ return this;
+ }
+
+ /**
+ * Adds the topics image.
+ *
+ * @param topicsImage The topics image.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder withTopicsImage(
+ TopicsImage topicsImage
+ ) {
+ this.topicsImage = topicsImage;
+ return this;
+ }
+
+ /**
+ * Adds or updates a member. This is useful when the updated member is not yet materialized in memory.
+ *
+ * @param memberId The member id.
+ * @param member The member to add or update.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder addOrUpdateMember(
+ String memberId,
+ StreamsGroupMember member
+ ) {
+ this.updatedMembers.put(memberId, member);
+ return this;
+ }
+
+ /**
+ * Removes a member. This is useful when the removed member is not yet materialized in memory.
+ *
+ * @param memberId The member id.
+ * @return This object.
+ */
+ public TargetAssignmentBuilder removeMember(
+ String memberId
+ ) {
+ return addOrUpdateMember(memberId, null);
+ }
+
+ /**
+ * Builds the new target assignment.
+ *
+ * @return A TargetAssignmentResult which contains the records to update the existing target assignment.
+ * @throws TaskAssignorException if the target assignment cannot be computed.
+ */
+ public TargetAssignmentResult build() throws TaskAssignorException {
+ Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+
+ // Prepare the member spec for all members.
+ members.forEach((memberId, member) -> memberSpecs.put(memberId, createAssignmentMemberSpec(
+ member,
+ targetAssignment.getOrDefault(memberId, org.apache.kafka.coordinator.group.streams.Assignment.EMPTY)
+ )));
+
+ // Update the member spec if updated or deleted members.
+ updatedMembers.forEach((memberId, updatedMemberOrNull) -> {
+ if (updatedMemberOrNull == null) {
+ memberSpecs.remove(memberId);
+ } else {
+ org.apache.kafka.coordinator.group.streams.Assignment assignment = targetAssignment.getOrDefault(memberId,
+ org.apache.kafka.coordinator.group.streams.Assignment.EMPTY);
+
+ // A new static member joins and needs to replace an existing departed one.
+ if (updatedMemberOrNull.instanceId() != null) {
+ String previousMemberId = staticMembers.get(updatedMemberOrNull.instanceId());
+ if (previousMemberId != null && !previousMemberId.equals(memberId)) {
+ assignment = targetAssignment.getOrDefault(previousMemberId,
+ org.apache.kafka.coordinator.group.streams.Assignment.EMPTY);
+ }
+ }
+
+ memberSpecs.put(memberId, createAssignmentMemberSpec(
+ updatedMemberOrNull,
+ assignment
+ ));
+ }
+ });
+
+ // Compute the assignment.
+ GroupAssignment newGroupAssignment = assignor.assign(
+ new GroupSpecImpl(
+ Collections.unmodifiableMap(memberSpecs),
+ new ArrayList<>(topology.subtopologies().keySet())
+ ),
+ new TopologyMetadata(subscriptionMetadata, topology)
+ );
+
+ // Compute delta from previous to new target assignment and create the
+ // relevant records.
+ List<CoordinatorRecord> records = new ArrayList<>();
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> newTargetAssignment = new HashMap<>();
+
+ memberSpecs.keySet().forEach(memberId -> {
+ org.apache.kafka.coordinator.group.streams.Assignment oldMemberAssignment = targetAssignment.get(memberId);
+ org.apache.kafka.coordinator.group.streams.Assignment newMemberAssignment = newMemberAssignment(newGroupAssignment, memberId);
+
+ newTargetAssignment.put(memberId, newMemberAssignment);
+
+ if (oldMemberAssignment == null) {
+ // If the member had no assignment, we always create a record for it.
+ records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord(
+ groupId,
+ memberId,
+ newMemberAssignment.activeTasks(),
+ newMemberAssignment.standbyTasks(),
+ newMemberAssignment.warmupTasks()
+ ));
+ } else {
+ // If the member had an assignment, we only create a record if the
+ // new assignment is different.
+ if (!newMemberAssignment.equals(oldMemberAssignment)) {
+ records.add(CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord(
+ groupId,
+ memberId,
+ newMemberAssignment.activeTasks(),
+ newMemberAssignment.standbyTasks(),
+ newMemberAssignment.warmupTasks()
+ ));
+ }
+ }
+ });
+
+ // Bump the target assignment epoch.
+ records.add(newTargetAssignmentEpochRecord(groupId, groupEpoch));
+
+ return new TargetAssignmentResult(records, newTargetAssignment);
+ }
+
+ private org.apache.kafka.coordinator.group.streams.Assignment newMemberAssignment(
+ org.apache.kafka.coordinator.group.taskassignor.GroupAssignment newGroupAssignment,
+ String memberId
+ ) {
+ MemberAssignment newMemberAssignment = newGroupAssignment.members().get(memberId);
+ if (newMemberAssignment != null) {
+ return new org.apache.kafka.coordinator.group.streams.Assignment(
+ newMemberAssignment.activeTasks(),
+ newMemberAssignment.standbyTasks(),
+ newMemberAssignment.warmupTasks()
+ );
+ } else {
+ return org.apache.kafka.coordinator.group.streams.Assignment.EMPTY;
+ }
+ }
+
+ static AssignmentMemberSpec createAssignmentMemberSpec(
+ StreamsGroupMember member,
+ Assignment targetAssignment
+ ) {
+ return new AssignmentMemberSpec(
+ Optional.ofNullable(member.instanceId()),
+ Optional.ofNullable(member.rackId()),
+ targetAssignment.activeTasks(),
+ targetAssignment.standbyTasks(),
+ targetAssignment.warmupTasks(),
+ member.processId(),
+ member.clientTags(),
+ member.assignmentConfigs(),
+ Collections.emptyMap() // TODO: TaskOffsets is missing
+ );
+ }
+}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicIds.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicIds.java
deleted file mode 100644
index 7beb9b9e4ff..00000000000
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicIds.java
+++ /dev/null
@@ -1,176 +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.coordinator.group.streams;
-
-import org.apache.kafka.common.Uuid;
-import org.apache.kafka.image.TopicImage;
-import org.apache.kafka.image.TopicsImage;
-
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * TopicIds is initialized with topic names (String) but exposes a Set of topic ids (Uuid) to the
- * user and performs the conversion lazily with TopicsImage.
- */
-public class TopicIds implements Set<Uuid> {
- private final Set<String> topicNames;
- private final TopicsImage image;
-
- public TopicIds(
- Set<String> topicNames,
- TopicsImage image
- ) {
- this.topicNames = Objects.requireNonNull(topicNames);
- this.image = Objects.requireNonNull(image);
- }
-
- @Override
- public int size() {
- return topicNames.size();
- }
-
- @Override
- public boolean isEmpty() {
- return topicNames.isEmpty();
- }
-
- @Override
- public boolean contains(Object o) {
- if (o instanceof Uuid) {
- Uuid topicId = (Uuid) o;
- TopicImage topicImage = image.getTopic(topicId);
- if (topicImage == null) return false;
- return topicNames.contains(topicImage.name());
- }
- return false;
- }
-
- private static class TopicIdIterator implements Iterator<Uuid> {
- final Iterator<String> iterator;
- final TopicsImage image;
- private Uuid next = null;
-
- private TopicIdIterator(
- Iterator<String> iterator,
- TopicsImage image
- ) {
- this.iterator = Objects.requireNonNull(iterator);
- this.image = Objects.requireNonNull(image);
- }
-
- @Override
- public boolean hasNext() {
- if (next != null) return true;
- Uuid result = null;
- do {
- if (!iterator.hasNext()) {
- return false;
- }
- String next = iterator.next();
- TopicImage topicImage = image.getTopic(next);
- if (topicImage != null) {
- result = topicImage.id();
- }
- } while (result == null);
- next = result;
- return true;
- }
-
- @Override
- public Uuid next() {
- if (!hasNext()) throw new NoSuchElementException();
- Uuid result = next;
- next = null;
- return result;
- }
- }
-
- @Override
- public Iterator<Uuid> iterator() {
- return new TopicIdIterator(topicNames.iterator(), image);
- }
-
- @Override
- public Object[] toArray() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public <T> T[] toArray(T[] a) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean add(Uuid o) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean remove(Object o) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean addAll(Collection c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void clear() {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean removeAll(Collection c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean retainAll(Collection c) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean containsAll(Collection c) {
- for (Object o : c) {
- if (!contains(o)) return false;
- }
- return true;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- TopicIds uuids = (TopicIds) o;
-
- if (!Objects.equals(topicNames, uuids.topicNames)) return false;
- return Objects.equals(image, uuids.image);
- }
-
- @Override
- public int hashCode() {
- int result = topicNames.hashCode();
- result = 31 * result + image.hashCode();
- return result;
- }
-}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java
index b03afea1be1..b319b78914a 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java
@@ -30,6 +30,7 @@ import java.util.Set;
* Immutable topic metadata.
*/
public class TopicMetadata {
+
/**
* The topic id.
*/
@@ -46,8 +47,8 @@ public class TopicMetadata {
private final int numPartitions;
/**
- * Map of every partition Id to a set of its rack Ids, if they exist.
- * If rack information is unavailable for all partitions, this is an empty map.
+ * Map of every partition Id to a set of its rack Ids, if they exist. If rack information is unavailable for all partitions, this is an
+ * empty map.
*/
private final Map<Integer, Set<String>> partitionRacks;
@@ -94,8 +95,8 @@ public class TopicMetadata {
}
/**
- * @return Every partition mapped to the set of corresponding available rack Ids of its replicas.
- * An empty map is returned if rack information is unavailable for all partitions.
+ * @return Every partition mapped to the set of corresponding available rack Ids of its replicas. An empty map is returned if rack
+ * information is unavailable for all partitions.
*/
public Map<Integer, Set<String>> partitionRacks() {
return this.partitionRacks;
@@ -103,14 +104,24 @@ public class TopicMetadata {
@Override
public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
TopicMetadata that = (TopicMetadata) o;
- if (!id.equals(that.id)) return false;
- if (!name.equals(that.name)) return false;
- if (numPartitions != that.numPartitions) return false;
+ if (!id.equals(that.id)) {
+ return false;
+ }
+ if (!name.equals(that.name)) {
+ return false;
+ }
+ if (numPartitions != that.numPartitions) {
+ return false;
+ }
return partitionRacks.equals(that.partitionRacks);
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
new file mode 100644
index 00000000000..ebc0f239f7d
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
@@ -0,0 +1,105 @@
+/*
+ * 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.coordinator.group.generated.StreamsGroupTopologyValue.Subtopology;
+import org.apache.kafka.coordinator.group.taskassignor.TopologyDescriber;
+
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The topology metadata class is used by the {@link org.apache.kafka.coordinator.group.taskassignor.TaskAssignor} to obtain topic and
+ * partition metadata for the topology that the streams group using.
+ */
+public class TopologyMetadata implements TopologyDescriber {
+
+ /**
+ * The topic Ids mapped to their corresponding {@link org.apache.kafka.coordinator.group.streams.TopicMetadata} object, which contains
+ * topic and partition metadata.
+ */
+ private final Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> topicMetadata;
+
+ private final StreamsTopology topology;
+
+ public TopologyMetadata(
+ Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> topicMetadata,
+ StreamsTopology topology
+ ) {
+ this.topicMetadata = Objects.requireNonNull(topicMetadata);
+ this.topology = Objects.requireNonNull(topology);
+ }
+
+ /**
+ * Map of topic names to topic metadata.
+ *
+ * @return The map of topic Ids to topic metadata.
+ */
+ public Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> topicMetadata() {
+ return this.topicMetadata;
+ }
+
+ public StreamsTopology topology() {
+ return topology;
+ }
+
+ /**
+ * The number of partitions for the given subtopology ID.
+ *
+ * @param subtopologyId ID of the corresponding subtopology
+ * @return The number of partitions corresponding to the given subtopology ID, or -1 if the subtopology ID does not exist.
+ */
+ @Override
+ public int numPartitions(String subtopologyId) {
+ final Subtopology subtopology = topology.subtopologies().get(subtopologyId);
+ if (subtopology == null) {
+ return -1;
+ }
+ // TODO: We need to validate the validity of the subtopology here or somewhere else
+ final String firstSourceTopic = subtopology.sourceTopics().get(0);
+ if (firstSourceTopic == null) {
+ return -1;
+ }
+ org.apache.kafka.coordinator.group.streams.TopicMetadata topic = this.topicMetadata.get(firstSourceTopic);
+ return topic == null ? -1 : topic.numPartitions();
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final TopologyMetadata that = (TopologyMetadata) o;
+ return Objects.equals(topicMetadata, that.topicMetadata) && Objects.equals(topology, that.topology);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicMetadata, topology);
+ }
+
+ @Override
+ public String toString() {
+ return "TopologyMetadata{" +
+ "topicMetadata=" + topicMetadata +
+ ", topology=" + topology +
+ '}';
+ }
+}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
index 6d8cb6b906c..30f775fa1f8 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupCurrentMemberAssignmentKey.json
@@ -20,9 +20,9 @@
"validVersions": "14",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "8",
+ { "name": "GroupId", "type": "string", "versions": "14",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "8",
+ { "name": "MemberId", "type": "string", "versions": "14",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
index bc54bc8010b..0834b3e41ee 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMemberMetadataKey.json
@@ -20,9 +20,9 @@
"validVersions": "11",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "5",
+ { "name": "GroupId", "type": "string", "versions": "11",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "5",
+ { "name": "MemberId", "type": "string", "versions": "11",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
index 0ad8fbce88b..8f287c05652 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupMetadataKey.json
@@ -20,7 +20,7 @@
"validVersions": "9",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "3",
+ { "name": "GroupId", "type": "string", "versions": "9",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
index 52b879dbcd1..954e5fd1182 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupPartitionMetadataKey.json
@@ -20,7 +20,7 @@
"validVersions": "10",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "4",
+ { "name": "GroupId", "type": "string", "versions": "10",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
index 6faac03a9e2..128bf44bba7 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMemberKey.json
@@ -20,9 +20,9 @@
"validVersions": "13",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "7",
+ { "name": "GroupId", "type": "string", "versions": "13",
"about": "The group id." },
- { "name": "MemberId", "type": "string", "versions": "7",
+ { "name": "MemberId", "type": "string", "versions": "13",
"about": "The member id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
index 91f58c07f64..b583b3224b4 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTargetAssignmentMetadataKey.json
@@ -20,7 +20,7 @@
"validVersions": "12",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "6",
+ { "name": "GroupId", "type": "string", "versions": "12",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
index 0e72d8c3372..11873d4403c 100644
--- a/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
+++ b/group-coordinator/src/main/resources/common/message/StreamsGroupTopologyKey.json
@@ -20,7 +20,7 @@
"validVersions": "15",
"flexibleVersions": "none",
"fields": [
- { "name": "GroupId", "type": "string", "versions": "3",
+ { "name": "GroupId", "type": "string", "versions": "15",
"about": "The group id." }
]
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java
index 9759f41d727..489d0d1fdd0 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/AssignmentTestUtil.java
@@ -71,25 +71,6 @@ public class AssignmentTestUtil {
return Collections.unmodifiableMap(assignment);
}
- @SafeVarargs
- public static Map<String, Set<Integer>> mkStreamsAssignment(Map.Entry<String, Set<Integer>>... entries) {
- Map<String, Set<Integer>> assignment = new HashMap<>();
- for (Map.Entry<String, Set<Integer>> entry : entries) {
- assignment.put(entry.getKey(), entry.getValue());
- }
- return assignment;
- }
-
- public static Map.Entry<String, Set<Integer>> mkTaskAssignment(
- String subtopology,
- Integer... partitions
- ) {
- return new AbstractMap.SimpleEntry<>(
- subtopology,
- new HashSet<>(Arrays.asList(partitions))
- );
- }
-
/**
* Verifies that the expected assignment is equal to the computed assignment for every member in the group.
*/
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentTest.java
index 49ba49753bc..ccec91a46db 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/AssignmentTest.java
@@ -26,8 +26,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkStreamsAssignment;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTaskAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTaskAssignment;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -42,13 +42,13 @@ public class AssignmentTest {
@Test
public void testAttributes() {
- Map<String, Set<Integer>> activeTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> activeTasks = mkAssignment(
mkTaskAssignment("subtopology1", 1, 2, 3)
);
- Map<String, Set<Integer>> standbyTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> standbyTasks = mkAssignment(
mkTaskAssignment("subtopology2", 9, 8, 7)
);
- Map<String, Set<Integer>> warmupTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> warmupTasks = mkAssignment(
mkTaskAssignment("subtopology3", 4, 5, 6)
);
Assignment assignment = new Assignment(activeTasks, standbyTasks, warmupTasks);
@@ -92,21 +92,21 @@ public class AssignmentTest {
Assignment assignment = Assignment.fromRecord(record);
assertEquals(
- mkStreamsAssignment(
+ mkAssignment(
mkTaskAssignment(subtopology1, 1, 2, 3),
mkTaskAssignment(subtopology2, 4, 5, 6)
),
assignment.activeTasks()
);
assertEquals(
- mkStreamsAssignment(
+ mkAssignment(
mkTaskAssignment(subtopology1, 7, 8, 9),
mkTaskAssignment(subtopology2, 1, 2, 3)
),
assignment.standbyTasks()
);
assertEquals(
- mkStreamsAssignment(
+ mkAssignment(
mkTaskAssignment(subtopology1, 4, 5, 6),
mkTaskAssignment(subtopology2, 7, 8, 9)
),
@@ -116,13 +116,13 @@ public class AssignmentTest {
@Test
public void testEquals() {
- Map<String, Set<Integer>> activeTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> activeTasks = mkAssignment(
mkTaskAssignment("subtopology1", 1, 2, 3)
);
- Map<String, Set<Integer>> standbyTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> standbyTasks = mkAssignment(
mkTaskAssignment("subtopology2", 9, 8, 7)
);
- Map<String, Set<Integer>> warmupTasks = mkStreamsAssignment(
+ Map<String, Set<Integer>> warmupTasks = mkAssignment(
mkTaskAssignment("subtopology3", 4, 5, 6)
);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
new file mode 100644
index 00000000000..e3531e11a37
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilderTest.java
@@ -0,0 +1,576 @@
+/*
+ * 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.Uuid;
+import org.apache.kafka.common.errors.FencedMemberEpochException;
+import org.apache.kafka.common.message.StreamsHeartbeatRequestData;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTaskAssignment;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class CurrentAssignmentBuilderTest {
+
+ @Test
+ public void testStableToStable() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testStableToStableWithNewTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6, 7)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testStableToUnrevokedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1),
+ mkTaskAssignment(subtopologyId2, 4)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testStableToUnreleasedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testStableToUnreleasedTasksWithOwnedTasksNotHavingRevokedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, __) ->
+ subtopologyId2.equals(subtopologyId) ? 10 : -1
+ )
+ .withOwnedActiveTasks(Collections.emptyList())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 4, 5)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnrevokedTasksToStable() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1),
+ mkTaskAssignment(subtopologyId2, 4)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1)
+ .withOwnedActiveTasks(Arrays.asList(
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId1)
+ .setPartitions(Arrays.asList(2, 3)),
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId2)
+ .setPartitions(Arrays.asList(5, 6))))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testRemainsInUnrevokedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1),
+ mkTaskAssignment(subtopologyId2, 4)))
+ .build();
+
+ org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder currentAssignmentBuilder = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(
+ member)
+ .withTargetAssignment(12, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1);
+
+ assertEquals(
+ member,
+ currentAssignmentBuilder
+ .withOwnedActiveTasks(null)
+ .build()
+ );
+
+ assertEquals(
+ member,
+ currentAssignmentBuilder
+ .withOwnedActiveTasks(Arrays.asList(
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId1)
+ .setPartitions(Arrays.asList(1, 2, 3)),
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId2)
+ .setPartitions(Arrays.asList(5, 6))))
+ .build()
+ );
+
+ assertEquals(
+ member,
+ currentAssignmentBuilder
+ .withOwnedActiveTasks(Arrays.asList(
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId1)
+ .setPartitions(Arrays.asList(2, 3)),
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId2)
+ .setPartitions(Arrays.asList(4, 5, 6))))
+ .build()
+ );
+ }
+
+ @Test
+ public void testUnrevokedTasksToUnrevokedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 1),
+ mkTaskAssignment(subtopologyId2, 4)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(12, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1)
+ .withOwnedActiveTasks(Arrays.asList(
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId1)
+ .setPartitions(Arrays.asList(2, 3)),
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId2)
+ .setPartitions(Arrays.asList(5, 6))))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2),
+ mkTaskAssignment(subtopologyId2, 5)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnrevokedTasksToUnreleasedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(10)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .withOwnedActiveTasks(Arrays.asList(
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId1)
+ .setPartitions(Arrays.asList(2, 3)),
+ new StreamsHeartbeatRequestData.TaskIds()
+ .setSubtopology(subtopologyId2)
+ .setPartitions(Arrays.asList(5, 6))))
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnreleasedTasksToStable() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(12, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(12)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnreleasedTasksToStableWithNewTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> -1)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.STABLE)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 5, 6, 7)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnreleasedTasksToUnreleasedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(11, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3, 4),
+ mkTaskAssignment(subtopologyId2, 5, 6, 7))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build();
+
+ assertEquals(member, updatedMember);
+ }
+
+ @Test
+ public void testUnreleasedTasksToUnrevokedTasks() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNRELEASED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2, 3),
+ mkTaskAssignment(subtopologyId2, 5, 6)))
+ .build();
+
+ StreamsGroupMember updatedMember = new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(12, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNREVOKED_TASKS)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2),
+ mkTaskAssignment(subtopologyId2, 5)))
+ .build(),
+ updatedMember
+ );
+ }
+
+ @Test
+ public void testUnknownState() {
+ String subtopologyId1 = Uuid.randomUuid().toString();
+ String subtopologyId2 = Uuid.randomUuid().toString();
+
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member")
+ .setState(org.apache.kafka.coordinator.group.streams.MemberState.UNKNOWN)
+ .setMemberEpoch(11)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6)))
+ .setActiveTasksPendingRevocation(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 2),
+ mkTaskAssignment(subtopologyId2, 5)))
+ .build();
+
+ // When the member is in an unknown state, the member is first to force
+ // a reset of the client side member state.
+ assertThrows(FencedMemberEpochException.class, () -> new org.apache.kafka.coordinator.group.streams.CurrentAssignmentBuilder(member)
+ .withTargetAssignment(12, new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 10)
+ .build());
+
+ // Then the member rejoins with no owned tasks.
+ StreamsGroupMember updatedMember = new CurrentAssignmentBuilder(member)
+ .withTargetAssignment(12, new Assignment(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6))))
+ .withCurrentPartitionEpoch((subtopologyId, partitionId) -> 11)
+ .withOwnedActiveTasks(Collections.emptyList())
+ .build();
+
+ assertEquals(
+ new StreamsGroupMember.Builder("member")
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(12)
+ .setPreviousMemberEpoch(11)
+ .setAssignedActiveTasks(mkAssignment(
+ mkTaskAssignment(subtopologyId1, 3),
+ mkTaskAssignment(subtopologyId2, 6)))
+ .build(),
+ updatedMember
+ );
+ }
+}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
index 5041aa9cf07..5995381e7b4 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
@@ -17,18 +17,23 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue.TaskIds;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue.KeyValue;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Optional;
import java.util.OptionalInt;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkStreamsAssignment;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTaskAssignment;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTaskAssignment;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
public class StreamsGroupMemberTest {
@@ -49,13 +54,13 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(mkMap(mkEntry("client", "tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")))
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
- .setAssignedStandbyTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
- .setAssignedWarmupTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
+ .setAssignmentConfigs(mkMap(mkEntry("assignment", "config")))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
+ .setAssignedStandbyTasks(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
+ .setAssignedWarmupTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
+ .setActiveTasksPendingRevocation(mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
.build();
assertEquals("member-id", member.memberId());
@@ -65,33 +70,33 @@ public class StreamsGroupMemberTest {
assertEquals("rack-id", member.rackId());
assertEquals("client-id", member.clientId());
assertEquals("hostname", member.clientHost());
- assertTrue(Arrays.equals("topology-hash".getBytes(), member.topologyHash()));
+ assertArrayEquals("topology-hash".getBytes(), member.topologyHash());
assertEquals("assignor", member.assignor().get());
assertEquals("process-id", member.processId());
assertEquals(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090), member.hostInfo());
assertEquals(
- Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")),
+ mkMap(mkEntry("client", "tag")),
member.clientTags()
);
- assertTrue(Arrays.equals("user-data".getBytes(), member.userData()));
+ assertArrayEquals("user-data".getBytes(), member.userData());
assertEquals(
- Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")),
+ mkMap(mkEntry("assignment", "config")),
member.assignmentConfigs()
);
assertEquals(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)),
+ mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)),
member.assignedActiveTasks()
);
assertEquals(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)),
+ mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)),
member.assignedStandbyTasks()
);
assertEquals(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)),
+ mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)),
member.assignedWarmupTasks()
);
assertEquals(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)),
+ mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)),
member.activeTasksPendingRevocation()
);
}
@@ -113,13 +118,13 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(mkMap(mkEntry("client", "tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")))
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
- .setAssignedStandbyTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
- .setAssignedWarmupTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
+ .setAssignmentConfigs(mkMap(mkEntry("assignment", "config")))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
+ .setAssignedStandbyTasks(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
+ .setAssignedWarmupTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
+ .setActiveTasksPendingRevocation(mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
.build();
StreamsGroupMember member2 = new StreamsGroupMember.Builder("member-id")
@@ -134,13 +139,13 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(mkMap(mkEntry("client", "tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")))
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
- .setAssignedStandbyTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
- .setAssignedWarmupTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
+ .setAssignmentConfigs(mkMap(mkEntry("assignment", "config")))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
+ .setAssignedStandbyTasks(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
+ .setAssignedWarmupTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
+ .setActiveTasksPendingRevocation(mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
.build();
StreamsGroupMember member3 = new StreamsGroupMember.Builder("member3-id")
@@ -155,13 +160,13 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(mkMap(mkEntry("client", "tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")))
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
- .setAssignedStandbyTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
- .setAssignedWarmupTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
+ .setAssignmentConfigs(mkMap(mkEntry("assignment", "config")))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
+ .setAssignedStandbyTasks(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
+ .setAssignedWarmupTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
+ .setActiveTasksPendingRevocation(mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
.build();
assertEquals(member1, member2);
@@ -185,13 +190,13 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(mkMap(mkEntry("client", "tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")))
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
- .setAssignedStandbyTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
- .setAssignedWarmupTasks(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
+ .setAssignmentConfigs(mkMap(mkEntry("assignment", "config")))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)))
+ .setAssignedStandbyTasks(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)))
+ .setAssignedWarmupTasks(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)))
+ .setActiveTasksPendingRevocation(mkAssignment(mkTaskAssignment(subtopologyId2, 3, 2, 1)))
.build();
// This is a no-op.
@@ -214,7 +219,7 @@ public class StreamsGroupMemberTest {
assertEquals("new-instance-id", updatedMember.instanceId());
assertEquals("new-rack-id", updatedMember.rackId());
assertEquals(6000, updatedMember.rebalanceTimeoutMs());
- assertEquals("new-assignor", updatedMember.assignor());
+ assertEquals("new-assignor", updatedMember.assignor().get());
}
@Test
@@ -229,9 +234,9 @@ public class StreamsGroupMemberTest {
.setAssignor("assignor")
.setProcessId("process-id")
.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090))
- .setClientTags(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")))
+ .setClientTags(Collections.singletonList(new KeyValue().setKey("client").setValue("tag")))
.setUserData("user-data".getBytes())
- .setAssignmentConfigs(Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")));
+ .setAssignmentConfigs(Collections.singletonList(new KeyValue().setKey("assignment").setValue("config")));
StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
.updateWith(record)
@@ -242,17 +247,17 @@ public class StreamsGroupMemberTest {
assertEquals("client-id", member.clientId());
assertEquals("host-id", member.clientHost());
assertEquals(1000, member.rebalanceTimeoutMs());
- assertTrue(Arrays.equals("topology-hash".getBytes(), member.topologyHash()));
+ assertArrayEquals("topology-hash".getBytes(), member.topologyHash());
assertEquals("assignor", member.assignor().get());
assertEquals("process-id", member.processId());
assertEquals(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090), member.hostInfo());
assertEquals(
- Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("client").setValue("tag")),
+ mkMap(mkEntry("client", "tag")),
member.clientTags()
);
- assertTrue(Arrays.equals("user-data".getBytes(), member.userData()));
+ assertArrayEquals("user-data".getBytes(), member.userData());
assertEquals(
- Arrays.asList(new StreamsGroupMemberMetadataValue.KeyValue().setKey("assignment").setValue("config")),
+ mkMap(mkEntry("assignment", "config")),
member.assignmentConfigs()
);
}
@@ -266,19 +271,19 @@ public class StreamsGroupMemberTest {
.setMemberEpoch(10)
.setPreviousMemberEpoch(9)
.setState((byte) 1)
- .setActiveTasks(Arrays.asList(new StreamsGroupCurrentMemberAssignmentValue.TaskIds()
+ .setActiveTasks(Collections.singletonList(new TaskIds()
.setSubtopology(subtopologyId1)
.setPartitions(Arrays.asList(1, 2, 3)))
)
- .setStandbyTasks(Arrays.asList(new StreamsGroupCurrentMemberAssignmentValue.TaskIds()
+ .setStandbyTasks(Collections.singletonList(new TaskIds()
.setSubtopology(subtopologyId2)
.setPartitions(Arrays.asList(6, 5, 4)))
)
- .setWarmupTasks(Arrays.asList(new StreamsGroupCurrentMemberAssignmentValue.TaskIds()
+ .setWarmupTasks(Collections.singletonList(new TaskIds()
.setSubtopology(subtopologyId1)
.setPartitions(Arrays.asList(7, 8, 9)))
)
- .setActiveTasksPendingRevocation(Arrays.asList(new StreamsGroupCurrentMemberAssignmentValue.TaskIds()
+ .setActiveTasksPendingRevocation(Collections.singletonList(new TaskIds()
.setSubtopology(subtopologyId2)
.setPartitions(Arrays.asList(2, 3, 1)))
);
@@ -290,9 +295,9 @@ public class StreamsGroupMemberTest {
assertEquals(10, member.memberEpoch());
assertEquals(9, member.previousMemberEpoch());
assertEquals(MemberState.STABLE, member.state());
- assertEquals(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)), member.assignedActiveTasks());
- assertEquals(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)), member.assignedStandbyTasks());
- assertEquals(mkStreamsAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)), member.assignedWarmupTasks());
- assertEquals(mkStreamsAssignment(mkTaskAssignment(subtopologyId2, 2, 3, 1)), member.activeTasksPendingRevocation());
+ assertEquals(mkAssignment(mkTaskAssignment(subtopologyId1, 1, 2, 3)), member.assignedActiveTasks());
+ assertEquals(mkAssignment(mkTaskAssignment(subtopologyId2, 6, 5, 4)), member.assignedStandbyTasks());
+ assertEquals(mkAssignment(mkTaskAssignment(subtopologyId1, 7, 8, 9)), member.assignedWarmupTasks());
+ assertEquals(mkAssignment(mkTaskAssignment(subtopologyId2, 2, 3, 1)), member.activeTasksPendingRevocation());
}
}
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 d2f2f4fd22d..4dd727b8050 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
@@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import java.util.Collections;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
@@ -38,9 +39,8 @@ import java.util.OptionalLong;
import static java.util.Collections.emptyMap;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkStreamsAssignment;
-import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTaskAssignment;
-import static org.apache.kafka.server.immutable.ImmutableSet.singleton;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTaskAssignment;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -164,9 +164,9 @@ public class StreamsGroupTest {
member = new StreamsGroupMember.Builder("member")
.setMemberEpoch(10)
- .setAssignedActiveTasks(mkStreamsAssignment(
+ .setAssignedActiveTasks(mkAssignment(
mkTaskAssignment(fooSubtopology, 1, 2, 3)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(barSubtopology, 4, 5, 6)))
.build();
@@ -184,9 +184,9 @@ public class StreamsGroupTest {
member = new StreamsGroupMember.Builder(member)
.setMemberEpoch(11)
- .setAssignedActiveTasks(mkStreamsAssignment(
+ .setAssignedActiveTasks(mkAssignment(
mkTaskAssignment(barSubtopology, 1, 2, 3)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(zarSubtopology, 4, 5, 6)))
.build();
@@ -215,7 +215,7 @@ public class StreamsGroupTest {
.setAssignedActiveTasks(emptyMap())
.setAssignedStandbyTasks(emptyMap())
.setAssignedWarmupTasks(emptyMap())
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(fooSubtopologyId, 1)))
.build();
@@ -225,7 +225,7 @@ public class StreamsGroupTest {
member = new StreamsGroupMember.Builder(member)
.setMemberEpoch(11)
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
.setAssignedStandbyTasks(emptyMap())
.setAssignedWarmupTasks(emptyMap())
.setActiveTasksPendingRevocation(emptyMap())
@@ -243,7 +243,7 @@ public class StreamsGroupTest {
StreamsGroupMember m1 = new StreamsGroupMember.Builder("m1")
.setMemberEpoch(10)
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
.setAssignedStandbyTasks(emptyMap())
.setAssignedWarmupTasks(emptyMap())
.build();
@@ -252,7 +252,7 @@ public class StreamsGroupTest {
StreamsGroupMember m2 = new StreamsGroupMember.Builder("m2")
.setMemberEpoch(10)
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
.setAssignedStandbyTasks(emptyMap())
.setAssignedWarmupTasks(emptyMap())
.build();
@@ -269,20 +269,20 @@ public class StreamsGroupTest {
// Removing should fail because there is no epoch set.
assertThrows(IllegalStateException.class, () -> streamsGroup.removeActiveTaskEpochs(
- mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
+ mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
10
));
StreamsGroupMember m1 = new StreamsGroupMember.Builder("m1")
.setMemberEpoch(10)
- .setAssignedActiveTasks(mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
+ .setAssignedActiveTasks(mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)))
.build();
streamsGroup.updateMember(m1);
// Removing should fail because the expected epoch is incorrect.
assertThrows(IllegalStateException.class, () -> streamsGroup.removeActiveTaskEpochs(
- mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
+ mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
11
));
}
@@ -293,7 +293,7 @@ public class StreamsGroupTest {
StreamsGroup streamsGroup = createStreamsGroup("foo");
streamsGroup.addTaskEpochs(
- mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
+ mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
emptyMap(),
emptyMap(),
10
@@ -302,10 +302,10 @@ public class StreamsGroupTest {
// Changing the epoch should fail because the owner of the partition
// should remove it first.
assertThrows(IllegalStateException.class, () -> streamsGroup.addTaskEpochs(
- mkStreamsAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
- emptyMap(),
- emptyMap(),
- 11
+ mkAssignment(mkTaskAssignment(fooSubtopologyId, 1)),
+ emptyMap(),
+ emptyMap(),
+ 11
));
}
@@ -320,9 +320,9 @@ public class StreamsGroupTest {
member = new StreamsGroupMember.Builder("member")
.setMemberEpoch(10)
- .setAssignedActiveTasks(mkStreamsAssignment(
+ .setAssignedActiveTasks(mkAssignment(
mkTaskAssignment(fooSubtopology, 1, 2, 3)))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(barSubtopology, 4, 5, 6)))
.build();
@@ -362,7 +362,7 @@ public class StreamsGroupTest {
StreamsGroup streamsGroup = createStreamsGroup("foo");
streamsGroup.updateTargetAssignment(memberId1,
new Assignment(
- mkStreamsAssignment(
+ mkAssignment(
mkTaskAssignment(fooSubtopology, 1, 2, 3),
mkTaskAssignment(zarSubtopology, 7, 8, 9)
),
@@ -373,10 +373,10 @@ public class StreamsGroupTest {
StreamsGroupMember member1 = new StreamsGroupMember.Builder(memberId1)
.setMemberEpoch(10)
.setState(MemberState.UNRELEASED_TASKS)
- .setAssignedActiveTasks(mkStreamsAssignment(
+ .setAssignedActiveTasks(mkAssignment(
mkTaskAssignment(fooSubtopology, 1, 2, 3)
))
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(barSubtopology, 4, 5, 6)
))
.build();
@@ -386,7 +386,7 @@ public class StreamsGroupTest {
StreamsGroupMember member2 = new StreamsGroupMember.Builder(memberId2)
.setMemberEpoch(10)
- .setActiveTasksPendingRevocation(mkStreamsAssignment(
+ .setActiveTasksPendingRevocation(mkAssignment(
mkTaskAssignment(zarSubtopology, 7)))
.build();
streamsGroup.updateMember(member2);
@@ -478,7 +478,7 @@ public class StreamsGroupTest {
// Initial assignment for member1
Assignment initialAssignment = new Assignment(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId, 0)),
+ mkAssignment(mkTaskAssignment(subtopologyId, 0)),
emptyMap(),
emptyMap()
);
@@ -494,7 +494,7 @@ public class StreamsGroupTest {
// New assignment for member1
Assignment newAssignment = new Assignment(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId, 1)),
+ mkAssignment(mkTaskAssignment(subtopologyId, 1)),
emptyMap(),
emptyMap()
);
@@ -510,7 +510,7 @@ public class StreamsGroupTest {
// New assignment for member2 to add partition 1
Assignment newAssignment2 = new Assignment(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId, 1)),
+ mkAssignment(mkTaskAssignment(subtopologyId, 1)),
emptyMap(),
emptyMap()
);
@@ -526,7 +526,7 @@ public class StreamsGroupTest {
// New assignment for member1 to revoke partition 1 and assign partition 0
Assignment newAssignment1 = new Assignment(
- mkStreamsAssignment(mkTaskAssignment(subtopologyId, 0)),
+ mkAssignment(mkTaskAssignment(subtopologyId, 0)),
emptyMap(),
emptyMap()
);
@@ -717,14 +717,14 @@ public class StreamsGroupTest {
);
StreamsGroup group = new StreamsGroup(snapshotRegistry, "group-foo", metricsShard);
snapshotRegistry.getOrCreateSnapshot(0);
- assertTrue(group.isInStates(singleton("empty"), 0));
- assertFalse(group.isInStates(singleton("Empty"), 0));
+ assertTrue(group.isInStates(Collections.singleton("empty"), 0));
+ assertFalse(group.isInStates(Collections.singleton("Empty"), 0));
group.updateMember(new StreamsGroupMember.Builder("member1")
.build());
snapshotRegistry.getOrCreateSnapshot(1);
- assertTrue(group.isInStates(singleton("empty"), 0));
- assertTrue(group.isInStates(singleton("stable"), 1));
- assertFalse(group.isInStates(singleton("empty"), 1));
+ assertTrue(group.isInStates(Collections.singleton("empty"), 0));
+ assertTrue(group.isInStates(Collections.singleton("stable"), 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/TargetAssignmentBuilderTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
new file mode 100644
index 00000000000..ec82b8e99a7
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
@@ -0,0 +1,821 @@
+/*
+ * 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.Uuid;
+import org.apache.kafka.coordinator.group.CoordinatorRecordHelpers;
+import org.apache.kafka.coordinator.group.MetadataImageBuilder;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
+import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
+import org.apache.kafka.coordinator.group.taskassignor.AssignmentMemberSpec;
+import org.apache.kafka.coordinator.group.taskassignor.GroupAssignment;
+import org.apache.kafka.coordinator.group.taskassignor.GroupSpecImpl;
+import org.apache.kafka.coordinator.group.taskassignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.taskassignor.TaskAssignor;
+import org.apache.kafka.image.TopicsImage;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+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.apache.kafka.coordinator.group.Assertions.assertUnorderedListEquals;
+import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpers.newTargetAssignmentEpochRecord;
+import static org.apache.kafka.coordinator.group.CoordinatorRecordHelpersTest.mkMapOfPartitionRacks;
+import static org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.createAssignmentMemberSpec;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkAssignment;
+import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTaskAssignment;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TargetAssignmentBuilderTest {
+
+ public static class TargetAssignmentBuilderTestContext {
+
+ private final String groupId;
+ private final int groupEpoch;
+ private final TaskAssignor assignor = mock(TaskAssignor.class);
+ private final StreamsTopology topology = new StreamsTopology(new byte[0], new HashMap<>());
+ private final Map<String, StreamsGroupMember> members = new HashMap<>();
+ private final Map<String, org.apache.kafka.coordinator.group.streams.TopicMetadata> subscriptionMetadata = new HashMap<>();
+ private final Map<String, StreamsGroupMember> updatedMembers = new HashMap<>();
+ private final Map<String, org.apache.kafka.coordinator.group.streams.Assignment> targetAssignment = new HashMap<>();
+ private final Map<String, MemberAssignment> memberAssignments = new HashMap<>();
+ private final Map<String, String> staticMembers = new HashMap<>();
+ private MetadataImageBuilder topicsImageBuilder = new MetadataImageBuilder();
+
+ public TargetAssignmentBuilderTestContext(
+ String groupId,
+ int groupEpoch
+ ) {
+ this.groupId = groupId;
+ this.groupEpoch = groupEpoch;
+ }
+
+ public void addGroupMember(
+ String memberId,
+ Map<String, Set<Integer>> targetTasks
+ ) {
+ addGroupMember(memberId, null, targetTasks);
+ }
+
+ private void addGroupMember(
+ String memberId,
+ String instanceId,
+ Map<String, Set<Integer>> targetTasks
+ ) {
+ StreamsGroupMember.Builder memberBuilder = new StreamsGroupMember.Builder(memberId);
+ memberBuilder.setProcessId("processId");
+ memberBuilder.setClientTags(Collections.emptyMap());
+ memberBuilder.setAssignmentConfigs(Collections.emptyMap());
+ memberBuilder.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090));
+
+ if (instanceId != null) {
+ memberBuilder.setInstanceId(instanceId);
+ staticMembers.put(instanceId, memberId);
+ }
+ members.put(memberId, memberBuilder.build());
+ targetAssignment.put(memberId, new org.apache.kafka.coordinator.group.streams.Assignment(targetTasks));
+ }
+
+ public String addSubtopologyWithSingleSourceTopic(
+ String topicName,
+ int numTasks,
+ Map<Integer, Set<String>> partitionRacks
+ ) {
+ String subtopologyId = Uuid.randomUuid().toString();
+ Uuid topicId = Uuid.randomUuid();
+ subscriptionMetadata.put(topicName, new org.apache.kafka.coordinator.group.streams.TopicMetadata(
+ topicId,
+ topicName,
+ numTasks,
+ partitionRacks
+ ));
+ topicsImageBuilder = topicsImageBuilder.addTopic(topicId, topicName, numTasks);
+ topology.subtopologies().put(subtopologyId, new StreamsGroupTopologyValue.Subtopology()
+ .setSubtopology(subtopologyId)
+ .setSourceTopics(Collections.singletonList(topicId.toString())));
+
+ return subtopologyId;
+ }
+
+ public void updateMemberSubscription(
+ String memberId
+ ) {
+ updateMemberSubscription(
+ memberId,
+ Optional.empty(),
+ Optional.empty()
+ );
+ }
+
+ public void updateMemberSubscription(
+ String memberId,
+ Optional<String> instanceId,
+ Optional<String> rackId
+ ) {
+ StreamsGroupMember existingMember = members.get(memberId);
+ StreamsGroupMember.Builder builder;
+ if (existingMember != null) {
+ builder = new StreamsGroupMember.Builder(existingMember);
+ } else {
+ builder = new StreamsGroupMember.Builder(memberId);
+ builder.setProcessId("processId");
+ builder.setClientTags(Collections.emptyMap());
+ builder.setAssignmentConfigs(Collections.emptyMap());
+ builder.setHostInfo(new StreamsGroupMemberMetadataValue.HostInfo().setHost("host").setPort(9090));
+ }
+ updatedMembers.put(memberId, builder
+ .maybeUpdateInstanceId(instanceId)
+ .maybeUpdateRackId(rackId)
+ .build());
+ }
+
+ public void removeMemberSubscription(
+ String memberId
+ ) {
+ this.updatedMembers.put(memberId, null);
+ }
+
+ public void prepareMemberAssignment(
+ String memberId,
+ Map<String, Set<Integer>> activeTasks
+ ) {
+ memberAssignments.put(memberId, new MemberAssignment(activeTasks, Collections.emptyMap(), Collections.emptyMap()));
+ }
+
+ public org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult build() {
+ TopicsImage topicsImage = topicsImageBuilder.build().topics();
+ // Prepare expected member specs.
+ Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+
+ // All the existing members are prepared.
+ members.forEach((memberId, member) ->
+ memberSpecs.put(memberId, createAssignmentMemberSpec(
+ member,
+ targetAssignment.getOrDefault(memberId, org.apache.kafka.coordinator.group.streams.Assignment.EMPTY)
+ )
+ ));
+
+ // All the updated are added and all the deleted
+ // members are removed.
+ updatedMembers.forEach((memberId, updatedMemberOrNull) -> {
+ if (updatedMemberOrNull == null) {
+ memberSpecs.remove(memberId);
+ } else {
+ org.apache.kafka.coordinator.group.streams.Assignment assignment = targetAssignment.getOrDefault(memberId,
+ org.apache.kafka.coordinator.group.streams.Assignment.EMPTY);
+
+ // A new static member joins and needs to replace an existing departed one.
+ if (updatedMemberOrNull.instanceId() != null) {
+ String previousMemberId = staticMembers.get(updatedMemberOrNull.instanceId());
+ if (previousMemberId != null && !previousMemberId.equals(memberId)) {
+ assignment = targetAssignment.getOrDefault(previousMemberId,
+ org.apache.kafka.coordinator.group.streams.Assignment.EMPTY);
+ }
+ }
+
+ memberSpecs.put(memberId, createAssignmentMemberSpec(
+ updatedMemberOrNull,
+ assignment
+ ));
+ }
+ });
+
+ // Prepare the expected subscription topic metadata.
+ TopologyMetadata topologyMetadata = new TopologyMetadata(subscriptionMetadata, topology);
+
+ // Prepare the member assignments per topic partition.
+ Map<String, Map<Integer, String>> invertedTargetAssignment = TaskAssignmentTestUtil.invertedTargetAssignment(memberSpecs);
+
+ // Prepare the expected assignment spec.
+ GroupSpecImpl groupSpec = new GroupSpecImpl(memberSpecs, new ArrayList<>(topology.subtopologies().keySet()));
+
+ // We use `any` here to always return an assignment but use `verify` later on
+ // to ensure that the input was correct.
+ when(assignor.assign(any(), any()))
+ .thenReturn(new GroupAssignment(memberAssignments));
+
+ // Create and populate the assignment builder.
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder builder = new org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder(
+ groupId, groupEpoch, assignor, topology)
+ .withMembers(members)
+ .withStaticMembers(staticMembers)
+ .withSubscriptionMetadata(subscriptionMetadata)
+ .withTargetAssignment(targetAssignment)
+ .withInvertedTargetAssignment(invertedTargetAssignment)
+ .withTopicsImage(topicsImage);
+
+ // Add the updated members or delete the deleted members.
+ updatedMembers.forEach((memberId, updatedMemberOrNull) -> {
+ if (updatedMemberOrNull != null) {
+ builder.addOrUpdateMember(memberId, updatedMemberOrNull);
+ } else {
+ builder.removeMember(memberId);
+ }
+ });
+
+ // Execute the builder.
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = builder.build();
+
+ // Verify that the assignor was called once with the expected
+ // assignment spec.
+ verify(assignor, times(1))
+ .assign(groupSpec, topologyMetadata);
+
+ return result;
+ }
+ }
+
+ @Test
+ public void testCreateAssignmentMemberSpec() {
+ Uuid fooTopicId = Uuid.randomUuid();
+ Uuid barTopicId = Uuid.randomUuid();
+ String fooSubtopologyId = Uuid.randomUuid().toString();
+ String barSubtopologyId = Uuid.randomUuid().toString();
+ TopicsImage topicsImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, "foo", 5)
+ .addTopic(barTopicId, "bar", 5)
+ .build()
+ .topics();
+
+ final Map<String, String> clientTags = mkMap(mkEntry("tag1", "value1"), mkEntry("tag2", "value2"));
+ final Map<String, String> assignmentConfigs = mkMap(mkEntry("conf1", "value1"), mkEntry("conf2", "value2"));
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
+ .setRackId("rackId")
+ .setInstanceId("instanceId")
+ .setProcessId("processId")
+ .setAssignor("assignor")
+ .setClientTags(clientTags)
+ .setAssignmentConfigs(assignmentConfigs)
+ .build();
+
+ org.apache.kafka.coordinator.group.streams.Assignment assignment = new org.apache.kafka.coordinator.group.streams.Assignment(
+ mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ AssignmentMemberSpec assignmentMemberSpec = createAssignmentMemberSpec(
+ member,
+ assignment
+ );
+
+ assertEquals(new AssignmentMemberSpec(
+ Optional.of("instanceId"),
+ Optional.of("rackId"),
+ assignment.activeTasks(),
+ assignment.standbyTasks(),
+ assignment.warmupTasks(),
+ "processId",
+ clientTags,
+ assignmentConfigs,
+ Collections.emptyMap()
+ ), assignmentMemberSpec);
+ }
+
+ @Test
+ public void testEmpty() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+ assertEquals(Collections.singletonList(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ )), result.records());
+ assertEquals(Collections.emptyMap(), result.targetAssignment());
+ }
+
+ @Test
+ public void testAssignmentHasNotChanged() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(Collections.singletonList(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ )), result.records());
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testAssignmentSwapped() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(3, result.records().size());
+
+ assertUnorderedListEquals(Arrays.asList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 2));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(2));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ )));
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testNewMember() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ context.updateMemberSubscription("member-3");
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.prepareMemberAssignment("member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(4, result.records().size());
+
+ assertUnorderedListEquals(Arrays.asList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 3));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(3));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ )));
+ expectedAssignment.put("member-3", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testUpdateMember() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.addGroupMember("member-3", mkAssignment(
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ context.updateMemberSubscription(
+ "member-3",
+ Optional.of("instance-id-3"),
+ Optional.of("rack-0")
+ );
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.prepareMemberAssignment("member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(4, result.records().size());
+
+ assertUnorderedListEquals(Arrays.asList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 3));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(3));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ )));
+ expectedAssignment.put("member-3", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testPartialAssignmentUpdate() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, mkMapOfPartitionRacks(6));
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, mkMapOfPartitionRacks(6));
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.addGroupMember("member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4, 5),
+ mkTaskAssignment(barSubtopologyId, 3, 4, 5)
+ ));
+
+ context.prepareMemberAssignment("member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 6),
+ mkTaskAssignment(barSubtopologyId, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(3, result.records().size());
+
+ // Member 1 has no record because its assignment did not change.
+ assertUnorderedListEquals(Arrays.asList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4, 5),
+ mkTaskAssignment(barSubtopologyId, 3, 4, 5)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 6),
+ mkTaskAssignment(barSubtopologyId, 6)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 2));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(2));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4, 5),
+ mkTaskAssignment(barSubtopologyId, 3, 4, 5)
+ )));
+ expectedAssignment.put("member-3", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 6),
+ mkTaskAssignment(barSubtopologyId, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testDeleteMember() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.addGroupMember("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.addGroupMember("member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ context.removeMemberSubscription("member-3");
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ));
+
+ org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(3, result.records().size());
+
+ assertUnorderedListEquals(Arrays.asList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ ), Collections.emptyMap(), Collections.emptyMap()),
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 2));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(2));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2, 3),
+ mkTaskAssignment(barSubtopologyId, 1, 2, 3)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 4, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 4, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+
+ @Test
+ public void testReplaceStaticMember() {
+ TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
+ "my-group",
+ 20
+ );
+
+ String fooSubtopologyId = context.addSubtopologyWithSingleSourceTopic("foo", 6, Collections.emptyMap());
+ String barSubtopologyId = context.addSubtopologyWithSingleSourceTopic("bar", 6, Collections.emptyMap());
+
+ context.addGroupMember("member-1", "instance-member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.addGroupMember("member-2", "instance-member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.addGroupMember("member-3", "instance-member-3", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ // Static member 3 leaves
+ context.removeMemberSubscription("member-3");
+
+ // Another static member joins with the same instance id as the departed one
+ context.updateMemberSubscription("member-3-a", Optional.of("instance-member-3"),
+ Optional.empty());
+
+ context.prepareMemberAssignment("member-1", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ ));
+
+ context.prepareMemberAssignment("member-2", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ ));
+
+ context.prepareMemberAssignment("member-3-a", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ));
+
+ TargetAssignmentBuilder.TargetAssignmentResult result = context.build();
+
+ assertEquals(2, result.records().size());
+
+ assertUnorderedListEquals(Collections.singletonList(
+ CoordinatorRecordHelpers.newStreamsTargetAssignmentRecord("my-group", "member-3-a", mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ ), Collections.emptyMap(), Collections.emptyMap())
+ ), result.records().subList(0, 1));
+
+ assertEquals(newTargetAssignmentEpochRecord(
+ "my-group",
+ 20
+ ), result.records().get(1));
+
+ Map<String, org.apache.kafka.coordinator.group.streams.Assignment> expectedAssignment = new HashMap<>();
+ expectedAssignment.put("member-1", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 1, 2),
+ mkTaskAssignment(barSubtopologyId, 1, 2)
+ )));
+ expectedAssignment.put("member-2", new org.apache.kafka.coordinator.group.streams.Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 3, 4),
+ mkTaskAssignment(barSubtopologyId, 3, 4)
+ )));
+
+ expectedAssignment.put("member-3-a", new Assignment(mkAssignment(
+ mkTaskAssignment(fooSubtopologyId, 5, 6),
+ mkTaskAssignment(barSubtopologyId, 5, 6)
+ )));
+
+ assertEquals(expectedAssignment, result.targetAssignment());
+ }
+}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TaskAssignmentTestUtil.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TaskAssignmentTestUtil.java
new file mode 100644
index 00000000000..d53398238f7
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TaskAssignmentTestUtil.java
@@ -0,0 +1,124 @@
+/*
+ * 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.coordinator.group.taskassignor.AssignmentMemberSpec;
+import org.apache.kafka.coordinator.group.taskassignor.GroupAssignment;
+
+import java.util.AbstractMap;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TaskAssignmentTestUtil {
+
+ public static Map.Entry<String, Set<Integer>> mkTaskAssignment(
+ String subtopologyId,
+ Integer... tasks
+ ) {
+ return new AbstractMap.SimpleEntry<>(
+ subtopologyId,
+ new HashSet<>(Arrays.asList(tasks))
+ );
+ }
+
+ public static Map.Entry<String, Set<Integer>> mkSortedTaskAssignment(
+ String subtopologyId,
+ Integer... tasks
+ ) {
+ return new AbstractMap.SimpleEntry<>(
+ subtopologyId,
+ new TreeSet<>(Arrays.asList(tasks))
+ );
+ }
+
+ @SafeVarargs
+ public static Map<String, Set<Integer>> mkAssignment(Map.Entry<String, Set<Integer>>... entries) {
+ Map<String, Set<Integer>> assignment = new HashMap<>();
+ for (Map.Entry<String, Set<Integer>> entry : entries) {
+ assignment.put(entry.getKey(), entry.getValue());
+ }
+ return assignment;
+ }
+
+ @SafeVarargs
+ public static Map<String, Set<Integer>> mkSortedAssignment(Map.Entry<String, Set<Integer>>... entries) {
+ Map<String, Set<Integer>> assignment = new LinkedHashMap<>();
+ for (Map.Entry<String, Set<Integer>> entry : entries) {
+ assignment.put(entry.getKey(), entry.getValue());
+ }
+ return assignment;
+ }
+
+ @SafeVarargs
+ public static Map<String, Set<Integer>> mkStreamsAssignment(Map.Entry<String, Set<Integer>>... entries) {
+ Map<String, Set<Integer>> assignment = new HashMap<>();
+ for (Map.Entry<String, Set<Integer>> entry : entries) {
+ assignment.put(entry.getKey(), entry.getValue());
+ }
+ return assignment;
+ }
+
+ /**
+ * Verifies that the expected assignment is equal to the computed assignment for every member in the group.
+ */
+ public static void assertAssignment(
+ Map<String, Map<String, Set<Integer>>> expectedAssignment,
+ GroupAssignment computedGroupAssignment
+ ) {
+ assertEquals(expectedAssignment.size(), computedGroupAssignment.members().size());
+ computedGroupAssignment.members().forEach((memberId, memberAssignment) -> {
+ Map<String, Set<Integer>> computedAssignmentForMember = memberAssignment.activeTasks();
+ assertEquals(expectedAssignment.get(memberId), computedAssignmentForMember);
+ });
+ }
+
+ /**
+ * Generate a reverse look up map of partition to member target assignments from the given member spec.
+ *
+ * @param memberSpec A map where the key is the member Id and the value is an AssignmentMemberSpec object containing the member's
+ * partition assignments.
+ * @return Map of topic partition to member assignments.
+ */
+ public static Map<String, Map<Integer, String>> invertedTargetAssignment(
+ Map<String, AssignmentMemberSpec> memberSpec
+ ) {
+ Map<String, Map<Integer, String>> invertedTargetAssignment = new HashMap<>();
+ for (Map.Entry<String, AssignmentMemberSpec> memberEntry : memberSpec.entrySet()) {
+ String memberId = memberEntry.getKey();
+ Map<String, Set<Integer>> topicsAndTasks = memberEntry.getValue().activeTasks();
+
+ for (Map.Entry<String, Set<Integer>> topicEntry : topicsAndTasks.entrySet()) {
+ String subtopologyId = topicEntry.getKey();
+ Set<Integer> tasks = topicEntry.getValue();
+
+ Map<Integer, String> partitionMap = invertedTargetAssignment.computeIfAbsent(subtopologyId, k -> new HashMap<>());
+
+ for (Integer partitionId : tasks) {
+ partitionMap.put(partitionId, memberId);
+ }
+ }
+ }
+ return invertedTargetAssignment;
+ }
+}