[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313285352
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be:
+ * balance > rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313213976
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+RackInfo rackInfo = new RackInfo(assignmentSpec,
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313434102
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313408646
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313393423
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313294383
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313292815
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313285352
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be:
+ * balance > rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313283639
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java:
##
@@ -0,0 +1,270 @@
+/*
+ * 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.coordinator.group.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * The Uniform Assignor distributes Kafka topic partitions among group members
for balanced assignment.
+ * The assignor employs two different strategies based on the nature of topic
+ * subscriptions across the group members:
+ *
+ *
+ * Optimized Uniform Assignment Builder: This strategy is
used when all members have subscribed
+ * to the same set of topics.
+ *
+ *
+ * General Uniform Assignment Builder: This strategy is used
when members have varied topic
+ * subscriptions.
+ *
+ *
+ *
+ * The appropriate strategy is automatically chosen based on the current
members' topic subscriptions.
+ *
+ * @see OptimizedUniformAssignmentBuilder
+ * @see GeneralUniformAssignmentBuilder
+ */
+public class UniformAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(UniformAssignor.class);
+public static final String UNIFORM_ASSIGNOR_NAME = "uniform";
+
+@Override
+public String name() {
+return UNIFORM_ASSIGNOR_NAME;
+}
+
+/**
+ * Perform the group assignment given the current members and
+ * topic metadata.
+ *
+ * @param assignmentSpecThe member assignment spec.
+ * @param subscribedTopicDescriber The topic and cluster metadata
describer {@link SubscribedTopicDescriber}.
+ * @return The new assignment for the group.
+ */
+@Override
+public GroupAssignment assign(
+AssignmentSpec assignmentSpec,
+SubscribedTopicDescriber subscribedTopicDescriber
+) throws PartitionAssignorException {
+
+AbstractAssignmentBuilder assignmentBuilder;
+if (allSubscriptionsEqual(assignmentSpec.members())) {
+log.debug("Detected that all members are subscribed to the same
set of topics, invoking the "
++ "optimized assignment algorithm");
+assignmentBuilder = new
OptimizedUniformAssignmentBuilder(assignmentSpec, subscribedTopicDescriber);
+} else {
+assignmentBuilder = new GeneralUniformAssignmentBuilder();
+log.debug("Detected that all members are subscribed to a different
set of topics, invoking the "
++ "general assignment algorithm");
+}
+return assignmentBuilder.buildAssignment();
+}
+
+/**
+ * Determines if all members are subscribed to the same list of topic IDs.
+ *
+ * @param members A map of member identifiers to their respective {@code
AssignmentMemberSpec}.
+ *Assumes the map is non-empty.
+ * @return true if all members have the same subscription list of topic
IDs,
+ * false otherwise.
+ */
+private boolean allSubscriptionsEqual(Map
members) {
+Set firstSubscriptionSet = new
HashSet<>(members.values().iterator().next().subscribedTopicIds());
Review Comment:
no, unless we have a bug. can we add a check in assign() and throw illegal
argument exception if members is empty?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313281824
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1313213976
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+RackInfo rackInfo = new RackInfo(assignmentSpec,
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1312351774
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1312351774
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1310869571
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1310874485
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilderTest.java:
##
@@ -0,0 +1,1070 @@
+/*
+ * 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.apache.kafka.coordinator.group.consumer.SubscribedTopicMetadata;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+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.TreeMap;
+
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
+import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
+import static
org.apache.kafka.coordinator.group.RecordHelpersTest.mkMapOfPartitionRacks;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class OptimizedUniformAssignmentBuilderTest {
+private final UniformAssignor assignor = new UniformAssignor();
+private final Uuid topic1Uuid = Uuid.fromString("T1-A4s3VTwiI5CTbEp6POw");
+private final Uuid topic2Uuid = Uuid.fromString("T2-B4s3VTwiI5YHbPp6YUe");
+private final Uuid topic3Uuid = Uuid.fromString("T3-CU8fVTLCz5YMkLoDQsa");
+private final String topic1Name = "topic1";
+private final String topic2Name = "topic2";
+private final String topic3Name = "topic3";
+private final String memberA = "A";
+private final String memberB = "B";
+private final String memberC = "C";
+
+@Test
+public void testOneMemberNoTopicSubscription() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.emptyList(),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void testOneMemberSubscribedToNonexistentTopic() {
+SubscribedTopicMetadata subscribedTopicMetadata = new
SubscribedTopicMetadata(
+Collections.singletonMap(
+topic1Uuid,
+new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+)
+)
+);
+
+Map members = Collections.singletonMap(
+memberA,
+new AssignmentMemberSpec(
+Optional.empty(),
+Optional.empty(),
+Collections.singletonList(topic2Uuid),
+Collections.emptyMap()
+)
+);
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec,
subscribedTopicMetadata);
+
+assertEquals(Collections.emptyMap(), groupAssignment.members());
+}
+
+@Test
+public void
testFirstAssignmentTwoMembersSubscribedToTwoTopicsNoMemberRacks() {
+Map topicMetadata = new HashMap<>();
+topicMetadata.put(topic1Uuid, new TopicMetadata(
+topic1Uuid,
+topic1Name,
+3,
+mkMapOfPartitionRacks(3)
+));
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1310880384
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+RackInfo rackInfo = new RackInfo(assignmentSpec,
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1309505394
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+RackInfo rackInfo = new RackInfo(assignmentSpec,
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1309504407
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
Review Comment:
can we add a test case?
--
This is an
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1309478864
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.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.assignor;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be:
+ * balance > rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list
+ * and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetAnExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+this.subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1309478048
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignor.java:
##
@@ -0,0 +1,220 @@
+/*
+ * 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.coordinator.group.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The Uniform Assignor distributes Kafka topic partitions among group members
for balanced assignment.
+ * The assignor employs two different strategies based on the nature of topic
+ * subscriptions across the group members:
+ *
+ *
+ * Optimized Uniform Assignor: This strategy is used when all
members have subscribed
+ * to the same set of topics.
+ *
+ *
+ * General Uniform Assignor: This strategy is used when
members have varied topic
+ * subscriptions.
+ *
+ *
+ *
+ * The appropriate strategy is automatically chosen based on the current
members' topic subscriptions.
+ *
+ * @see OptimizedUniformAssignor
+ * @see GeneralUniformAssignor
+ */
+public abstract class AbstractUniformAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(AbstractUniformAssignor.class);
+public static final String UNIFORM_ASSIGNOR_NAME = "uniform";
+
+@Override
+public String name() {
+return UNIFORM_ASSIGNOR_NAME;
+}
+
+/**
+ * Perform the group assignment given the current members and
+ * topic metadata.
+ *
+ * @param assignmentSpec The member assignment spec.
+ * @param subscribedTopicDescriber The topic and cluster metadata
describer {@link SubscribedTopicDescriber}.
+ * @return The new assignment for the group.
+ */
+@Override
+public GroupAssignment assign(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) throws
PartitionAssignorException {
+if (allSubscriptionsEqual(assignmentSpec.members())) {
+log.debug("Detected that all members are subscribed to the same
set of topics, invoking the "
++ "optimized assignment algorithm");
+OptimizedUniformAssignor optimizedUniformAssignor = new
OptimizedUniformAssignor(assignmentSpec, subscribedTopicDescriber);
+return optimizedUniformAssignor.build();
+} else {
+GeneralUniformAssignor generalUniformAssignor = new
GeneralUniformAssignor();
+log.debug("Detected that all members are subscribed to a different
set of topics, invoking the "
++ "general assignment algorithm");
+return generalUniformAssignor.build();
+}
+}
+
+/**
+ * Builds the assignment for the given topic and member metadata.
+ *
+ * @return The new assignment for the group.
+ */
+protected abstract GroupAssignment build();
+
+private boolean allSubscriptionsEqual(Map
members) {
+boolean areAllSubscriptionsEqual = true;
+Collection firstSubscriptionList =
members.values().iterator().next().subscribedTopicIds();
+for (AssignmentMemberSpec memberSpec : members.values()) {
+if
(!firstSubscriptionList.equals(memberSpec.subscribedTopicIds())) {
+areAllSubscriptionsEqual = false;
+break;
+}
+}
+return areAllSubscriptionsEqual;
+}
+
+protected boolean useRackAwareAssignment(Set consumerRacks,
Set partitionRacks, Map>
racksPerPartition) {
+if (consumerRacks.isEmpty() || Collections.disjoint(consumerRacks,
partitionRacks))
+return false;
+else {
+return
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1296343128
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignmentBuilder.java:
##
@@ -0,0 +1,399 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ * The order of priority of properties during the assignment will be: balance
> rack matching (when applicable) > stickiness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
current owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignmentBuilder extends
UniformAssignor.AbstractAssignmentBuilder {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignmentBuilder.class);
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final RackInfo rackInfo;
+// Count of members to receive an extra partition beyond the minimum quota,
+// to account for the distribution of the remaining partitions.
+private int remainingMembersToGetExtraPartition;
+// Map of members to the remaining number of partitions needed to meet the
minimum quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the remaining number of partitions needed to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the current owner of each partition when using rack-aware
strategy.
+// Current refers to the existing assignment.
+private final Map currentPartitionOwners;
+// Indicates if a rack aware assignment can be done.
+// True if racks are defined for both members and partitions.
+boolean useRackAwareStrategy;
+
+OptimizedUniformAssignmentBuilder(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) {
+this.assignmentSpec = assignmentSpec;
+this.subscribedTopicDescriber = subscribedTopicDescriber;
+subscriptionList = new
ArrayList<>(assignmentSpec.members().values().iterator().next().subscribedTopicIds());
+
+RackInfo rackInfo = new RackInfo(assignmentSpec,
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1291721804
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignor.java:
##
@@ -0,0 +1,220 @@
+/*
+ * 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.coordinator.group.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The Uniform Assignor distributes Kafka topic partitions among group members
for balanced assignment.
+ * The assignor employs two different strategies based on the nature of topic
+ * subscriptions across the group members:
+ *
+ *
+ * Optimized Uniform Assignor: This strategy is used when all
members have subscribed
+ * to the same set of topics.
+ *
+ *
+ * General Uniform Assignor: This strategy is used when
members have varied topic
+ * subscriptions.
+ *
+ *
+ *
+ * The appropriate strategy is automatically chosen based on the current
members' topic subscriptions.
+ *
+ * @see OptimizedUniformAssignor
+ * @see GeneralUniformAssignor
+ */
+public abstract class AbstractUniformAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(AbstractUniformAssignor.class);
+public static final String UNIFORM_ASSIGNOR_NAME = "uniform";
+
+@Override
+public String name() {
+return UNIFORM_ASSIGNOR_NAME;
+}
+
+/**
+ * Perform the group assignment given the current members and
+ * topic metadata.
+ *
+ * @param assignmentSpec The member assignment spec.
+ * @param subscribedTopicDescriber The topic and cluster metadata
describer {@link SubscribedTopicDescriber}.
+ * @return The new assignment for the group.
+ */
+@Override
+public GroupAssignment assign(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) throws
PartitionAssignorException {
+if (allSubscriptionsEqual(assignmentSpec.members())) {
+log.debug("Detected that all members are subscribed to the same
set of topics, invoking the "
++ "optimized assignment algorithm");
+OptimizedUniformAssignor optimizedUniformAssignor = new
OptimizedUniformAssignor(assignmentSpec, subscribedTopicDescriber);
+return optimizedUniformAssignor.build();
+} else {
+GeneralUniformAssignor generalUniformAssignor = new
GeneralUniformAssignor();
+log.debug("Detected that all members are subscribed to a different
set of topics, invoking the "
++ "general assignment algorithm");
+return generalUniformAssignor.build();
+}
+}
+
+/**
+ * Builds the assignment for the given topic and member metadata.
+ *
+ * @return The new assignment for the group.
+ */
+protected abstract GroupAssignment build();
+
+private boolean allSubscriptionsEqual(Map
members) {
+boolean areAllSubscriptionsEqual = true;
+Collection firstSubscriptionList =
members.values().iterator().next().subscribedTopicIds();
+for (AssignmentMemberSpec memberSpec : members.values()) {
+if
(!firstSubscriptionList.equals(memberSpec.subscribedTopicIds())) {
+areAllSubscriptionsEqual = false;
+break;
+}
+}
+return areAllSubscriptionsEqual;
+}
+
+protected boolean useRackAwareAssignment(Set consumerRacks,
Set partitionRacks, Map>
racksPerPartition) {
+if (consumerRacks.isEmpty() || Collections.disjoint(consumerRacks,
partitionRacks))
+return false;
+else {
+return
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1291714013
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignor.java:
##
@@ -0,0 +1,220 @@
+/*
+ * 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.coordinator.group.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The Uniform Assignor distributes Kafka topic partitions among group members
for balanced assignment.
+ * The assignor employs two different strategies based on the nature of topic
+ * subscriptions across the group members:
+ *
+ *
+ * Optimized Uniform Assignor: This strategy is used when all
members have subscribed
+ * to the same set of topics.
+ *
+ *
+ * General Uniform Assignor: This strategy is used when
members have varied topic
+ * subscriptions.
+ *
+ *
+ *
+ * The appropriate strategy is automatically chosen based on the current
members' topic subscriptions.
+ *
+ * @see OptimizedUniformAssignor
+ * @see GeneralUniformAssignor
+ */
+public abstract class AbstractUniformAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(AbstractUniformAssignor.class);
+public static final String UNIFORM_ASSIGNOR_NAME = "uniform";
+
+@Override
+public String name() {
+return UNIFORM_ASSIGNOR_NAME;
+}
+
+/**
+ * Perform the group assignment given the current members and
+ * topic metadata.
+ *
+ * @param assignmentSpec The member assignment spec.
+ * @param subscribedTopicDescriber The topic and cluster metadata
describer {@link SubscribedTopicDescriber}.
+ * @return The new assignment for the group.
+ */
+@Override
+public GroupAssignment assign(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) throws
PartitionAssignorException {
+if (allSubscriptionsEqual(assignmentSpec.members())) {
+log.debug("Detected that all members are subscribed to the same
set of topics, invoking the "
++ "optimized assignment algorithm");
+OptimizedUniformAssignor optimizedUniformAssignor = new
OptimizedUniformAssignor(assignmentSpec, subscribedTopicDescriber);
+return optimizedUniformAssignor.build();
+} else {
+GeneralUniformAssignor generalUniformAssignor = new
GeneralUniformAssignor();
+log.debug("Detected that all members are subscribed to a different
set of topics, invoking the "
++ "general assignment algorithm");
+return generalUniformAssignor.build();
+}
+}
+
+/**
+ * Builds the assignment for the given topic and member metadata.
+ *
+ * @return The new assignment for the group.
+ */
+protected abstract GroupAssignment build();
Review Comment:
AssignorBuilder.build() makes sense. Assignor.build() implies that we're
building an assignor, not building an assignment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1291714013
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignor.java:
##
@@ -0,0 +1,220 @@
+/*
+ * 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.coordinator.group.common.TopicIdPartition;
+import org.apache.kafka.common.Uuid;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The Uniform Assignor distributes Kafka topic partitions among group members
for balanced assignment.
+ * The assignor employs two different strategies based on the nature of topic
+ * subscriptions across the group members:
+ *
+ *
+ * Optimized Uniform Assignor: This strategy is used when all
members have subscribed
+ * to the same set of topics.
+ *
+ *
+ * General Uniform Assignor: This strategy is used when
members have varied topic
+ * subscriptions.
+ *
+ *
+ *
+ * The appropriate strategy is automatically chosen based on the current
members' topic subscriptions.
+ *
+ * @see OptimizedUniformAssignor
+ * @see GeneralUniformAssignor
+ */
+public abstract class AbstractUniformAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(AbstractUniformAssignor.class);
+public static final String UNIFORM_ASSIGNOR_NAME = "uniform";
+
+@Override
+public String name() {
+return UNIFORM_ASSIGNOR_NAME;
+}
+
+/**
+ * Perform the group assignment given the current members and
+ * topic metadata.
+ *
+ * @param assignmentSpec The member assignment spec.
+ * @param subscribedTopicDescriber The topic and cluster metadata
describer {@link SubscribedTopicDescriber}.
+ * @return The new assignment for the group.
+ */
+@Override
+public GroupAssignment assign(AssignmentSpec assignmentSpec,
SubscribedTopicDescriber subscribedTopicDescriber) throws
PartitionAssignorException {
+if (allSubscriptionsEqual(assignmentSpec.members())) {
+log.debug("Detected that all members are subscribed to the same
set of topics, invoking the "
++ "optimized assignment algorithm");
+OptimizedUniformAssignor optimizedUniformAssignor = new
OptimizedUniformAssignor(assignmentSpec, subscribedTopicDescriber);
+return optimizedUniformAssignor.build();
+} else {
+GeneralUniformAssignor generalUniformAssignor = new
GeneralUniformAssignor();
+log.debug("Detected that all members are subscribed to a different
set of topics, invoking the "
++ "general assignment algorithm");
+return generalUniformAssignor.build();
+}
+}
+
+/**
+ * Builds the assignment for the given topic and member metadata.
+ *
+ * @return The new assignment for the group.
+ */
+protected abstract GroupAssignment build();
Review Comment:
AssignorBuilder.build() makes sense. Assignor.build() implies that we're
building an assignor, not building an assignment
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AbstractUniformAssignor.java:
##
@@ -0,0 +1,220 @@
+/*
+ * 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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1290813753
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignor.java:
##
@@ -0,0 +1,396 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
previous owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignor extends UniformAssignor {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignor.class);
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+private final RackInfo rackInfo;
+// The minimum required quota that each member needs to meet for a
balanced assignment.
+// This is the same for all members.
+private final int minQuota;
+// Count of members expected to receive an extra partition beyond the
minimum quota,
+// to account for the distribution of the remaining partitions.
+private int expectedNumMembersWithExtraPartition;
+// Map of members to their remaining partitions needed to meet the minimum
quota,
+// including members eligible for an extra partition.
+private final Map potentiallyUnfilledMembers;
+// Members mapped to the number of partitions they still need to meet the
full quota.
+// Full quota = minQuota + one extra partition (if applicable).
+private Map unfilledMembers;
+// Partitions that still need to be assigned.
+private List unassignedPartitions;
+private final Map newAssignment;
+// Tracks the previous owner of each partition when using rack-aware
strategy.
+private final Map partitionToPrevOwner;
Review Comment:
it is implied by the type that the previous owners are keyed by
TopicIdPartition. like how we are using `potentiallyUnfilledMembers`,
`unfilledMembers`, or `newAssignment`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1290813322
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignor.java:
##
@@ -0,0 +1,396 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
previous owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignor extends UniformAssignor {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignor.class);
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final AssignmentSpec assignmentSpec;
+private final SubscribedTopicDescriber subscribedTopicDescriber;
+private final RackInfo rackInfo;
+// The minimum required quota that each member needs to meet for a
balanced assignment.
+// This is the same for all members.
+private final int minQuota;
+// Count of members expected to receive an extra partition beyond the
minimum quota,
+// to account for the distribution of the remaining partitions.
+private int expectedNumMembersWithExtraPartition;
Review Comment:
`numMembersWithExtraPartition` : number of members to get the extra
partition, no?
there is no expected vs. actual though, we always distribute that number
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14182: KAFKA 14515: Optimized Uniform Rack Aware Assignor
jeffkbkim commented on code in PR #14182:
URL: https://github.com/apache/kafka/pull/14182#discussion_r1290585581
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignor.java:
##
@@ -0,0 +1,396 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ *
Review Comment:
I would include the priorities when considering uniform vs. sticky vs.
rack-aware
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/OptimizedUniformAssignor.java:
##
@@ -0,0 +1,396 @@
+/*
+ * 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.apache.kafka.coordinator.group.common.TopicIdPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * Assigns Kafka partitions to members of a consumer group ensuring a balanced
distribution with
+ * considerations for sticky assignments and rack-awareness.
+ *
+ * Here's the step-by-step breakdown of the assignment process:
+ *
+ *
+ * Compute the quotas of partitions for each member based on the
total partitions and member count.
+ * For existing assignments, retain partitions based on the
determined quota and member's rack compatibility.
+ * If a partition's rack mismatches with its member, track it with
its prior owner.
+ * Identify members that haven't fulfilled their partition quota or
are eligible to receive extra partitions.
+ * Derive the unassigned partitions by taking the difference between
total partitions and the sticky assignments.
+ * Depending on members needing extra partitions, select members from
the potentially unfilled list and add them to the unfilled list.
+ * Proceed with a round-robin assignment adhering to rack awareness.
+ * For each unassigned partition, locate the first compatible member
from the unfilled list.
+ * If no rack-compatible member is found, revert to the tracked
previous owner.
+ * If that member can't accommodate the partition due to quota
limits, resort to a generic round-robin assignment.
+ *
+ */
+public class OptimizedUniformAssignor extends UniformAssignor {
+private static final Logger log =
LoggerFactory.getLogger(OptimizedUniformAssignor.class);
+// List of topics subscribed to by all members.
+private final List subscriptionList;
+private final
