KAFKA-20790: add AssignmentConfigs interface

group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/AssignmentConfigs.java

@@ -0,0 +1,41 @@
+/*
+ * 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.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.List;
+
+/**
+ * The assignment configurations that the group coordinator passes to the task assignor.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface AssignmentConfigs {
+
+    /**
+     * @return The number of standby replicas for each task.
+     */
+    int numStandbyReplicas();
+
+    /**
+     * @return The client tags used to distribute standby tasks across racks. The list is unmodifiable.
+     */
+    List<String> rackAwareAssignmentTags();
+
+}

group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupSpec.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 import java.util.Collection;
-import java.util.Map;
 
 /**
  * The group metadata specifications required to compute the target assignment.
@@ -51,8 +50,8 @@ public interface GroupSpec {
     MemberAssignmentState memberAssignmentState(String memberId);
 
     /**
-     * @return Any configurations passed to the assignor. The map is unmodifiable.
+     * @return The assignment configurations passed to the assignor.
      */
-    Map<String, String> configs();
+    AssignmentConfigs configs();
 
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java

@@ -19,10 +19,12 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
 import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
 import org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment;
 import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignment;
 import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
 import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignorException;
+import org.apache.kafka.coordinator.group.streams.assignor.AssignmentConfigsImpl;
 import org.apache.kafka.coordinator.group.streams.assignor.GroupSpecImpl;
 import org.apache.kafka.coordinator.group.streams.assignor.MemberMetadataAndStateImpl;
 import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology;
