(kafka) branch trunk updated: MINOR: Add Uniform assignor to the default config (#14826)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 34e1dbbaba1 MINOR: Add Uniform assignor to the default config (#14826)
34e1dbbaba1 is described below

commit 34e1dbbaba15e8f053f68e998bb21e1ed62bdaf9
Author: David Jacot <dj...@confluent.io>
AuthorDate: Tue Dec 5 09:32:50 2023 +0100

    MINOR: Add Uniform assignor to the default config (#14826)
    
    This patch adds the `Uniform` assignor to the default list of supported assignors. It also do small changes in the code.
    
    Reviewers: Justine Olshan <jo...@confluent.io>
---
 core/src/main/scala/kafka/server/KafkaConfig.scala                   | 4 ++--
 .../org/apache/kafka/coordinator/group/GroupMetadataManager.java     | 5 +++--
 .../org/apache/kafka/coordinator/group/assignor/UniformAssignor.java | 3 ++-
 3 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 603e8c90b26..8aef3cab22b 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -42,7 +42,7 @@ import org.apache.kafka.common.security.auth.KafkaPrincipalSerde
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
 import org.apache.kafka.common.utils.Utils
-import org.apache.kafka.coordinator.group.assignor.{PartitionAssignor, RangeAssignor}
+import org.apache.kafka.coordinator.group.assignor.{PartitionAssignor, RangeAssignor, UniformAssignor}
 import org.apache.kafka.raft.RaftConfig
 import org.apache.kafka.server.authorizer.Authorizer
 import org.apache.kafka.server.common.{MetadataVersion, MetadataVersionValidator}
@@ -176,7 +176,7 @@ object Defaults {
   val ConsumerGroupMinHeartbeatIntervalMs = 5000
   val ConsumerGroupMaxHeartbeatIntervalMs = 15000
   val ConsumerGroupMaxSize = Int.MaxValue
-  val ConsumerGroupAssignors = List(classOf[RangeAssignor].getName).asJava
+  val ConsumerGroupAssignors = List(classOf[UniformAssignor].getName, classOf[RangeAssignor].getName).asJava
 
   /** ********* Offset management configuration ***********/
   val OffsetMetadataMaxSize = OffsetConfig.DefaultMaxMetadataSize
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 1de63d714f9..3f47b49240f 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -1109,8 +1109,9 @@ public class GroupMetadataManager {
                     assignmentResult = assignmentResultBuilder
                         .build();
                 }
-                log.info("[GroupId {}] Computed a new target assignment for epoch {}: {}.",
-                    groupId, groupEpoch, assignmentResult.targetAssignment());
+
+                log.info("[GroupId {}] Computed a new target assignment for epoch {} with '{}' assignor: {}.",
+                    groupId, groupEpoch, preferredServerAssignor, assignmentResult.targetAssignment());
 
                 records.addAll(assignmentResult.records());
                 targetAssignment = assignmentResult.targetAssignment().get(memberId);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java
index 1dc34fd2cb3..228acc688c5 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/assignor/UniformAssignor.java
@@ -96,7 +96,8 @@ public class UniformAssignor implements PartitionAssignor {
     private boolean allSubscriptionsEqual(Map<String, AssignmentMemberSpec> members) {
         Set<Uuid> firstSubscriptionSet = new HashSet<>(members.values().iterator().next().subscribedTopicIds());
         for (AssignmentMemberSpec memberSpec : members.values()) {
-            if (!firstSubscriptionSet.equals(new HashSet<>(memberSpec.subscribedTopicIds()))) {
+            if (firstSubscriptionSet.size() != memberSpec.subscribedTopicIds().size() ||
+                !firstSubscriptionSet.containsAll(memberSpec.subscribedTopicIds())) {
                 return false;
             }
         }