From e8d36070e0b4f799bd10b6388fa1ef85e8d51328 Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Thu, 23 Mar 2023 13:02:31 -0700
Subject: [PATCH 01/44] Server Side Sticky Range Assignor full implementation
---
checkstyle/suppressions.xml | 8 +
.../group/assignor/AssignmentMemberSpec.java | 29 +-
.../group/assignor/GroupAssignment.java | 4 +
.../group/assignor/MemberAssignment.java | 33 +-
.../ServerSideStickyRangeAssignor.java | 268 +++++++++++
.../ServerSideStickyRangeAssignorTest.java | 428 ++++++++++++++++++
6 files changed, 741 insertions(+), 29 deletions(-)
create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 93ae7c30d0f3e..3ff095f3760c9 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -320,6 +320,14 @@
+
+
+
+
+
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
index 2db33ddc74158..fe3d7c967e537 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
@@ -16,11 +16,13 @@
*/
package org.apache.kafka.coordinator.group.assignor;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
-import java.util.Collection;
-import java.util.Objects;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.Optional;
+import java.util.Objects;
/**
* The assignment specification for a consumer group member.
@@ -37,29 +39,28 @@ public class AssignmentMemberSpec {
final Optional rackId;
/**
- * The topics that the member is subscribed to.
+ * The topicIds of topics that the member is subscribed to.
*/
- final Collection subscribedTopics;
+ final List subscribedTopics;
/**
- * The current target partitions of the member.
+ * Maps the partitions assigned for this member per topicId
*/
- final Collection targetPartitions;
+ final Map> currentAssignmentPerTopic;
public AssignmentMemberSpec(
Optional instanceId,
Optional rackId,
- Collection subscribedTopics,
- Collection targetPartitions
+ List subscribedTopics,
+ Map> currentAssignmentPerTopic
) {
Objects.requireNonNull(instanceId);
Objects.requireNonNull(rackId);
Objects.requireNonNull(subscribedTopics);
- Objects.requireNonNull(targetPartitions);
this.instanceId = instanceId;
this.rackId = rackId;
this.subscribedTopics = subscribedTopics;
- this.targetPartitions = targetPartitions;
+ this.currentAssignmentPerTopic = currentAssignmentPerTopic;
}
@Override
@@ -72,7 +73,7 @@ public boolean equals(Object o) {
if (!instanceId.equals(that.instanceId)) return false;
if (!rackId.equals(that.rackId)) return false;
if (!subscribedTopics.equals(that.subscribedTopics)) return false;
- return targetPartitions.equals(that.targetPartitions);
+ return currentAssignmentPerTopic.equals(that.currentAssignmentPerTopic);
}
@Override
@@ -80,7 +81,7 @@ public int hashCode() {
int result = instanceId.hashCode();
result = 31 * result + rackId.hashCode();
result = 31 * result + subscribedTopics.hashCode();
- result = 31 * result + targetPartitions.hashCode();
+ result = 31 * result + currentAssignmentPerTopic.hashCode();
return result;
}
@@ -89,7 +90,7 @@ public String toString() {
return "AssignmentMemberSpec(instanceId=" + instanceId +
", rackId=" + rackId +
", subscribedTopics=" + subscribedTopics +
- ", targetPartitions=" + targetPartitions +
+ ", currentAssignment=" + currentAssignmentPerTopic +
')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
index 5c0199aaec5a5..6256b7453a835 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
@@ -35,6 +35,10 @@ public GroupAssignment(
this.members = members;
}
+ public Map getMembers() {
+ return this.members;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
index 86c235d718771..3f9dd7f8d4378 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
@@ -16,25 +16,28 @@
*/
package org.apache.kafka.coordinator.group.assignor;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
-import java.util.Collection;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+
/**
* The partition assignment for a consumer group member.
*/
public class MemberAssignment {
- /**
- * The target partitions assigned to this member.
- */
- final Collection targetPartitions;
-
- public MemberAssignment(
- Collection targetPartitions
- ) {
- Objects.requireNonNull(targetPartitions);
- this.targetPartitions = targetPartitions;
+
+ private final Map> assignmentPerTopic;
+
+
+ public MemberAssignment(Map> assignmentPerTopic) {
+ Objects.requireNonNull(assignmentPerTopic);
+ this.assignmentPerTopic = assignmentPerTopic;
+ }
+
+ public Map> getAssignmentPerTopic() {
+ return this.assignmentPerTopic;
}
@Override
@@ -44,16 +47,16 @@ public boolean equals(Object o) {
MemberAssignment that = (MemberAssignment) o;
- return targetPartitions.equals(that.targetPartitions);
+ return assignmentPerTopic.equals(that.assignmentPerTopic);
}
@Override
public int hashCode() {
- return targetPartitions.hashCode();
+ return assignmentPerTopic.hashCode();
}
@Override
public String toString() {
- return "MemberAssignment(targetPartitions=" + targetPartitions + ')';
+ return "MemberAssignment(targetPartitions=" + assignmentPerTopic + ')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
new file mode 100644
index 0000000000000..186f8af1f0220
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
@@ -0,0 +1,268 @@
+/*
+ * 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.
+ */
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range assignor and the sticky assignor.
+ * Properties are :-
+ *
+ * - 1) Each consumer must get at least one partition per topic that it is subscribed to whenever the number of consumers is
+ * less than or equal to the number of partitions for that topic. (Range)
+ * - 2) Partitions should be assigned to consumers in a way that facilitates join operations where required. (Range)
+ * This can only be done if the topics are co-partitioned in the first place
+ * Co-partitioned:-
+ * Two streams are co-partitioned if the following conditions are met:-
+ * ->The keys must have the same schemas
+ * ->The topics involved must have the same number of partitions
+ * - 3) Consumers should retain as much as their previous assignment as possible. (Sticky)
+ *
+ *
+ *
+ * The algorithm works mainly in 5 steps described below
+ *
+ * - 1) Get a map of the consumersPerTopic created using the member subscriptions.
+ * - 2) Get a list of consumers (potentiallyUnfilled) that have not met the minimum required quota for assignment AND
+ * get a list of sticky partitions that we want to retain in the new assignment.
+ * - 3) Add consumers from potentiallyUnfilled to Unfilled if they haven't met the total required quota = minQuota + (if necessary) extraPartition
+ * - 4) Get a list of available partitions by calculating the difference between total partitions and assigned sticky partitions
+ * - 5) Iterate through unfilled consumers and assign partitions from available partitions
+ *
+ *
+ *
+ */
+package org.apache.kafka.coordinator.group.assignor;
+
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.HashMap;
+import java.util.Collections;
+
+import static java.lang.Math.min;
+
+public class ServerSideStickyRangeAssignor implements PartitionAssignor {
+
+ public static final String RANGE_ASSIGNOR_NAME = "range-sticky";
+
+ @Override
+ public String name() {
+ return RANGE_ASSIGNOR_NAME;
+ }
+
+ protected static void putList(Map> map, K key, V value) {
+ List list = map.computeIfAbsent(key, k -> new ArrayList<>());
+ list.add(value);
+ }
+
+ protected static void putSet(Map> map, K key, V value) {
+ Set set = map.computeIfAbsent(key, k -> new HashSet<>());
+ set.add(value);
+ }
+
+ static class Pair {
+ private final T first;
+ private final U second;
+
+ public Pair(T first, U second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public T getFirst() {
+ return first;
+ }
+
+ public U getSecond() {
+ return second;
+ }
+
+ @Override
+ public String toString() {
+ return "(" + first + ", " + second + ")";
+ }
+ }
+
+ // Returns a map of the list of consumers per Topic (keyed by topicId)
+ private Map> consumersPerTopic(AssignmentSpec assignmentSpec) {
+ Map> mapTopicsToConsumers = new HashMap<>();
+ Map membersData = assignmentSpec.members;
+
+ for (Map.Entry memberEntry : membersData.entrySet()) {
+ String memberId = memberEntry.getKey();
+ AssignmentMemberSpec memberMetadata = memberEntry.getValue();
+ List topics = new ArrayList<>(memberMetadata.subscribedTopics);
+ for (Uuid topicId: topics) {
+ putList(mapTopicsToConsumers, topicId, memberId);
+ }
+ }
+ return mapTopicsToConsumers;
+ }
+
+ private Map> getAvailablePartitionsPerTopic(AssignmentSpec assignmentSpec, Map> assignedStickyPartitionsPerTopic) {
+ Map> availablePartitionsPerTopic = new HashMap<>();
+ Map topicsMetadata = assignmentSpec.topics;
+ // Iterate through the topics map provided in assignmentSpec
+ for (Map.Entry topicMetadataEntry : topicsMetadata.entrySet()) {
+ Uuid topicId = topicMetadataEntry.getKey();
+ availablePartitionsPerTopic.put(topicId, new ArrayList<>());
+ int numPartitions = topicsMetadata.get(topicId).numPartitions;
+ // since the loop iterates from 0 to n, the partitions will be in ascending order within the list of available partitions per topic
+ Set assignedStickyPartitionsForTopic = assignedStickyPartitionsPerTopic.get(topicId);
+ for (int i = 0; i < numPartitions; i++) {
+ if (assignedStickyPartitionsForTopic == null || !assignedStickyPartitionsForTopic.contains(i)) {
+ availablePartitionsPerTopic.get(topicId).add(i);
+ }
+ }
+ }
+ return availablePartitionsPerTopic;
+ }
+
+ @Override
+ public GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAssignorException {
+ Map>> membersWithNewAssignmentPerTopic = new HashMap<>();
+ // Step 1
+ Map> consumersPerTopic = consumersPerTopic(assignmentSpec);
+ // Step 2
+ Map>> unfilledConsumersPerTopic = new HashMap<>();
+ Map> assignedStickyPartitionsPerTopic = new HashMap<>();
+
+ for (Map.Entry> topicEntry : consumersPerTopic.entrySet()) {
+ Uuid topicId = topicEntry.getKey();
+ // For each topic we have a temporary list of consumers stored in potentiallyUnfilledConsumers.
+ // The list is populated with consumers that satisfy one of the two conditions :-
+ // 1) Consumers that have the minimum required number of partitions .i.e numPartitionsPerConsumer BUT they could be assigned an extra partition later on.
+ // In this case we add the consumer to the unfilled consumers map iff an extra partition needs to be assigned to it.
+ // 2) Consumers that don't have the minimum required partitions, so irrespective of whether they get an extra partition or not they get added to the unfilled map later.
+ List> potentiallyUnfilledConsumers = new ArrayList<>();
+ List consumersForTopic = topicEntry.getValue();
+
+ AssignmentTopicMetadata topicData = assignmentSpec.topics.get(topicId);
+ int numPartitionsForTopic = topicData.numPartitions;
+ int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
+ // Each consumer can get only one extra partition per topic after receiving the minimum quota = numPartitionsPerConsumer
+ int numConsumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
+
+ for (String memberId: consumersForTopic) {
+ // Size of the older assignment, this will be 0 when assign is called for the first time.
+ // The older assignment is required when a reassignment occurs to ensure stickiness.
+ int currentAssignmentSize = 0;
+ List currentAssignmentListForTopic = new ArrayList<>();
+ // Convert the set to a list first and sort the partitions in numeric order since we want the same partition numbers from each topic
+ // to go to the same consumer in case of co-partitioned topics.
+ Set currentAssignmentSetForTopic = assignmentSpec.members.get(memberId).currentAssignmentPerTopic.get(topicId);
+ // We want to make sure that the currentAssignment is not null to avoid a null pointer exception
+ if (currentAssignmentSetForTopic != null) {
+ currentAssignmentListForTopic = new ArrayList<>(currentAssignmentSetForTopic);
+ currentAssignmentSize = currentAssignmentListForTopic.size();
+ }
+ // Initially, minRequiredQuota is the minimum number of partitions that each consumer should get.
+ // The value is at least 1 unless numConsumers subscribed is greater than numPartitions. In such cases, all consumers get assigned "extra partitions".
+ int minRequiredQuota = numPartitionsPerConsumer;
+ // We need to make sure that the partitions we're keeping are still part of the topic metadata.
+ // Ex:- In case the number of partitions is reduced for a topic, the sticky partitions must be part of the new set of partitions available.
+ for (int i = 0; i < currentAssignmentListForTopic.size() && currentAssignmentSize > 0; i++) {
+ // numPartitionsForTopic - 1 is the max possible partition number
+ if (currentAssignmentListForTopic.get(i) >= numPartitionsForTopic) {
+ currentAssignmentSize--;
+ }
+ }
+ // If there are previously assigned partitions present, we want to retain
+ if (currentAssignmentSize > 0) {
+ // We either need to retain currentSize number of partitions when currentSize < required OR required number of partitions otherwise.
+ int retainedPartitionsCount = min(currentAssignmentSize, minRequiredQuota);
+ for (int i = 0; i < retainedPartitionsCount; i++) {
+ Collections.sort(currentAssignmentListForTopic);
+ putSet(assignedStickyPartitionsPerTopic, topicId, currentAssignmentListForTopic.get(i));
+ membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).add(currentAssignmentListForTopic.get(i));
+ }
+ }
+ // Number of partitions left to reach the minRequiredQuota
+ int remaining = minRequiredQuota - currentAssignmentSize;
+
+ // There are 3 cases w.r.t value of remaining
+ // 1) remaining < 0 this means that the consumer has more than the min required amount.
+ // It could have an extra partition, so we check for that.
+ if (remaining < 0 && numConsumersWithExtraPartition > 0) {
+ // In order to remain as sticky as possible, since the order of members can be different, we want the consumers that already had extra
+ // partitions to retain them if it's still required, instead of assigning the extras to the first few consumers directly.
+ // Ex:- If two consumers out of 3 are supposed to get an extra partition and currently 1 of them already has the extra, we want this consumer
+ // to retain it first and later if we have partitions left they will be assigned to the first few consumers and the unfilled map is updated
+ numConsumersWithExtraPartition--;
+ // Since we already added the minimumRequiredQuota of partitions in the previous step (until minReq - 1), we just need to
+ // add the extra partition which will be present at the index right after min quota is satisfied.
+ putSet(assignedStickyPartitionsPerTopic, topicId, currentAssignmentListForTopic.get(minRequiredQuota));
+ membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).add(currentAssignmentListForTopic.get(minRequiredQuota));
+ } else {
+ // 3) If remaining = 0 it has min req partitions but there is scope for getting an extra partition later on, so it is a potentialUnfilledConsumer.
+ // 4) If remaining > 0 it doesn't even have the min required partitions, and it definitely is unfilled so it should be added to potentialUnfilledConsumers.
+ Pair newPair = new Pair<>(memberId, remaining);
+ potentiallyUnfilledConsumers.add(newPair);
+ }
+ }
+
+ // Step 3
+ // Iterate through potentially unfilled consumers and if remaining > 0 after considering the extra partitions assignment, add to the unfilled list.
+ for (Pair pair : potentiallyUnfilledConsumers) {
+ String memberId = pair.getFirst();
+ Integer remaining = pair.getSecond();
+ if (numConsumersWithExtraPartition > 0) {
+ remaining++;
+ numConsumersWithExtraPartition--;
+ }
+ if (remaining > 0) {
+ Pair newPair = new Pair<>(memberId, remaining);
+ putList(unfilledConsumersPerTopic, topicId, newPair);
+ }
+ }
+ }
+
+ // Step 4
+ // Find the difference between the total partitions per topic and the already assigned sticky partitions for the topic to get available partitions.
+ Map> availablePartitionsPerTopic = getAvailablePartitionsPerTopic(assignmentSpec, assignedStickyPartitionsPerTopic);
+
+ // Step 5
+ // Iterate through the unfilled consumers list and assign remaining number of partitions from the availablePartitions list
+ for (Map.Entry>> unfilledEntry : unfilledConsumersPerTopic.entrySet()) {
+ Uuid topicId = unfilledEntry.getKey();
+ List> unfilledConsumersForTopic = unfilledEntry.getValue();
+
+ for (Pair stringIntegerPair : unfilledConsumersForTopic) {
+ String memberId = stringIntegerPair.getFirst();
+ int remaining = stringIntegerPair.getSecond();
+ // assign the first few partitions from the list and then delete them from the availablePartitions list for this topic
+ List subList = availablePartitionsPerTopic.get(topicId).subList(0, remaining);
+ membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).addAll(subList);
+ availablePartitionsPerTopic.get(topicId).removeAll(subList);
+ }
+ }
+
+ // Consolidate the maps into MemberAssignment and then finally map each consumer to a MemberAssignment.
+ Map membersWithNewAssignment = new HashMap<>();
+ for (Map.Entry>> consumer : membersWithNewAssignmentPerTopic.entrySet()) {
+ String consumerId = consumer.getKey();
+ Map> assignmentPerTopic = consumer.getValue();
+ membersWithNewAssignment.computeIfAbsent(consumerId, k -> new MemberAssignment(assignmentPerTopic));
+ }
+
+ return new GroupAssignment(membersWithNewAssignment);
+ }
+}
+
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
new file mode 100644
index 0000000000000..228157280f610
--- /dev/null
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
@@ -0,0 +1,428 @@
+/*
+ * 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.assignor;
+
+
+import org.apache.kafka.common.Uuid;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.Set;
+
+
+
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class ServerSideStickyRangeAssignorTest {
+
+ private final ServerSideStickyRangeAssignor assignor = new ServerSideStickyRangeAssignor();
+
+ private final String topic1Name = "topic1";
+ private final Uuid topic1Uuid = Uuid.randomUuid();
+
+ private final String topic2Name = "topic2";
+ private final Uuid topic2Uuid = Uuid.randomUuid();
+
+ private final String topic3Name = "topic3";
+ private final Uuid topic3Uuid = Uuid.randomUuid();
+
+ private final String consumerA = "A";
+ private final String consumerB = "B";
+ private final String consumerC = "C";
+
+ @Test
+ public void testOneConsumerNoTopic() {
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ Map members = new HashMap<>();
+ List subscribedTopics = new ArrayList<>();
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopics, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment groupAssignment = assignor.assign(assignmentSpec);
+
+ assertTrue(groupAssignment.getMembers().isEmpty());
+ }
+
+ @Test
+ public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() {
+ // A -> T1, T3 // B -> T1, T3 // T1 -> 3 Partitions // T3 -> 2 Partitions
+ // Topics
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic3Uuid, new AssignmentTopicMetadata(topic3Name, 2));
+ // Members
+ Map members = new HashMap<>();
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+ members.put(consumerA, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, new HashMap<>()));
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+ members.put(consumerB, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ // Topic 3 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ }
+
+ @Test
+ public void testFirstAssignmentThreeConsumersThreeTopicsDifferentSubscriptions() {
+ // A -> T1, T2 // B -> T3 // C -> T2, T3 // T1 -> 3 Partitions // T2 -> 3 Partitions // T3 -> 2 Partitions
+ // Topics
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic3Uuid, new AssignmentTopicMetadata(topic3Name, 2));
+ // Members
+ Map members = new HashMap<>();
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ members.put(consumerA, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, new HashMap<>()));
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Collections.singletonList(topic3Uuid));
+ members.put(consumerB, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, new HashMap<>()));
+ // Consumer B
+ List subscribedTopicsC = new ArrayList<>(Arrays.asList(topic2Uuid, topic3Uuid));
+ members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1, 2)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ // Topic 3 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ }
+
+ @Test
+ public void testFirstAssignmentNumConsumersGreaterThanNumPartitions() {
+ // Topics
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ // Topic 3 has 2 partitions but three consumers subscribed to it - one of them will not get an assignment
+ topics.put(topic3Uuid, new AssignmentTopicMetadata(topic3Name, 2));
+ // Members
+ Map members = new HashMap<>();
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+ members.put(consumerA, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, new HashMap<>()));
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+ members.put(consumerB, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, new HashMap<>()));
+ // Consumer C
+ List subscribedTopicsC = new ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+ members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentNumConsumersGreaterThanNumPartitionsWhenOneConsumerAdded() {
+ // Topics
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 2));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 2));
+ // Members
+ Map members = new HashMap<>();
+ // Add a new consumer to trigger a re-assignment
+ // Consumer C
+ List subscribedTopicsC = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForA = new HashMap<>();
+ currentAssignmentForA.put(topic1Uuid, new HashSet<>(Collections.singletonList(0)));
+ currentAssignmentForA.put(topic2Uuid, new HashSet<>(Collections.singletonList(0)));
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(1)));
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(1)));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ // Consumer C shouldn't get any assignment, due to stickiness A, B retain their assignments
+ assertNull(computedAssignment.getMembers().get(consumerC));
+ assertAssignment(expectedAssignment, computedAssignment);
+ assertCoPartitionJoinProperty(computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentWhenOnePartitionAddedForTwoConsumersTwoTopics() {
+ // T1, T2 both have 3 partitions each when first assignment was calculated -> currentAssignmentForX
+ Map members = new HashMap<>();
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForA = new HashMap<>();
+ currentAssignmentForA.put(topic1Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ currentAssignmentForA.put(topic2Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
+
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(2)));
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+
+ // Simulating adding a partition - originally T1 -> 3 Partitions and T2 -> 3 Partitions
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 4));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 4));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(2, 3)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(2, 3)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ assertCoPartitionJoinProperty(computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentWhenOnePartitionRemovedForTwoConsumersTwoTopics() {
+ Map members = new HashMap<>();
+ // T1, T2 both have 3 partitions each initially and after removal have 2 partitions each
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForA = new HashMap<>();
+ currentAssignmentForA.put(topic1Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ currentAssignmentForA.put(topic2Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
+
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(2)));
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+
+ // Remove a partition from both topic1 and topic 2 -> total partitions = 2
+ // Simulating removing a partition - originally T1 -> 3 Partitions and T2 -> 3 Partitions
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 2));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 2));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ assertCoPartitionJoinProperty(computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentWhenOneConsumerAddedAfterInitialAssignmentWithTwoConsumersTwoTopics() {
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 3));
+
+ Map members = new HashMap<>();
+ // Consumer A
+ List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForA = new HashMap<>();
+ currentAssignmentForA.put(topic1Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ currentAssignmentForA.put(topic2Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(2)));
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+ // Add a new consumer to trigger a re-assignment
+ // Consumer C
+ List subscribedTopicsC = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ assertCoPartitionJoinProperty(computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentWhenOneConsumerRemovedAfterInitialAssignmentWithTwoConsumersTwoTopics() {
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 3));
+
+ Map members = new HashMap<>();
+ // Consumer A was removed
+
+ // Consumer B
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(2)));
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1, 2)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1, 2)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ assertCoPartitionJoinProperty(computedAssignment);
+ }
+
+ @Test
+ public void testReassignmentWhenMultipleSubscriptionsRemovedAfterInitialAssignmentWithThreeConsumersTwoTopics() {
+
+ Map topics = new HashMap<>();
+ topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 3));
+ topics.put(topic3Uuid, new AssignmentTopicMetadata(topic3Name, 2));
+
+ // Let initial subscriptions be A -> T1, T2 // B -> T2 // C -> T2, T3
+ // Change the subscriptions to A -> T1 // B -> T1, T2, T3 // C -> T2
+ Map members = new HashMap<>();
+ // Consumer A
+ Map> currentAssignmentForA = new HashMap<>();
+ currentAssignmentForA.put(topic1Uuid, new HashSet<>(Arrays.asList(0, 1, 2)));
+ currentAssignmentForA.put(topic2Uuid, new HashSet<>(Collections.singletonList(0)));
+ // Change subscriptions
+ List subscribedTopicsA = new ArrayList<>(Collections.singletonList(topic1Uuid));
+ members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
+ // Consumer B
+ Map> currentAssignmentForB = new HashMap<>();
+ currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(1)));
+ // Change subscriptions
+ List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid, topic3Uuid));
+ members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
+ // Consumer C
+ Map> currentAssignmentForC = new HashMap<>();
+ currentAssignmentForC.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
+ currentAssignmentForC.put(topic3Uuid, new HashSet<>(Arrays.asList(0, 1)));
+ // Change subscriptions
+ List subscribedTopicsC = new ArrayList<>(Collections.singletonList(topic2Uuid));
+ members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
+
+ AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+ GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+ Map>> expectedAssignment = new HashMap<>();
+ // Topic 1 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ // Topic 2 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+ expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+ // Topic 3 Partitions Assignment
+ expectedAssignment.computeIfAbsent(topic3Uuid, k -> new HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+
+ assertAssignment(expectedAssignment, computedAssignment);
+ }
+
+ // We have a set of sets with the partitions that should be distributed amongst the consumers, if it exists then remove it from the set.
+ private void assertAssignment(Map>> expectedAssignment, GroupAssignment computedGroupAssignment) {
+ for (Map.Entry member : computedGroupAssignment.getMembers().entrySet()) {
+ Map> computedAssignmentForMember = member.getValue().getAssignmentPerTopic();
+ for (Map.Entry> assignmentForTopic : computedAssignmentForMember.entrySet()) {
+ Uuid topicId = assignmentForTopic.getKey();
+ Set assignmentPartitionsSet = assignmentForTopic.getValue();
+ assertTrue(expectedAssignment.get(topicId).contains(assignmentPartitionsSet));
+ expectedAssignment.remove(assignmentPartitionsSet);
+ }
+ }
+ }
+
+ private void assertCoPartitionJoinProperty(GroupAssignment groupAssignment) {
+ for (Map.Entry member : groupAssignment.getMembers().entrySet()) {
+ Map> computedAssignmentForMember = member.getValue().getAssignmentPerTopic();
+ Set compareSet = new HashSet<>();
+ for (Map.Entry> topicAssignment : computedAssignmentForMember.entrySet()) {
+ Set partitionsForTopicSet = topicAssignment.getValue();
+ if (compareSet.isEmpty()) {
+ compareSet = partitionsForTopicSet;
+ }
+ assertEquals(compareSet, partitionsForTopicSet);
+ }
+ }
+ }
+}
From 3df34605e078beb9babcbeadc7b119a70da63796 Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Mon, 27 Mar 2023 20:17:14 -0700
Subject: [PATCH 02/44] removed reduce partitions case
---
.../ServerSideStickyRangeAssignor.java | 19 +++------
.../ServerSideStickyRangeAssignorTest.java | 39 -------------------
2 files changed, 6 insertions(+), 52 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
index 186f8af1f0220..e9d90ccfb7848 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
@@ -108,7 +108,7 @@ private Map> consumersPerTopic(AssignmentSpec assignmentSpec)
for (Map.Entry memberEntry : membersData.entrySet()) {
String memberId = memberEntry.getKey();
AssignmentMemberSpec memberMetadata = memberEntry.getValue();
- List topics = new ArrayList<>(memberMetadata.subscribedTopics);
+ List topics = memberMetadata.subscribedTopics;
for (Uuid topicId: topics) {
putList(mapTopicsToConsumers, topicId, memberId);
}
@@ -176,20 +176,13 @@ public GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAss
// Initially, minRequiredQuota is the minimum number of partitions that each consumer should get.
// The value is at least 1 unless numConsumers subscribed is greater than numPartitions. In such cases, all consumers get assigned "extra partitions".
int minRequiredQuota = numPartitionsPerConsumer;
- // We need to make sure that the partitions we're keeping are still part of the topic metadata.
- // Ex:- In case the number of partitions is reduced for a topic, the sticky partitions must be part of the new set of partitions available.
- for (int i = 0; i < currentAssignmentListForTopic.size() && currentAssignmentSize > 0; i++) {
- // numPartitionsForTopic - 1 is the max possible partition number
- if (currentAssignmentListForTopic.get(i) >= numPartitionsForTopic) {
- currentAssignmentSize--;
- }
- }
+
// If there are previously assigned partitions present, we want to retain
if (currentAssignmentSize > 0) {
// We either need to retain currentSize number of partitions when currentSize < required OR required number of partitions otherwise.
int retainedPartitionsCount = min(currentAssignmentSize, minRequiredQuota);
+ Collections.sort(currentAssignmentListForTopic);
for (int i = 0; i < retainedPartitionsCount; i++) {
- Collections.sort(currentAssignmentListForTopic);
putSet(assignedStickyPartitionsPerTopic, topicId, currentAssignmentListForTopic.get(i));
membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).add(currentAssignmentListForTopic.get(i));
}
@@ -243,14 +236,14 @@ public GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAss
for (Map.Entry>> unfilledEntry : unfilledConsumersPerTopic.entrySet()) {
Uuid topicId = unfilledEntry.getKey();
List> unfilledConsumersForTopic = unfilledEntry.getValue();
-
+ int newStartPointer = 0;
for (Pair stringIntegerPair : unfilledConsumersForTopic) {
String memberId = stringIntegerPair.getFirst();
int remaining = stringIntegerPair.getSecond();
// assign the first few partitions from the list and then delete them from the availablePartitions list for this topic
- List subList = availablePartitionsPerTopic.get(topicId).subList(0, remaining);
+ List subList = availablePartitionsPerTopic.get(topicId).subList(newStartPointer, newStartPointer + remaining);
+ newStartPointer += remaining;
membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).addAll(subList);
- availablePartitionsPerTopic.get(topicId).removeAll(subList);
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
index 228157280f610..54a73229718bb 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
@@ -242,45 +242,6 @@ public void testReassignmentWhenOnePartitionAddedForTwoConsumersTwoTopics() {
assertCoPartitionJoinProperty(computedAssignment);
}
- @Test
- public void testReassignmentWhenOnePartitionRemovedForTwoConsumersTwoTopics() {
- Map members = new HashMap<>();
- // T1, T2 both have 3 partitions each initially and after removal have 2 partitions each
- // Consumer A
- List subscribedTopicsA = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
- Map> currentAssignmentForA = new HashMap<>();
- currentAssignmentForA.put(topic1Uuid, new HashSet<>(Arrays.asList(0, 1)));
- currentAssignmentForA.put(topic2Uuid, new HashSet<>(Arrays.asList(0, 1)));
- members.computeIfAbsent(consumerA, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsA, currentAssignmentForA));
-
- // Consumer B
- List subscribedTopicsB = new ArrayList<>(Arrays.asList(topic1Uuid, topic2Uuid));
- Map> currentAssignmentForB = new HashMap<>();
- currentAssignmentForB.put(topic1Uuid, new HashSet<>(Collections.singletonList(2)));
- currentAssignmentForB.put(topic2Uuid, new HashSet<>(Collections.singletonList(2)));
- members.computeIfAbsent(consumerB, k -> new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, currentAssignmentForB));
-
- // Remove a partition from both topic1 and topic 2 -> total partitions = 2
- // Simulating removing a partition - originally T1 -> 3 Partitions and T2 -> 3 Partitions
- Map topics = new HashMap<>();
- topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 2));
- topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 2));
-
- AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
- GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
-
- Map>> expectedAssignment = new HashMap<>();
- // Topic 1 Partitions Assignment
- expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
- expectedAssignment.computeIfAbsent(topic1Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
- // Topic 2 Partitions Assignment
- expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
- expectedAssignment.computeIfAbsent(topic2Uuid, k -> new HashSet<>()).add(new HashSet<>(Collections.singletonList(1)));
-
- assertAssignment(expectedAssignment, computedAssignment);
- assertCoPartitionJoinProperty(computedAssignment);
- }
-
@Test
public void testReassignmentWhenOneConsumerAddedAfterInitialAssignmentWithTwoConsumersTwoTopics() {
Map topics = new HashMap<>();
From 74105291cc687fe41b75de692fc624ab1335341a Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Tue, 28 Mar 2023 12:21:33 -0700
Subject: [PATCH 03/44] removed reduce partitions case
---
.../group/assignor/ServerSideStickyRangeAssignor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
index e9d90ccfb7848..8c5f021405bde 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
@@ -60,7 +60,7 @@
public class ServerSideStickyRangeAssignor implements PartitionAssignor {
- public static final String RANGE_ASSIGNOR_NAME = "range-sticky";
+ public static final String RANGE_ASSIGNOR_NAME = "range";
@Override
public String name() {
From 3b6e65ea3b8452dd923991c3dba49d370c8f0d4e Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Wed, 29 Mar 2023 12:30:13 -0700
Subject: [PATCH 04/44] Addressed PR comments
---
.../ServerSideStickyRangeAssignor.java | 72 +++++++++----------
1 file changed, 34 insertions(+), 38 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
index 8c5f021405bde..c2e2dcae54c5c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
@@ -14,6 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.kafka.coordinator.group.assignor;
+
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.HashMap;
+import java.util.Collections;
+
+import static java.lang.Math.min;
/**
* The Server Side Sticky Range Assignor inherits properties of both the range assignor and the sticky assignor.
@@ -43,21 +57,6 @@
*
*
*/
-package org.apache.kafka.coordinator.group.assignor;
-
-
-import org.apache.kafka.common.Uuid;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.HashMap;
-import java.util.Collections;
-
-import static java.lang.Math.min;
-
public class ServerSideStickyRangeAssignor implements PartitionAssignor {
public static final String RANGE_ASSIGNOR_NAME = "range";
@@ -67,12 +66,12 @@ public String name() {
return RANGE_ASSIGNOR_NAME;
}
- protected static void putList(Map> map, K key, V value) {
+ private static void putList(Map> map, K key, V value) {
List list = map.computeIfAbsent(key, k -> new ArrayList<>());
list.add(value);
}
- protected static void putSet(Map> map, K key, V value) {
+ private static void putSet(Map> map, K key, V value) {
Set set = map.computeIfAbsent(key, k -> new HashSet<>());
set.add(value);
}
@@ -119,18 +118,19 @@ private Map> consumersPerTopic(AssignmentSpec assignmentSpec)
private Map> getAvailablePartitionsPerTopic(AssignmentSpec assignmentSpec, Map> assignedStickyPartitionsPerTopic) {
Map> availablePartitionsPerTopic = new HashMap<>();
Map topicsMetadata = assignmentSpec.topics;
- // Iterate through the topics map provided in assignmentSpec
+
for (Map.Entry topicMetadataEntry : topicsMetadata.entrySet()) {
Uuid topicId = topicMetadataEntry.getKey();
- availablePartitionsPerTopic.put(topicId, new ArrayList<>());
+ ArrayList availablePartitionsForTopic = new ArrayList<>();
int numPartitions = topicsMetadata.get(topicId).numPartitions;
// since the loop iterates from 0 to n, the partitions will be in ascending order within the list of available partitions per topic
- Set assignedStickyPartitionsForTopic = assignedStickyPartitionsPerTopic.get(topicId);
+ Set assignedStickyPartitionsForTopic = assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
for (int i = 0; i < numPartitions; i++) {
- if (assignedStickyPartitionsForTopic == null || !assignedStickyPartitionsForTopic.contains(i)) {
- availablePartitionsPerTopic.get(topicId).add(i);
+ if (!assignedStickyPartitionsForTopic.contains(i)) {
+ availablePartitionsForTopic.add(i);
}
}
+ availablePartitionsPerTopic.put(topicId, availablePartitionsForTopic);
}
return availablePartitionsPerTopic;
}
@@ -156,28 +156,23 @@ public GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAss
AssignmentTopicMetadata topicData = assignmentSpec.topics.get(topicId);
int numPartitionsForTopic = topicData.numPartitions;
- int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
+ // Initially, minRequiredQuota is the minimum number of partitions that each consumer should get i.e numPartitionsPerConsumer.
+ // Idle consumers case :- The numConsumers subscribed to a topic is greater than numPartitions. In such cases, all consumers get assigned via the "extra partitions" logic since min = 0.
+ int minRequiredQuota = numPartitionsForTopic / consumersForTopic.size();
// Each consumer can get only one extra partition per topic after receiving the minimum quota = numPartitionsPerConsumer
int numConsumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
for (String memberId: consumersForTopic) {
- // Size of the older assignment, this will be 0 when assign is called for the first time.
- // The older assignment is required when a reassignment occurs to ensure stickiness.
- int currentAssignmentSize = 0;
- List currentAssignmentListForTopic = new ArrayList<>();
+
// Convert the set to a list first and sort the partitions in numeric order since we want the same partition numbers from each topic
// to go to the same consumer in case of co-partitioned topics.
- Set currentAssignmentSetForTopic = assignmentSpec.members.get(memberId).currentAssignmentPerTopic.get(topicId);
- // We want to make sure that the currentAssignment is not null to avoid a null pointer exception
- if (currentAssignmentSetForTopic != null) {
- currentAssignmentListForTopic = new ArrayList<>(currentAssignmentSetForTopic);
- currentAssignmentSize = currentAssignmentListForTopic.size();
- }
- // Initially, minRequiredQuota is the minimum number of partitions that each consumer should get.
- // The value is at least 1 unless numConsumers subscribed is greater than numPartitions. In such cases, all consumers get assigned "extra partitions".
- int minRequiredQuota = numPartitionsPerConsumer;
+ Set currentAssignmentSetForTopic = assignmentSpec.members.get(memberId).currentAssignmentPerTopic.getOrDefault(topicId, new HashSet<>());
+ // Size of the older assignment, this will be 0 when assign is called for the first time.
+ // The older assignment is required when a reassignment occurs to ensure stickiness.
+ int currentAssignmentSize = currentAssignmentSetForTopic.size();
+ List currentAssignmentListForTopic = new ArrayList<>(currentAssignmentSetForTopic);
- // If there are previously assigned partitions present, we want to retain
+ // If there are previously assigned partitions present, we want to retain them.
if (currentAssignmentSize > 0) {
// We either need to retain currentSize number of partitions when currentSize < required OR required number of partitions otherwise.
int retainedPartitionsCount = min(currentAssignmentSize, minRequiredQuota);
@@ -187,7 +182,8 @@ public GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAss
membersWithNewAssignmentPerTopic.computeIfAbsent(memberId, k -> new HashMap<>()).computeIfAbsent(topicId, k -> new HashSet<>()).add(currentAssignmentListForTopic.get(i));
}
}
- // Number of partitions left to reach the minRequiredQuota
+
+ // Number of partitions left to reach the minRequiredQuota.
int remaining = minRequiredQuota - currentAssignmentSize;
// There are 3 cases w.r.t value of remaining
From 4b321aa5374427a15346be9b87b7c545ccb7db61 Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Mon, 3 Apr 2023 11:54:59 -0700
Subject: [PATCH 05/44] Addressed PR comments
---
.../assignor/ServerSideStickyRangeAssignorTest.java | 13 ++++++-------
1 file changed, 6 insertions(+), 7 deletions(-)
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
index 54a73229718bb..5704d68cc5fde 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
@@ -108,7 +108,7 @@ public void testFirstAssignmentThreeConsumersThreeTopicsDifferentSubscriptions()
// Consumer B
List subscribedTopicsB = new ArrayList<>(Collections.singletonList(topic3Uuid));
members.put(consumerB, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsB, new HashMap<>()));
- // Consumer B
+ // Consumer C
List subscribedTopicsC = new ArrayList<>(Arrays.asList(topic2Uuid, topic3Uuid));
members.put(consumerC, new AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopicsC, new HashMap<>()));
@@ -362,8 +362,8 @@ public void testReassignmentWhenMultipleSubscriptionsRemovedAfterInitialAssignme
// We have a set of sets with the partitions that should be distributed amongst the consumers, if it exists then remove it from the set.
private void assertAssignment(Map>> expectedAssignment, GroupAssignment computedGroupAssignment) {
- for (Map.Entry member : computedGroupAssignment.getMembers().entrySet()) {
- Map> computedAssignmentForMember = member.getValue().getAssignmentPerTopic();
+ for (MemberAssignment member : computedGroupAssignment.getMembers().values()) {
+ Map> computedAssignmentForMember = member.getAssignmentPerTopic();
for (Map.Entry> assignmentForTopic : computedAssignmentForMember.entrySet()) {
Uuid topicId = assignmentForTopic.getKey();
Set assignmentPartitionsSet = assignmentForTopic.getValue();
@@ -374,11 +374,10 @@ private void assertAssignment(Map>> expectedAssignment, G
}
private void assertCoPartitionJoinProperty(GroupAssignment groupAssignment) {
- for (Map.Entry member : groupAssignment.getMembers().entrySet()) {
- Map> computedAssignmentForMember = member.getValue().getAssignmentPerTopic();
+ for (MemberAssignment member : groupAssignment.getMembers().values()) {
+ Map> computedAssignmentForMember = member.getAssignmentPerTopic();
Set compareSet = new HashSet<>();
- for (Map.Entry> topicAssignment : computedAssignmentForMember.entrySet()) {
- Set partitionsForTopicSet = topicAssignment.getValue();
+ for (Set partitionsForTopicSet : computedAssignmentForMember.values()) {
if (compareSet.isEmpty()) {
compareSet = partitionsForTopicSet;
}
From 7352cbb5683f676382f56e6ae5c926f3cae98824 Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Thu, 6 Apr 2023 10:55:06 -0700
Subject: [PATCH 06/44] minor
---
.../group/assignor/ServerSideStickyRangeAssignorTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
index 5704d68cc5fde..73a7860239e0a 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
@@ -98,7 +98,7 @@ public void testFirstAssignmentThreeConsumersThreeTopicsDifferentSubscriptions()
// Topics
Map topics = new HashMap<>();
topics.put(topic1Uuid, new AssignmentTopicMetadata(topic1Name, 3));
- topics.put(topic2Uuid, new AssignmentTopicMetadata(topic1Name, 3));
+ topics.put(topic2Uuid, new AssignmentTopicMetadata(topic2Name, 3));
topics.put(topic3Uuid, new AssignmentTopicMetadata(topic3Name, 2));
// Members
Map members = new HashMap<>();
From 57b94ca208cb374b22d61eb044ef8004ea09479f Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Thu, 6 Apr 2023 11:12:16 -0700
Subject: [PATCH 07/44] Made subscribed topics a collection, removed * import
exception
---
checkstyle/suppressions.xml | 2 --
.../group/assignor/AssignmentMemberSpec.java | 7 ++++---
.../assignor/ServerSideStickyRangeAssignor.java | 12 ++++++------
.../ServerSideStickyRangeAssignorTest.java | 16 ++++++++++------
4 files changed, 20 insertions(+), 17 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index 1bf330e52fc73..e2a501ae3a575 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -323,8 +323,6 @@
files="ServerSideStickyRangeAssignor.java"/>
-
subscribedTopics;
+ final Collection subscribedTopics;
/**
* Maps the partitions assigned for this member per topicId
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
index c2e2dcae54c5c..bb89296d0c0ee 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignor.java
@@ -16,16 +16,16 @@
*/
package org.apache.kafka.coordinator.group.assignor;
-
import org.apache.kafka.common.Uuid;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.HashMap;
-import java.util.Collections;
import static java.lang.Math.min;
@@ -107,7 +107,7 @@ private Map> consumersPerTopic(AssignmentSpec assignmentSpec)
for (Map.Entry memberEntry : membersData.entrySet()) {
String memberId = memberEntry.getKey();
AssignmentMemberSpec memberMetadata = memberEntry.getValue();
- List topics = memberMetadata.subscribedTopics;
+ Collection topics = memberMetadata.subscribedTopics;
for (Uuid topicId: topics) {
putList(mapTopicsToConsumers, topicId, memberId);
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
index 73a7860239e0a..52d36eece5610 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/ServerSideStickyRangeAssignorTest.java
@@ -17,19 +17,23 @@
package org.apache.kafka.coordinator.group.assignor;
-
import org.apache.kafka.common.Uuid;
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.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
-
-
-import java.util.*;
-
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class ServerSideStickyRangeAssignorTest {
From 5065109332845541c5591a52683f21d37982d80a Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Thu, 6 Apr 2023 13:43:26 -0700
Subject: [PATCH 08/44] Interface changes for assignment map and subscribed
topics, added new TopicIdToPartition data structure
---
.../group/assignor/AssignmentMemberSpec.java | 27 +++++-----
.../group/assignor/MemberAssignment.java | 27 +++++-----
.../group/common/TopicIdToPartition.java | 50 +++++++++++++++++++
3 files changed, 80 insertions(+), 24 deletions(-)
create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
index 2db33ddc74158..b3e63994d980b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
@@ -16,11 +16,13 @@
*/
package org.apache.kafka.coordinator.group.assignor;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
import java.util.Collection;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
/**
* The assignment specification for a consumer group member.
@@ -37,29 +39,30 @@ public class AssignmentMemberSpec {
final Optional rackId;
/**
- * The topics that the member is subscribed to.
+ * The topicIds of topics that the member is subscribed to.
*/
- final Collection subscribedTopics;
+ final Collection subscribedTopics;
/**
- * The current target partitions of the member.
+ * Partitions assigned for this member grouped by topicId
*/
- final Collection targetPartitions;
+ final Map> currentAssignmentTopicIdPartitions;
public AssignmentMemberSpec(
+
Optional instanceId,
Optional rackId,
- Collection subscribedTopics,
- Collection targetPartitions
+ Collection subscribedTopics,
+ Map> currentAssignmentTopicIdPartitions
) {
Objects.requireNonNull(instanceId);
Objects.requireNonNull(rackId);
Objects.requireNonNull(subscribedTopics);
- Objects.requireNonNull(targetPartitions);
+ Objects.requireNonNull(currentAssignmentTopicIdPartitions);
this.instanceId = instanceId;
this.rackId = rackId;
this.subscribedTopics = subscribedTopics;
- this.targetPartitions = targetPartitions;
+ this.currentAssignmentTopicIdPartitions = currentAssignmentTopicIdPartitions;
}
@Override
@@ -72,7 +75,7 @@ public boolean equals(Object o) {
if (!instanceId.equals(that.instanceId)) return false;
if (!rackId.equals(that.rackId)) return false;
if (!subscribedTopics.equals(that.subscribedTopics)) return false;
- return targetPartitions.equals(that.targetPartitions);
+ return currentAssignmentTopicIdPartitions.equals(that.currentAssignmentTopicIdPartitions);
}
@Override
@@ -80,7 +83,7 @@ public int hashCode() {
int result = instanceId.hashCode();
result = 31 * result + rackId.hashCode();
result = 31 * result + subscribedTopics.hashCode();
- result = 31 * result + targetPartitions.hashCode();
+ result = 31 * result + currentAssignmentTopicIdPartitions.hashCode();
return result;
}
@@ -89,7 +92,7 @@ public String toString() {
return "AssignmentMemberSpec(instanceId=" + instanceId +
", rackId=" + rackId +
", subscribedTopics=" + subscribedTopics +
- ", targetPartitions=" + targetPartitions +
+ ", targetPartitions=" + currentAssignmentTopicIdPartitions +
')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
index 86c235d718771..fa05e4398222b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
@@ -16,25 +16,28 @@
*/
package org.apache.kafka.coordinator.group.assignor;
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
-import java.util.Collection;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
/**
* The partition assignment for a consumer group member.
*/
public class MemberAssignment {
/**
- * The target partitions assigned to this member.
+ * The target partitions assigned to this member grouped by topicId.
*/
- final Collection targetPartitions;
+ private final Map> topicIdPartitionsMap;
- public MemberAssignment(
- Collection targetPartitions
- ) {
- Objects.requireNonNull(targetPartitions);
- this.targetPartitions = targetPartitions;
+ public MemberAssignment(Map> topicIdPartitionsForAssignment) {
+ Objects.requireNonNull(topicIdPartitionsForAssignment);
+ this.topicIdPartitionsMap = topicIdPartitionsForAssignment;
+ }
+
+ public Map> getTopicIdPartitionsMap() {
+ return this.topicIdPartitionsMap;
}
@Override
@@ -44,16 +47,16 @@ public boolean equals(Object o) {
MemberAssignment that = (MemberAssignment) o;
- return targetPartitions.equals(that.targetPartitions);
+ return topicIdPartitionsMap.equals(that.topicIdPartitionsMap);
}
@Override
public int hashCode() {
- return targetPartitions.hashCode();
+ return topicIdPartitionsMap.hashCode();
}
@Override
public String toString() {
- return "MemberAssignment(targetPartitions=" + targetPartitions + ')';
+ return "MemberAssignment ( Assignment per topic Id = " + topicIdPartitionsMap + ')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
new file mode 100644
index 0000000000000..c84e0c1384123
--- /dev/null
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
@@ -0,0 +1,50 @@
+/*
+ * 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.common;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+public class TopicIdToPartition {
+ private final Uuid topicId;
+ private final Integer partition;
+ private final Optional> rackIds;
+
+ public TopicIdToPartition(Uuid topicId, Integer topicPartition, Optional> rackIds) {
+ this.topicId = Objects.requireNonNull(topicId, "topicId can not be null");
+ this.partition = Objects.requireNonNull(topicPartition, "topicPartition can not be null");
+ this.rackIds = rackIds;
+ }
+
+ /**
+ * @return Universally unique id representing this topic partition.
+ */
+ public Uuid topicId() {
+ return topicId;
+ }
+
+ /**
+ * @return the partition number.
+ */
+ public int partition() {
+ return partition;
+ }
+
+}
From 100a04e5e064be0bda392ca8f02cb7a7e89f16c8 Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Fri, 7 Apr 2023 11:25:33 -0700
Subject: [PATCH 09/44] Added toString method and hash
---
.../group/common/TopicIdToPartition.java | 26 +++++++++++++++++++
1 file changed, 26 insertions(+)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
index c84e0c1384123..ea7d1114feb32 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
@@ -47,4 +47,30 @@ public int partition() {
return partition;
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TopicIdToPartition that = (TopicIdToPartition) o;
+ return topicId.equals(that.topicId) &&
+ partition.equals(that.partition);
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = prime + topicId.hashCode();
+ result = prime * result + partition.hashCode();
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return topicId() + "-" + partition();
+ }
+
}
From 73c7fdcaf824a31af442b1a4aa2db81b37cc5a9e Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Tue, 11 Apr 2023 15:09:44 -0700
Subject: [PATCH 10/44] Made all attributes private and added getter methods,
changed names according to PR comments
---
.../group/assignor/AssignmentMemberSpec.java | 39 +++++++++++++------
.../group/assignor/AssignmentSpec.java | 12 +++++-
.../assignor/AssignmentTopicMetadata.java | 12 +++++-
.../group/assignor/GroupAssignment.java | 6 ++-
.../group/assignor/MemberAssignment.java | 16 ++++----
.../group/common/TopicIdToPartition.java | 6 +--
6 files changed, 63 insertions(+), 28 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
index b3e63994d980b..33dfe71165c85 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
@@ -31,38 +31,53 @@ public class AssignmentMemberSpec {
/**
* The instance ID if provided.
*/
- final Optional instanceId;
+ private final Optional instanceId;
/**
* The rack ID if provided.
*/
- final Optional rackId;
+ private final Optional rackId;
/**
* The topicIds of topics that the member is subscribed to.
*/
- final Collection subscribedTopics;
+ private final Collection subscribedTopics;
/**
- * Partitions assigned for this member grouped by topicId
+ * Partitions assigned for this member keyed by topicId
*/
- final Map> currentAssignmentTopicIdPartitions;
+ private final Map> assignedTopicIdPartitions;
- public AssignmentMemberSpec(
+ public Optional instanceId() {
+ return instanceId;
+ }
+ public Optional rackId() {
+ return rackId;
+ }
+
+ public Collection subscribedTopics() {
+ return subscribedTopics;
+ }
+
+ public Map> assignmentTopicIdPartitions() {
+ return assignedTopicIdPartitions;
+ }
+
+ public AssignmentMemberSpec(
Optional instanceId,
Optional rackId,
Collection subscribedTopics,
- Map> currentAssignmentTopicIdPartitions
+ Map> assignedTopicIdPartitions
) {
Objects.requireNonNull(instanceId);
Objects.requireNonNull(rackId);
Objects.requireNonNull(subscribedTopics);
- Objects.requireNonNull(currentAssignmentTopicIdPartitions);
+ Objects.requireNonNull(assignedTopicIdPartitions);
this.instanceId = instanceId;
this.rackId = rackId;
this.subscribedTopics = subscribedTopics;
- this.currentAssignmentTopicIdPartitions = currentAssignmentTopicIdPartitions;
+ this.assignedTopicIdPartitions = assignedTopicIdPartitions;
}
@Override
@@ -75,7 +90,7 @@ public boolean equals(Object o) {
if (!instanceId.equals(that.instanceId)) return false;
if (!rackId.equals(that.rackId)) return false;
if (!subscribedTopics.equals(that.subscribedTopics)) return false;
- return currentAssignmentTopicIdPartitions.equals(that.currentAssignmentTopicIdPartitions);
+ return assignedTopicIdPartitions.equals(that.assignedTopicIdPartitions);
}
@Override
@@ -83,7 +98,7 @@ public int hashCode() {
int result = instanceId.hashCode();
result = 31 * result + rackId.hashCode();
result = 31 * result + subscribedTopics.hashCode();
- result = 31 * result + currentAssignmentTopicIdPartitions.hashCode();
+ result = 31 * result + assignedTopicIdPartitions.hashCode();
return result;
}
@@ -92,7 +107,7 @@ public String toString() {
return "AssignmentMemberSpec(instanceId=" + instanceId +
", rackId=" + rackId +
", subscribedTopics=" + subscribedTopics +
- ", targetPartitions=" + currentAssignmentTopicIdPartitions +
+ ", assignedTopicIdPartitions=" + assignedTopicIdPartitions +
')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentSpec.java
index f56b272b787c3..e36a66fd4ebd2 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentSpec.java
@@ -28,12 +28,12 @@ public class AssignmentSpec {
/**
* The members keyed by member id.
*/
- final Map members;
+ private final Map members;
/**
* The topics' metadata keyed by topic id
*/
- final Map topics;
+ private final Map topics;
public AssignmentSpec(
Map members,
@@ -45,6 +45,14 @@ public AssignmentSpec(
this.topics = topics;
}
+ public Map members() {
+ return members;
+ }
+
+ public Map topics() {
+ return topics;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentTopicMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentTopicMetadata.java
index ba79c82b5acd4..43f9a003efcf7 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentTopicMetadata.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentTopicMetadata.java
@@ -25,12 +25,12 @@ public class AssignmentTopicMetadata {
/**
* The topic name.
*/
- final String topicName;
+ private final String topicName;
/**
* The number of partitions.
*/
- final int numPartitions;
+ private final int numPartitions;
public AssignmentTopicMetadata(
String topicName,
@@ -41,6 +41,14 @@ public AssignmentTopicMetadata(
this.numPartitions = numPartitions;
}
+ public String topicName() {
+ return topicName;
+ }
+
+ public int numPartitions() {
+ return numPartitions;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
index 5c0199aaec5a5..70ff9bafcca3f 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java
@@ -26,7 +26,7 @@ public class GroupAssignment {
/**
* The member assignments keyed by member id.
*/
- final Map members;
+ private final Map members;
public GroupAssignment(
Map members
@@ -35,6 +35,10 @@ public GroupAssignment(
this.members = members;
}
+ public Map members() {
+ return members;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
index fa05e4398222b..09bf1789df982 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java
@@ -27,17 +27,17 @@
*/
public class MemberAssignment {
/**
- * The target partitions assigned to this member grouped by topicId.
+ * The target partitions assigned to this member keyed by topicId.
*/
- private final Map> topicIdPartitionsMap;
+ private final Map> assignedTopicIdPartitions;
public MemberAssignment(Map> topicIdPartitionsForAssignment) {
Objects.requireNonNull(topicIdPartitionsForAssignment);
- this.topicIdPartitionsMap = topicIdPartitionsForAssignment;
+ this.assignedTopicIdPartitions = topicIdPartitionsForAssignment;
}
- public Map> getTopicIdPartitionsMap() {
- return this.topicIdPartitionsMap;
+ public Map> assignedTopicIdPartitions() {
+ return this.assignedTopicIdPartitions;
}
@Override
@@ -47,16 +47,16 @@ public boolean equals(Object o) {
MemberAssignment that = (MemberAssignment) o;
- return topicIdPartitionsMap.equals(that.topicIdPartitionsMap);
+ return assignedTopicIdPartitions.equals(that.assignedTopicIdPartitions);
}
@Override
public int hashCode() {
- return topicIdPartitionsMap.hashCode();
+ return assignedTopicIdPartitions.hashCode();
}
@Override
public String toString() {
- return "MemberAssignment ( Assignment per topic Id = " + topicIdPartitionsMap + ')';
+ return "MemberAssignment (Assignment per topic Id = " + assignedTopicIdPartitions + ')';
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
index ea7d1114feb32..3aaccbcef0436 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
@@ -28,9 +28,9 @@ public class TopicIdToPartition {
private final Optional> rackIds;
public TopicIdToPartition(Uuid topicId, Integer topicPartition, Optional> rackIds) {
- this.topicId = Objects.requireNonNull(topicId, "topicId can not be null");
- this.partition = Objects.requireNonNull(topicPartition, "topicPartition can not be null");
- this.rackIds = rackIds;
+ this.topicId = Objects.requireNonNull(topicId, "topicId cannot be null");
+ this.partition = Objects.requireNonNull(topicPartition, "topicPartition cannot be null");
+ this.rackIds = Objects.requireNonNull(rackIds, "rackId cannot be null");
}
/**
From 68d53b37b5e3fd78b8f96d7917b1fdc565e188dd Mon Sep 17 00:00:00 2001
From: Ritika Reddy
Date: Wed, 12 Apr 2023 13:00:07 -0700
Subject: [PATCH 11/44] Removed topicIdToPartition class, changed attribute
names and added java doc for getter methods
---
.../group/assignor/AssignmentMemberSpec.java | 52 ++++++++-----
.../group/assignor/AssignmentSpec.java | 8 +-
.../assignor/AssignmentTopicMetadata.java | 6 ++
.../group/assignor/GroupAssignment.java | 3 +
.../group/assignor/MemberAssignment.java | 21 ++---
.../group/common/TopicIdToPartition.java | 76 -------------------
6 files changed, 60 insertions(+), 106 deletions(-)
delete mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/common/TopicIdToPartition.java
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
index 33dfe71165c85..8bcc08dc19eb3 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java
@@ -39,45 +39,57 @@ public class AssignmentMemberSpec {
private final Optional rackId;
/**
- * The topicIds of topics that the member is subscribed to.
+ * Topics Ids that the member is subscribed to.
*/
- private final Collection subscribedTopics;
+ private final Collection subscribedTopicIds;
/**
- * Partitions assigned for this member keyed by topicId
+ * Partitions assigned keyed by topicId.
*/
- private final Map> assignedTopicIdPartitions;
+ private final Map> assignedPartitions;
+ /**
+ * @return The instance ID as an Optional.
+ */
public Optional instanceId() {
return instanceId;
}
+ /**
+ * @return The rack ID as an Optional.
+ */
public Optional rackId() {
return rackId;
}
- public Collection subscribedTopics() {
- return subscribedTopics;
+ /**
+ * @return Collection of subscribed topic Ids.
+ */
+ public Collection subscribedTopicIds() {
+ return subscribedTopicIds;
}
- public Map> assignmentTopicIdPartitions() {
- return assignedTopicIdPartitions;
+ /**
+ * @return Assigned partitions keyed by topic Ids.
+ */
+ public Map> assignedPartitions() {
+ return assignedPartitions;
}
public AssignmentMemberSpec(
Optional instanceId,
Optional rackId,
- Collection subscribedTopics,
- Map