@@ -70,7 +72,7 @@ public class TargetAssignmentBuilder {
     /**
      * The assignment configs.
      */
-    private final Map<String, String> assignmentConfigs;
+    private final AssignmentConfigs assignmentConfigs;
 
     /**
      * The members in the group.
@@ -114,7 +116,22 @@ public TargetAssignmentBuilder(
         this.groupId = Objects.requireNonNull(groupId);
         this.groupEpoch = groupEpoch;
         this.assignor = Objects.requireNonNull(assignor);
-        this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
+        this.assignmentConfigs = toAssignmentConfigs(Objects.requireNonNull(assignmentConfigs));
+    }
+
+    /**
+     * Converts the raw assignment configs computed for the group into the typed configs passed to the assignor.
+     */
+    private static AssignmentConfigs toAssignmentConfigs(Map<String, String> assignmentConfigs) {
+        // Both configs can be absent: the rack-aware assignment tags are only set when any are configured, and the
+        // whole map is empty when it was replayed from a group metadata record written before the last assignment
+        // configs were persisted.
+        String numStandbyReplicas = assignmentConfigs.get("num.standby.replicas");
+        String rackAwareAssignmentTags = assignmentConfigs.get("rack.aware.assignment.tags");
+        return new AssignmentConfigsImpl(
+            numStandbyReplicas == null ? 0 : Integer.parseInt(numStandbyReplicas),
+            rackAwareAssignmentTags == null ? List.of() : List.of(rackAwareAssignmentTags.trim().split("\\s*,\\s*", -1))
+        );
     }
 
     static MemberMetadataAndStateImpl createMemberMetadataAndState(

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

@@ -0,0 +1,39 @@
+/*
+ * 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.streams.assignor;
+
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas      The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+    int numStandbyReplicas,
+    List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+    public AssignmentConfigsImpl {
+        // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+        rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+    }
+}

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.coordinator.group.streams.assignor;
 
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
 import org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec;
 import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentMetadata;
 import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentState;
@@ -30,17 +31,17 @@
  *
  * @param members The member metadata keyed by member Id. Each value provides both the
  *                {@link MemberAssignmentMetadata} and the {@link MemberAssignmentState} for the member.
- * @param configs Any configurations passed to the assignor.
+ * @param configs The assignment configurations passed to the assignor.
  */
 public record GroupSpecImpl(
     Map<String, MemberMetadataAndStateImpl> members,
-    Map<String, String> configs
+    AssignmentConfigs configs
 ) implements GroupSpec {
 
     public GroupSpecImpl {
-        // Both maps are exposed to a custom assignor through the public GroupSpec interface.
+        // The map is exposed to a custom assignor through the public GroupSpec interface.
         members = Collections.unmodifiableMap(Objects.requireNonNull(members));
-        configs = Collections.unmodifiableMap(Objects.requireNonNull(configs));
+        Objects.requireNonNull(configs);
     }
 
     @Override

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

@@ -95,9 +95,7 @@ private static LinkedList<TaskId> taskIds(final TopologyDescriber topologyDescri
 
     private static LocalState initialize(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
         final LocalState localState = new LocalState();
-        localState.numStandbyReplicas =
-            groupSpec.configs().isEmpty() ? 0
-                : Integer.parseInt(groupSpec.configs().get("num.standby.replicas"));
+        localState.numStandbyReplicas = groupSpec.configs().numStandbyReplicas();
 
         // Helpers for computing active tasks per member, and tasks per member
         localState.totalActiveTasks = 0;

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java

@@ -154,6 +154,7 @@
 import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole;
 import org.apache.kafka.coordinator.group.streams.TasksTuple;
 import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
+import org.apache.kafka.coordinator.group.streams.assignor.AssignmentConfigsImpl;
 import org.apache.kafka.image.MetadataDelta;
 import org.apache.kafka.image.MetadataImage;
 import org.apache.kafka.image.MetadataProvenance;
@@ -23654,7 +23655,7 @@ public void testStreamsGroupDynamicConfigs() {
                     .setWarmupTasks(List.of()));
         assertEquals(2, result.response().data().memberEpoch());
         assertEquals(
-            getDefaultAssignmentConfigs(),
+            new AssignmentConfigsImpl(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT, List.of()),
             assignor.lastPassedAssignmentConfigs()
         );
 
@@ -23693,7 +23694,7 @@ public void testStreamsGroupDynamicConfigs() {
 
         // Verify that the new number of standby replicas is used
         assertEquals(
-            Map.of("num.standby.replicas", "2"),
+            new AssignmentConfigsImpl(2, List.of()),
             assignor.lastPassedAssignmentConfigs()
         );
 
@@ -24041,7 +24042,7 @@ public void testStreamsGroupEvaluatedConfigs() {
         context.assertSessionTimeout(groupId, memberId,
             GroupCoordinatorConfig.STREAMS_GROUP_SESSION_TIMEOUT_MS_DEFAULT);
         assertEquals(
-            getDefaultAssignmentConfigs(),
+            new AssignmentConfigsImpl(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT, List.of()),
             assignor.lastPassedAssignmentConfigs());
         assertEquals(GroupCoordinatorConfig.STREAMS_GROUP_TASK_OFFSET_INTERVAL_MS_DEFAULT,
             result.response().data().taskOffsetIntervalMs());
@@ -24081,7 +24082,7 @@ public void testStreamsGroupEvaluatedConfigs() {
         // Verify that the number of standby replicas is evaluated to max,
         // and task offset interval is evaluated to min
         assertEquals(
-            Map.of("num.standby.replicas", String.valueOf(GroupCoordinatorConfig.STREAMS_GROUP_MAX_STANDBY_REPLICAS_DEFAULT)),
+            new AssignmentConfigsImpl(GroupCoordinatorConfig.STREAMS_GROUP_MAX_STANDBY_REPLICAS_DEFAULT, List.of()),
             assignor.lastPassedAssignmentConfigs());
         assertEquals(GroupCoordinatorConfig.STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_DEFAULT,
             result.response().data().taskOffsetIntervalMs());

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.coordinator.group.streams;
 
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
 import org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment;
 import org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec;
 import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignment;
@@ -31,7 +32,7 @@ public class MockTaskAssignor implements TaskAssignor {
 
     private final String name;
     private GroupAssignment preparedGroupAssignment = null;
-    private Map<String, String> assignmentConfigs = Map.of();
+    private AssignmentConfigs assignmentConfigs = null;
 
     public MockTaskAssignor(String name) {
         this.name = name;
@@ -53,7 +54,7 @@ public void prepareGroupAssignment(Map<String, TasksTuple> memberAssignments) {
                     })));
     }
 
-    public Map<String, String> lastPassedAssignmentConfigs() {
+    public AssignmentConfigs lastPassedAssignmentConfigs() {
         return assignmentConfigs;
     }
 

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java

@@ -27,6 +27,7 @@
 import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
 import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
 import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole;
+import org.apache.kafka.coordinator.group.streams.assignor.AssignmentConfigsImpl;
 import org.apache.kafka.coordinator.group.streams.assignor.GroupSpecImpl;
 import org.apache.kafka.coordinator.group.streams.assignor.MemberMetadataAndStateImpl;
 import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
@@ -479,7 +480,7 @@ public org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.Target
             TopologyMetadata topologyMetadata = new TopologyMetadata(metadataImage, subtopologies);
 
             // Prepare the expected assignment spec.
-            GroupSpecImpl groupSpec = new GroupSpecImpl(memberMetadataMap, new HashMap<>());
+            GroupSpecImpl groupSpec = new GroupSpecImpl(memberMetadataMap, new AssignmentConfigsImpl(0, List.of()));
 
             // We use `any` here to always return an assignment but use `verify` later on
             // to ensure that the input was correct.

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImplTest.java

@@ -19,13 +19,14 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 
 public class GroupSpecImplTest {
@@ -53,7 +54,7 @@ void setUp() {
 
         groupSpec = new GroupSpecImpl(
             members,
-            new HashMap<>()
+            new AssignmentConfigsImpl(2, new ArrayList<>(List.of("test-tag")))
         );
     }
 
@@ -80,13 +81,14 @@ void testMemberNotFound() {
 
     @Test
     void testConfigs() {
-        assertTrue(groupSpec.configs().isEmpty());
+        assertEquals(2, groupSpec.configs().numStandbyReplicas());
+        assertEquals(List.of("test-tag"), groupSpec.configs().rackAwareAssignmentTags());
     }
 
     @Test
     void testMembersAndConfigsAreUnmodifiable() {
         assertThrows(UnsupportedOperationException.class, () -> groupSpec.members().put("other-member", member));
-        assertThrows(UnsupportedOperationException.class, () -> groupSpec.configs().put("key", "value"));
+        assertThrows(UnsupportedOperationException.class, () -> groupSpec.configs().rackAwareAssignmentTags().add("other-tag"));
     }
 
 }

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java

@@ -23,7 +23,6 @@
 
 import org.junit.jupiter.api.Test;
 
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -54,7 +53,7 @@ public void testZeroMembers() {
         TaskAssignorException ex = assertThrows(TaskAssignorException.class, () -> assignor.assign(
             new GroupSpecImpl(
                 Map.of(),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(5, List.of("test-subtopology"))
         ));
@@ -92,7 +91,7 @@ public void testDoubleAssignment() {
         TaskAssignorException ex = assertThrows(TaskAssignorException.class, () -> assignor.assign(
             new GroupSpecImpl(
                 Map.of("member1", memberMetadata1, "member2", memberMetadata2),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(5, List.of("test-subtopology"))
         ));
@@ -106,7 +105,7 @@ public void testBasicScenario() {
         final GroupAssignment result = assignor.assign(
             new GroupSpecImpl(
                 Map.of(),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(5, List.of())
         );
@@ -133,7 +132,7 @@ public void testSingleMember() {
         final GroupAssignment result = assignor.assign(
             new GroupSpecImpl(
                 Map.of("test_member", memberMetadata),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(4, List.of("test-subtopology"))
         );
@@ -177,7 +176,7 @@ public void testTwoMembersTwoSubtopologies() {
         final GroupAssignment result = assignor.assign(
             new GroupSpecImpl(
                 mkMap(mkEntry("test_member1", memberMetadata1), mkEntry("test_member2", memberMetadata2)),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(4, List.of("test-subtopology1", "test-subtopology2"))
         );
@@ -235,7 +234,7 @@ public void testTwoMembersTwoSubtopologiesStickiness() {
         final GroupAssignment result = assignor.assign(
             new GroupSpecImpl(
                 mkMap(mkEntry("test_member1", memberMetadata1), mkEntry("test_member2", memberMetadata2)),
-                new HashMap<>()
+                new AssignmentConfigsImpl(0, List.of())
             ),
             new TopologyDescriberImpl(4, List.of("test-subtopology1", "test-subtopology2"))
         );