[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1178059673
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,258 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * The properties are as follows:
+ *
+ * Each member must get at least one partition from every topic that it
is subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+// Used in the potentiallyUnfilledMembers map and the UnfilledMembers map.
+private static class MemberWithRemainingAssignments {
+private final String memberId;
+/**
+ * Number of partitions required to meet the assignment quota
+ */
+private final Integer remaining;
+
+public MemberWithRemainingAssignments(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+/**
+ * @return memberId
+ */
+public String memberId() {
+return memberId;
+}
+ /**
+ * @return Remaining number of partitions
+ */
+public Integer remaining() {
+return remaining;
+}
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn("Member " + memberId + " subscribed to topic " +
topicId + " which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+/**
+ * The algorithm includes the following steps:
+ *
+ * Generate a map of membersPerTopic using the given member
subscriptions.
Review Comment:
I got comments before to add tags and put the variable names, thats
why I did it
--
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 t
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177178765
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
Review Comment:
not straightforward anymore with the new code, so kept it
--
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 co
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177177698
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java:
##
@@ -0,0 +1,514 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RangeAssignorTest {
Review Comment:
added another test just in case
--
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] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177168484
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177168484
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177168353
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177156668
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
Review Comment:
done
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics in
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177156557
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
Review Comment:
removed, since it was derived from a generic pair class, missed removing it,
thanks for the catch!
--
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] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177139606
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
Review Comment:
this whole function is removed now
--
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] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177136376
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177135296
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177128215
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177128655
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177128215
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177109065
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1177109065
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1175703194
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1175701652
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1175689958
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1175567311
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1175552910
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,261 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the range
assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of partitions
for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ * This can only be done if every member is subscribed to the same topics and
the topics are co-partitioned.
+ * Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much of their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic
+.computeIfAbsent(topicId, k -> new ArrayList<>())
+.add(memberId);
+} else {
+log.warn(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map> getUnassignedPartitionsPerTopic(final
AssignmentSpec assignmentSpec, Map>
assignedStickyPartitionsPerTopic) {
+Map> unassignedPartitionsPerTopic = new
HashMap<>();
+Map topicsMetadata =
assignmentSpec.topics();
+
+topicsMetadata.forEach((topicId, assignmentTopicMetadata) -> {
+ArrayList unassignedPartitionsForTopic = new
ArrayList<>();
+int numPartitions = assignmentTopicMetadata.numPartitions();
+// List of unassigned partitions per topic contains the partitions
in ascending order.
+Set assignedStickyPartitionsForTopic =
assignedStickyPartitionsPerTopic.getOrDefault(topicId, new HashSet<>());
+for (int i = 0; i < numPartitions; i++) {
+if (!assignedStickyPartitionsForTopic.contains(i)
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1173179837
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,240 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the
range assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of
partitions for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ *This can only be done if every member is subscribed to the same topics
and the topics are co-partitioned.
+ *Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much as their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ *
+ *
+ * The algorithm includes the following steps:
+ *
+ * Generate a map of membersPerTopic using the given member
subscriptions.
+ * Generate a list of members (potentiallyUnfilledMembers)
that have not met the minimum required quota for assignment AND
+ * get a list of sticky partitions that we want to retain in the new
assignment.
+ * Add members from the potentiallyUnfilled list to the
Unfilled list if they haven't met the total required quota i.e.
minimum number of partitions per member + 1 (if member is designated to receive
one of the excess partitions)
+ * Generate a list of unassigned partitions by calculating the difference
between total partitions and already assigned (sticky) partitions
+ * Iterate through unfilled members and assign partitions from the
unassigned partitions
+ *
+ *
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic.computeIfAbsent(topicId, k -> new
ArrayList<>()).add(memberId);
+} else {
+log.info(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map>
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1173177108
##
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java:
##
@@ -0,0 +1,240 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static java.lang.Math.min;
+
+/**
+ * The Server Side Sticky Range Assignor inherits properties of both the
range assignor and the sticky assignor.
+ * Properties are as follows:
+ *
+ * Each member must get at least one partition for every topic that it is
subscribed to. The only exception is when
+ * the number of subscribed members is greater than the number of
partitions for that topic. (Range)
+ * Partitions should be assigned to members in a way that facilitates the
join operation when required. (Range)
+ *This can only be done if every member is subscribed to the same topics
and the topics are co-partitioned.
+ *Two streams are co-partitioned if the following conditions are met:
+ *
+ * The keys must have the same schemas.
+ * The topics involved must have the same number of partitions.
+ *
+ * Members should retain as much as their previous assignment as possible
to reduce the number of partition movements during reassignment. (Sticky)
+ *
+ *
+ *
+ * The algorithm includes the following steps:
+ *
+ * Generate a map of membersPerTopic using the given member
subscriptions.
+ * Generate a list of members (potentiallyUnfilledMembers)
that have not met the minimum required quota for assignment AND
+ * get a list of sticky partitions that we want to retain in the new
assignment.
+ * Add members from the potentiallyUnfilled list to the
Unfilled list if they haven't met the total required quota i.e.
minimum number of partitions per member + 1 (if member is designated to receive
one of the excess partitions)
+ * Generate a list of unassigned partitions by calculating the difference
between total partitions and already assigned (sticky) partitions
+ * Iterate through unfilled members and assign partitions from the
unassigned partitions
+ *
+ *
+ *
+ */
+public class RangeAssignor implements PartitionAssignor {
+
+private static final Logger log =
LoggerFactory.getLogger(RangeAssignor.class);
+
+public static final String RANGE_ASSIGNOR_NAME = "range";
+
+@Override
+public String name() {
+return RANGE_ASSIGNOR_NAME;
+}
+
+static class RemainingAssignmentsForMember {
+private final String memberId;
+private final Integer remaining;
+
+public RemainingAssignmentsForMember(String memberId, Integer
remaining) {
+this.memberId = memberId;
+this.remaining = remaining;
+}
+
+public String memberId() {
+return memberId;
+}
+
+public Integer remaining() {
+return remaining;
+}
+
+}
+
+private Map> membersPerTopic(final AssignmentSpec
assignmentSpec) {
+Map> membersPerTopic = new HashMap<>();
+Map membersData =
assignmentSpec.members();
+
+membersData.forEach((memberId, memberMetadata) -> {
+Collection topics = memberMetadata.subscribedTopicIds();
+for (Uuid topicId: topics) {
+// Only topics that are present in both the subscribed topics
list and the topic metadata should be considered for assignment.
+if (assignmentSpec.topics().containsKey(topicId)) {
+membersPerTopic.computeIfAbsent(topicId, k -> new
ArrayList<>()).add(memberId);
+} else {
+log.info(memberId + " subscribed to topic " + topicId + "
which doesn't exist in the topic metadata");
+}
+}
+});
+
+return membersPerTopic;
+}
+
+private Map>
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1173174594
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java:
##
@@ -0,0 +1,416 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RangeAssignorTest {
+private final RangeAssignor assignor = new RangeAssignor();
+private final Uuid topic1Uuid = Uuid.randomUuid();
+private final Uuid topic2Uuid = Uuid.randomUuid();
+private final Uuid topic3Uuid = Uuid.randomUuid();
+private final String consumerA = "A";
+private final String consumerB = "B";
+private final String consumerC = "C";
+
+@Test
+public void testOneConsumerNoTopic() {
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
+Map members = new HashMap<>();
+List subscribedTopics = new ArrayList<>();
+members.computeIfAbsent(consumerA, k -> new
AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopics, new
HashMap<>()));
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec);
+
+assertTrue(groupAssignment.members().isEmpty());
+}
+
+@Test
+public void testOneConsumerNonExistentTopic() {
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
+Map members = new HashMap<>();
+List subscribedTopics = new ArrayList<>();
+subscribedTopics.add(topic2Uuid);
+members.computeIfAbsent(consumerA, k -> new
AssignmentMemberSpec(Optional.empty(), Optional.empty(), subscribedTopics, new
HashMap<>()));
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec);
+
+assertTrue(groupAssignment.members().isEmpty());
+}
+
+@Test
+public void testFirstAssignmentTwoConsumersTwoTopicsSameSubscriptions() {
+// A -> T1, T3 // B -> T1, T3 // T1 -> 3 Partitions // T3 -> 2
Partitions
+// Topics
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
+topics.put(topic3Uuid, new AssignmentTopicMetadata(2));
+// Members
+Map members = new HashMap<>();
+// Consumer A
+List subscribedTopicsA = new
ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+members.put(consumerA, new AssignmentMemberSpec(Optional.empty(),
Optional.empty(), subscribedTopicsA, new HashMap<>()));
+// Consumer B
+List subscribedTopicsB = new
ArrayList<>(Arrays.asList(topic1Uuid, topic3Uuid));
+members.put(consumerB, new AssignmentMemberSpec(Optional.empty(),
Optional.empty(), subscribedTopicsB, new HashMap<>()));
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+GroupAssignment computedAssignment = assignor.assign(assignmentSpec);
+
+Map>> expectedAssignment = new HashMap<>();
+// Topic 1 Partitions Assignment
+expectedAssignment.computeIfAbsent(topic1Uuid, k -> new
HashSet<>()).add(new HashSet<>(Arrays.asList(0, 1)));
+expectedAssignment.computeIfAbsent(topic1Uuid, k -> new
HashSet<>()).add(new HashSet<>(Collections.singletonList(2)));
+// Topic 3 Partitions Assignment
+expectedAssignment.computeIfAbsent(topic3Uuid, k -> new
HashSet<>()).add(new HashSet<>(Collections.singletonList(0)));
+expectedAssignment.computeIfAbsent(topic3Uuid, k -> new
HashSet<>()
[GitHub] [kafka] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1173172297
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java:
##
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RangeAssignorTest {
+private final RangeAssignor assignor = new RangeAssignor();
+private final Uuid topic1Uuid = Uuid.randomUuid();
+private final Uuid topic2Uuid = Uuid.randomUuid();
+private final Uuid topic3Uuid = Uuid.randomUuid();
+private final String consumerA = "A";
+private final String consumerB = "B";
+private final String consumerC = "C";
+
+@Test
+public void testOneConsumerNoTopic() {
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
+Map members = new HashMap<>();
+members.put(consumerA, new AssignmentMemberSpec(
+Optional.empty(),
+ Optional.empty(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
+
+AssignmentSpec assignmentSpec = new AssignmentSpec(members, topics);
+GroupAssignment groupAssignment = assignor.assign(assignmentSpec);
+
+assertTrue(groupAssignment.members().isEmpty());
+}
+
+@Test
+public void testOneConsumerNonExistentTopic() {
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
Review Comment:
done
--
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] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443: URL: https://github.com/apache/kafka/pull/13443#discussion_r1173050527 ## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/RangeAssignor.java: ## @@ -0,0 +1,240 @@ +/* + * 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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.lang.Math.min; + +/** + * The Server Side Sticky Range Assignor inherits properties of both the range assignor and the sticky assignor. Review Comment: okay -- 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] rreddy-22 commented on a diff in pull request #13443: KAFKA-14514: Add Sticky Range Assignor on the Server (KIP-848)
rreddy-22 commented on code in PR #13443:
URL: https://github.com/apache/kafka/pull/13443#discussion_r1172824513
##
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/RangeAssignorTest.java:
##
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.coordinator.group.assignor;
+
+import org.apache.kafka.common.Uuid;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class RangeAssignorTest {
+private final RangeAssignor assignor = new RangeAssignor();
+private final Uuid topic1Uuid = Uuid.randomUuid();
+private final Uuid topic2Uuid = Uuid.randomUuid();
+private final Uuid topic3Uuid = Uuid.randomUuid();
+private final String consumerA = "A";
+private final String consumerB = "B";
+private final String consumerC = "C";
+
+@Test
+public void testOneConsumerNoTopic() {
+Map topics = new HashMap<>();
+topics.put(topic1Uuid, new AssignmentTopicMetadata(3));
+Map members = new HashMap<>();
+members.put(consumerA, new AssignmentMemberSpec(
+Optional.empty(),
+ Optional.empty(),
+ Collections.emptyList(),
+ Collections.emptyMap()));
Review Comment:
I had changed it in my IDE and it looked good, idky the formatting changed
in the PR :( Will take a look at it thanks!
--
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
