KAFKA-14514: Add Range Assignor on the Server (KIP-848)

checkstyle/suppressions.xml

@@ -320,6 +320,14 @@
     <suppress checks="AvoidStarImport"
               files="MetadataVersionTest.java"/>
 
+    <!-- group coordinator  -->
+    <suppress checks="CyclomaticComplexity"
+              files="ServerSideStickyRangeAssignor.java"/>
+    <suppress checks="NPathComplexity"
+              files="ServerSideStickyRangeAssignor.java"/>
+    <suppress checks="AvoidStarImport"
+              files="ServerSideStickyRangeAssignorTest.java"/>
+
     <!-- storage -->
     <suppress checks="CyclomaticComplexity"
               files="(LogValidator|RemoteLogManagerConfig).java"/>

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/AssignmentMemberSpec.java

@@ -16,11 +16,13 @@
  */
 package org.apache.kafka.coordinator.group.assignor;
 
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 
-import java.util.Collection;
-import java.util.Objects;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.Optional;
+import java.util.Objects;
 
 /**
  * The assignment specification for a consumer group member.
@@ -37,29 +39,28 @@ public class AssignmentMemberSpec {
     final Optional<String> rackId;
 
     /**
-     * The topics that the member is subscribed to.
+     * The topicIds of topics that the member is subscribed to.
      */
-    final Collection<String> subscribedTopics;
+    final List<Uuid> subscribedTopics;
 
     /**
-     * The current target partitions of the member.
+     * Maps the partitions assigned for this member per topicId
      */
-    final Collection<TopicPartition> targetPartitions;
+    final Map<Uuid, Set<Integer>> currentAssignmentPerTopic;
 
     public AssignmentMemberSpec(
         Optional<String> instanceId,
         Optional<String> rackId,
-        Collection<String> subscribedTopics,
-        Collection<TopicPartition> targetPartitions
+        List<Uuid> subscribedTopics,
+        Map<Uuid, Set<Integer>> currentAssignmentPerTopic
     ) {
         Objects.requireNonNull(instanceId);
         Objects.requireNonNull(rackId);
         Objects.requireNonNull(subscribedTopics);
-        Objects.requireNonNull(targetPartitions);
         this.instanceId = instanceId;
         this.rackId = rackId;
         this.subscribedTopics = subscribedTopics;
-        this.targetPartitions = targetPartitions;
+        this.currentAssignmentPerTopic = currentAssignmentPerTopic;
     }
 
     @Override
@@ -72,15 +73,15 @@ public boolean equals(Object o) {
         if (!instanceId.equals(that.instanceId)) return false;
         if (!rackId.equals(that.rackId)) return false;
         if (!subscribedTopics.equals(that.subscribedTopics)) return false;
-        return targetPartitions.equals(that.targetPartitions);
+        return currentAssignmentPerTopic.equals(that.currentAssignmentPerTopic);
     }
 
     @Override
     public int hashCode() {
         int result = instanceId.hashCode();
         result = 31 * result + rackId.hashCode();
         result = 31 * result + subscribedTopics.hashCode();
-        result = 31 * result + targetPartitions.hashCode();
+        result = 31 * result + currentAssignmentPerTopic.hashCode();
         return result;
     }
 
@@ -89,7 +90,7 @@ public String toString() {
         return "AssignmentMemberSpec(instanceId=" + instanceId +
             ", rackId=" + rackId +
             ", subscribedTopics=" + subscribedTopics +
-            ", targetPartitions=" + targetPartitions +
+            ", currentAssignment=" + currentAssignmentPerTopic +
             ')';
     }
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/GroupAssignment.java

@@ -35,6 +35,10 @@ public GroupAssignment(
         this.members = members;
     }
 
+    public Map<String, MemberAssignment> getMembers() {
+        return this.members;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/MemberAssignment.java

@@ -16,25 +16,28 @@
  */
 package org.apache.kafka.coordinator.group.assignor;
 
-import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
 
-import java.util.Collection;
+import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
+
 
 /**
  * The partition assignment for a consumer group member.
  */
 public class MemberAssignment {
-    /**
-     * The target partitions assigned to this member.
-     */
-    final Collection<TopicPartition> targetPartitions;
-
-    public MemberAssignment(
-        Collection<TopicPartition> targetPartitions
-    ) {
-        Objects.requireNonNull(targetPartitions);
-        this.targetPartitions = targetPartitions;
+
+    private final Map<Uuid, Set<Integer>> assignmentPerTopic;
+
+
+    public MemberAssignment(Map<Uuid, Set<Integer>> assignmentPerTopic) {
+        Objects.requireNonNull(assignmentPerTopic);
+        this.assignmentPerTopic = assignmentPerTopic;
+    }
+
+    public Map<Uuid, Set<Integer>> getAssignmentPerTopic() {
+        return this.assignmentPerTopic;
     }
 
     @Override
@@ -44,16 +47,16 @@ public boolean equals(Object o) {
 
         MemberAssignment that = (MemberAssignment) o;
 
-        return targetPartitions.equals(that.targetPartitions);
+        return assignmentPerTopic.equals(that.assignmentPerTopic);
     }
 
     @Override
     public int hashCode() {
-        return targetPartitions.hashCode();
+        return assignmentPerTopic.hashCode();
     }
 
     @Override
     public String toString() {
-        return "MemberAssignment(targetPartitions=" + targetPartitions + ')';
+        return "MemberAssignment(targetPartitions=" + assignmentPerTopic + ')';
     }
 }