You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2026/07/24 18:07:01 UTC
(kafka) branch trunk updated: KAFKA-20790: Promote streams assignor API to public module (#22788)
This is an automated email from the ASF dual-hosted git repository.
mjsax 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 e51258d9e8f KAFKA-20790: Promote streams assignor API to public module (#22788)
e51258d9e8f is described below
commit e51258d9e8fe43671fc1134d024015792a42fec8
Author: gabriellafu <10...@users.noreply.github.com>
AuthorDate: Fri Jul 24 14:06:53 2026 -0400
KAFKA-20790: Promote streams assignor API to public module (#22788)
Part of KIP-1357.
This PR adds the newly defined public API as defined in KIP-1357 to
support customer broker side "streams" task assignors. It also
refactors the existing runtime code and StickyTaskAssignor to
use/implement the new public APIs.
`org.apache.kafka.coordinator.group.api.assignor.streams` (in
`group-coordinator-api`), all annotated with `@InterfaceAudience.Public`
and `@InterfaceStability.Evolving`.
Also fixes a latent bug: the old code incorrectly passed the old target
assignment into the assignor. This PR now ensures that we correctly pass
the current assignment instead.
Reviewers: Matthias J. Sax <ma...@confluent.io>, Sean Quah
<sq...@confluent.io>
---
.../api/streams/assignor/GroupAssignment.java | 66 +++
.../group/api}/streams/assignor/GroupSpec.java | 30 +-
.../api/streams/assignor/MemberAssignment.java | 78 ++++
.../streams/assignor/MemberAssignmentMetadata.java | 56 +++
.../streams/assignor/MemberAssignmentState.java | 60 +++
.../group/api}/streams/assignor/TaskAssignor.java | 11 +-
.../streams/assignor/TaskAssignorException.java | 10 +-
.../api}/streams/assignor/TopologyDescriber.java | 11 +-
.../group/api/streams/assignor/package-info.java} | 16 +-
.../coordinator/group/GroupMetadataManager.java | 4 +-
.../group/streams/MemberTaskOffsets.java | 19 +-
.../group/streams/StreamsGroupMember.java | 19 +-
.../group/streams/TargetAssignmentBuilder.java | 58 ++-
.../group/streams/TopologyMetadata.java | 4 +-
.../group/streams/assignor/GroupSpecImpl.java | 41 +-
.../group/streams/assignor/MemberAssignment.java | 43 --
...erSpec.java => MemberMetadataAndStateImpl.java} | 50 ++-
.../group/streams/assignor/MockAssignor.java | 16 +-
.../group/streams/assignor/ProcessState.java | 2 +
.../group/streams/assignor/StickyTaskAssignor.java | 34 +-
.../coordinator/group/streams/assignor/TaskId.java | 2 +-
.../group/GroupMetadataManagerTest.java | 21 +-
.../group/GroupMetadataManagerTestContext.java | 2 +-
.../group/streams/MemberTaskOffsetsTest.java | 33 +-
.../group/streams/MockTaskAssignor.java | 16 +-
.../group/streams/StreamsGroupMemberTest.java | 23 +-
.../group/streams/StreamsGroupTest.java | 17 +-
.../group/streams/TargetAssignmentBuilderTest.java | 105 +++--
.../group/streams/assignor/GroupSpecImplTest.java | 37 +-
.../group/streams/assignor/MockAssignorTest.java | 49 ++-
.../streams/assignor/StickyTaskAssignorTest.java | 449 +++++++++++----------
.../assignor/StreamsAssignorBenchmarkUtils.java | 12 +-
.../assignor/StreamsStickyAssignorBenchmark.java | 31 +-
33 files changed, 901 insertions(+), 524 deletions(-)
diff --git a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupAssignment.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupAssignment.java
new file mode 100644
index 00000000000..de4f609a133
--- /dev/null
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupAssignment.java
@@ -0,0 +1,66 @@
+/*
+ * 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.Map;
+import java.util.Objects;
+
+/**
+ * The task assignment for a streams group.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class GroupAssignment {
+ /**
+ * The member assignments keyed by member ID.
+ */
+ private final Map<String, MemberAssignment> members;
+
+ public GroupAssignment(
+ Map<String, MemberAssignment> members
+ ) {
+ this.members = Objects.requireNonNull(members);
+ }
+
+ /**
+ * @return The member assignments keyed by member ID.
+ */
+ public Map<String, MemberAssignment> members() {
+ return members;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ GroupAssignment that = (GroupAssignment) o;
+ return members.equals(that.members);
+ }
+
+ @Override
+ public int hashCode() {
+ return members.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "GroupAssignment(members=" + members + ')';
+ }
+}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpec.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupSpec.java
similarity index 54%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpec.java
rename to group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupSpec.java
index 1a8e7edc01c..486b2f8e179 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpec.java
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupSpec.java
@@ -14,23 +14,45 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.coordinator.group.streams.assignor;
+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.Collection;
import java.util.Map;
/**
* The group metadata specifications required to compute the target assignment.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public interface GroupSpec {
/**
- * @return Member metadata keyed by member Id.
+ * @return The member Ids of all members in the group.
+ */
+ Collection<String> memberIds();
+
+ /**
+ * Gets the static metadata for a given member.
+ *
+ * @param memberId The member Id.
+ * @return The static member metadata.
+ */
+ MemberAssignmentMetadata memberMetadata(String memberId);
+
+ /**
+ * Gets the current assignment state for a given member.
+ *
+ * @param memberId The member Id.
+ * @return The current member assignment state.
*/
- Map<String, AssignmentMemberSpec> members();
+ MemberAssignmentState memberAssignmentState(String memberId);
/**
* @return Any configurations passed to the assignor.
*/
- Map<String, String> assignmentConfigs();
+ Map<String, String> configs();
}
diff --git a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignment.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignment.java
new file mode 100644
index 00000000000..2ed2bd51dbb
--- /dev/null
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignment.java
@@ -0,0 +1,78 @@
+/*
+ * 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.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * The task assignment for a streams group member.
+ *
+ * <p>Only active and standby tasks are assigned by the {@link TaskAssignor}. Warm-up tasks are
+ * not assigned by the assignor; they are decided during reconciliation.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class MemberAssignment {
+
+ private final Map<String, Set<Integer>> activeTasks;
+ private final Map<String, Set<Integer>> standbyTasks;
+
+ public MemberAssignment(
+ Map<String, Set<Integer>> activeTasks,
+ Map<String, Set<Integer>> standbyTasks
+ ) {
+ this.activeTasks = Objects.requireNonNull(activeTasks);
+ this.standbyTasks = Objects.requireNonNull(standbyTasks);
+ }
+
+ /**
+ * @return The active tasks assigned to this member keyed by subtopology Id.
+ */
+ public Map<String, Set<Integer>> activeTasks() {
+ return activeTasks;
+ }
+
+ /**
+ * @return The standby tasks assigned to this member keyed by subtopology Id.
+ */
+ public Map<String, Set<Integer>> standbyTasks() {
+ return standbyTasks;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MemberAssignment that = (MemberAssignment) o;
+ return activeTasks.equals(that.activeTasks) && standbyTasks.equals(that.standbyTasks);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(activeTasks, standbyTasks);
+ }
+
+ @Override
+ public String toString() {
+ return "MemberAssignment(activeTasks=" + activeTasks + ", standbyTasks=" + standbyTasks + ')';
+ }
+}
diff --git a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentMetadata.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentMetadata.java
new file mode 100644
index 00000000000..ef725075d6d
--- /dev/null
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentMetadata.java
@@ -0,0 +1,56 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+/**
+ * Interface representing the static metadata for a streams group member.
+ *
+ * <p>The metadata contains the per-member information that does not change during the assignment
+ * computation. The member's current task assignment state is exposed separately through
+ * {@link MemberAssignmentState}.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface MemberAssignmentMetadata {
+
+ /**
+ * @return The instance ID if provided.
+ */
+ Optional<String> instanceId();
+
+ /**
+ * @return The rack ID if provided.
+ */
+ Optional<String> rackId();
+
+ /**
+ * @return The process ID.
+ */
+ String processId();
+
+ /**
+ * @return The client tags for a rack-aware assignment.
+ */
+ Map<String, String> clientTags();
+
+}
diff --git a/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java
new file mode 100644
index 00000000000..df0154d994e
--- /dev/null
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/MemberAssignmentState.java
@@ -0,0 +1,60 @@
+/*
+ * 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.Map;
+import java.util.Set;
+
+/**
+ * The active, standby, and warm-up tasks that a streams group member currently has,
+ * used by the {@link TaskAssignor} to compute a new target assignment.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface MemberAssignmentState {
+
+ /**
+ * @return The member's current active tasks keyed by subtopology Id.
+ */
+ Map<String, Set<Integer>> activeTasks();
+
+ /**
+ * @return The member's current standby tasks keyed by subtopology Id.
+ */
+ Map<String, Set<Integer>> standbyTasks();
+
+ /**
+ * @return The tasks the member is currently warming up, keyed by subtopology Id.
+ */
+ Map<String, Set<Integer>> warmupTasks();
+
+ /**
+ * @return The last received cumulative task offsets of assigned tasks or dormant tasks.
+ * The outer map is keyed by subtopology ID and the inner map by partition.
+ */
+ Map<String, Map<Integer, Long>> taskOffsets();
+
+ /**
+ * @return The last received cumulative task end offsets of assigned tasks or dormant tasks.
+ * The outer map is keyed by subtopology ID and the inner map by partition.
+ */
+ Map<String, Map<Integer, Long>> taskEndOffsets();
+
+}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignor.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignor.java
similarity index 77%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignor.java
rename to group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignor.java
index 73bff075ca6..f6532b8c840 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignor.java
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignor.java
@@ -14,15 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.coordinator.group.streams.assignor;
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
/**
* Server side task assignor used by streams groups.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public interface TaskAssignor {
/**
- * Unique name for this assignor.
+ * Unique name for this assignor. Used in configuration to select this assignor.
*/
String name();
@@ -33,7 +38,7 @@ public interface TaskAssignor {
* @param topologyDescriber The task metadata describer.
* @return The new assignment for the group.
*
- * @throws TaskAssignorException For empty groups
+ * @throws TaskAssignorException If the assignment cannot be computed.
*/
GroupAssignment assign(
GroupSpec groupSpec,
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignorException.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignorException.java
similarity index 73%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignorException.java
rename to group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignorException.java
index 2fda6a9e9ec..872ab7825c0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskAssignorException.java
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TaskAssignorException.java
@@ -14,13 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.coordinator.group.streams.assignor;
+package org.apache.kafka.coordinator.group.api.streams.assignor;
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.errors.ApiException;
/**
- * Exception thrown by {@link TaskAssignor#assign(GroupSpec, TopologyDescriber)}}. The exception is only used internally.
+ * Exception thrown by {@link TaskAssignor#assign(GroupSpec, TopologyDescriber)} when the group's tasks
+ * cannot be assigned. Custom {@link TaskAssignor} implementations should throw this exception to signal
+ * an assignment failure.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public class TaskAssignorException extends ApiException {
public TaskAssignorException(String message) {
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TopologyDescriber.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TopologyDescriber.java
similarity index 85%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TopologyDescriber.java
rename to group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TopologyDescriber.java
index d6f7a3ab579..326de177550 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TopologyDescriber.java
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/TopologyDescriber.java
@@ -14,7 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.coordinator.group.streams.assignor;
+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;
import java.util.NoSuchElementException;
@@ -22,12 +25,14 @@ import java.util.NoSuchElementException;
/**
* The topology describer is used by the {@link TaskAssignor} to get topic and task metadata of the group's topology.
*/
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
public interface TopologyDescriber {
/**
- * Map of topic names to topic metadata.
+ * The IDs of all subtopologies in the group's topology.
*
- * @return The list of subtopologies IDs.
+ * @return The list of subtopology IDs.
*/
List<String> subtopologies();
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupAssignment.java b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/package-info.java
similarity index 68%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupAssignment.java
rename to group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/package-info.java
index c4ac5803b35..b8c31e7e537 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupAssignment.java
+++ b/group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/package-info.java
@@ -14,20 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.kafka.coordinator.group.streams.assignor;
-
-import java.util.Map;
-import java.util.Objects;
/**
- * The task assignment for a streams group.
- *
- * @param members The member assignments keyed by member ID.
+ * Provides the public API for broker-side custom task assignors used by Kafka Streams groups.
*/
-public record GroupAssignment(Map<String, MemberAssignment> members) {
-
- public GroupAssignment {
- Objects.requireNonNull(members);
- }
-
-}
+package org.apache.kafka.coordinator.group.api.streams.assignor;
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 f2442c70463..43ef8e90331 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
@@ -92,6 +92,8 @@ import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
import org.apache.kafka.coordinator.group.api.assignor.ShareGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.SubscriptionType;
+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.assignor.SimpleAssignor;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.classic.ClassicGroupMember;
@@ -162,8 +164,6 @@ import org.apache.kafka.coordinator.group.streams.StreamsTopology;
import org.apache.kafka.coordinator.group.streams.TasksTuple;
import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
import org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignorException;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology;
import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsets.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsets.java
index ed2434a79bf..82ebe9d0961 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsets.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsets.java
@@ -18,7 +18,6 @@ package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
import java.util.HashMap;
import java.util.List;
@@ -32,10 +31,10 @@ import java.util.Map;
* constantly changing"); they are held in memory on the group coordinator and re-reported by the member on the
* task-offset interval.
*
- * @param taskOffsets Cumulative changelog offsets per task.
- * @param taskEndOffsets Cumulative changelog end-offsets per task.
+ * @param taskOffsets Cumulative changelog offsets, keyed by subtopology ID then partition.
+ * @param taskEndOffsets Cumulative changelog end-offsets, keyed by subtopology ID then partition.
*/
-public record MemberTaskOffsets(Map<TaskId, Long> taskOffsets, Map<TaskId, Long> taskEndOffsets) {
+public record MemberTaskOffsets(Map<String, Map<Integer, Long>> taskOffsets, Map<String, Map<Integer, Long>> taskEndOffsets) {
public static final MemberTaskOffsets EMPTY = new MemberTaskOffsets(Map.of(), Map.of());
@@ -52,19 +51,19 @@ public record MemberTaskOffsets(Map<TaskId, Long> taskOffsets, Map<TaskId, Long>
final List<StreamsGroupHeartbeatRequestData.TaskOffset> reportedTaskEndOffsets
) {
return new MemberTaskOffsets(
- reportedTaskOffsets == null ? taskOffsets : toTaskIdMap(reportedTaskOffsets),
- reportedTaskEndOffsets == null ? taskEndOffsets : toTaskIdMap(reportedTaskEndOffsets)
+ reportedTaskOffsets == null ? taskOffsets : toNestedMap(reportedTaskOffsets),
+ reportedTaskEndOffsets == null ? taskEndOffsets : toNestedMap(reportedTaskEndOffsets)
);
}
- private static Map<TaskId, Long> toTaskIdMap(final List<StreamsGroupHeartbeatRequestData.TaskOffset> taskOffsets) {
- final Map<TaskId, Long> result = new HashMap<>(taskOffsets.size());
+ private static Map<String, Map<Integer, Long>> toNestedMap(final List<StreamsGroupHeartbeatRequestData.TaskOffset> taskOffsets) {
+ final Map<String, Map<Integer, Long>> result = new HashMap<>();
for (final StreamsGroupHeartbeatRequestData.TaskOffset taskOffset : taskOffsets) {
- final TaskId taskId = new TaskId(taskOffset.subtopologyId(), taskOffset.partition());
+ final Map<Integer, Long> byPartition = result.computeIfAbsent(taskOffset.subtopologyId(), k -> new HashMap<>());
// The reported values come straight from the client heartbeat and the protocol does not enforce uniqueness
// of (subtopologyId, partition). Reject a duplicate with a clear client error rather than silently picking
// one of the values.
- if (result.putIfAbsent(taskId, taskOffset.offset()) != null) {
+ if (byPartition.putIfAbsent(taskOffset.partition(), taskOffset.offset()) != null) {
throw new InvalidRequestException("Task offsets contain a duplicate entry for subtopology "
+ taskOffset.subtopologyId() + " and partition " + taskOffset.partition() + ".");
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
index 313f3e2c1f4..0b4579c197a 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMember.java
@@ -19,7 +19,6 @@ package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
import org.slf4j.Logger;
@@ -417,14 +416,16 @@ public record StreamsGroupMember(String memberId,
);
}
- private static List<StreamsGroupDescribeResponseData.TaskOffset> taskOffsetsFromMap(Map<TaskId, Long> offsets) {
- return offsets.entrySet().stream()
- .sorted(Map.Entry.comparingByKey())
- .map(entry -> new StreamsGroupDescribeResponseData.TaskOffset()
- .setSubtopologyId(entry.getKey().subtopologyId())
- .setPartition(entry.getKey().partition())
- .setOffset(entry.getValue()))
- .toList();
+ private static List<StreamsGroupDescribeResponseData.TaskOffset> taskOffsetsFromMap(Map<String, Map<Integer, Long>> offsets) {
+ List<StreamsGroupDescribeResponseData.TaskOffset> taskOffsets = new ArrayList<>();
+ offsets.keySet().stream().sorted().forEach(subtopologyId ->
+ offsets.get(subtopologyId).entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .forEach(entry -> taskOffsets.add(new StreamsGroupDescribeResponseData.TaskOffset()
+ .setSubtopologyId(subtopologyId)
+ .setPartition(entry.getKey())
+ .setOffset(entry.getValue()))));
+ return taskOffsets;
}
private static List<StreamsGroupDescribeResponseData.TaskIds> taskIdsFromMap(Map<String, Set<Integer>> tasks) {
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
index 6ce2632b6f9..22cd1745808 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java
@@ -19,20 +19,22 @@ package org.apache.kafka.coordinator.group.streams;
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.streams.assignor.AssignmentMemberSpec;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment;
+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.GroupSpecImpl;
-import org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignorException;
+import org.apache.kafka.coordinator.group.streams.assignor.MemberMetadataAndStateImpl;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -116,19 +118,25 @@ public class TargetAssignmentBuilder {
this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
}
- static AssignmentMemberSpec createAssignmentMemberSpec(
+ static MemberMetadataAndStateImpl createMemberMetadataAndState(
StreamsGroupMember member,
- TasksTuple targetAssignment,
MemberTaskOffsets taskOffsets
) {
- return new AssignmentMemberSpec(
+ // Active, standby and warm-up tasks all reflect the tasks the member currently has, not the
+ // target assignment. Active tasks are stored with epochs; MemberAssignmentState exposes them
+ // without, so drop the epoch.
+ TasksTupleWithEpochs currentAssignment = member.assignedTasks();
+ Map<String, Set<Integer>> activeTasks = new HashMap<>();
+ currentAssignment.activeTasksWithEpochs().forEach((subtopologyId, partitionsWithEpochs) ->
+ activeTasks.put(subtopologyId, new HashSet<>(partitionsWithEpochs.keySet())));
+ return new MemberMetadataAndStateImpl(
member.instanceId(),
member.rackId(),
- targetAssignment.activeTasks(),
- targetAssignment.standbyTasks(),
- targetAssignment.warmupTasks(),
member.processId(),
member.clientTags(),
+ activeTasks,
+ currentAssignment.standbyTasks(),
+ currentAssignment.warmupTasks(),
taskOffsets.taskOffsets(),
taskOffsets.taskEndOffsets()
);
@@ -217,12 +225,11 @@ public class TargetAssignmentBuilder {
* @throws TaskAssignorException if the target assignment cannot be computed.
*/
public TargetAssignmentResult build() throws TaskAssignorException {
- Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> memberMetadataMap = new HashMap<>();
- // Prepare the member spec for all members.
- members.forEach((memberId, member) -> memberSpecs.put(memberId, createAssignmentMemberSpec(
+ // Prepare the member metadata for all members.
+ members.forEach((memberId, member) -> memberMetadataMap.put(memberId, createMemberMetadataAndState(
member,
- targetAssignment.getOrDefault(memberId, org.apache.kafka.coordinator.group.streams.TasksTuple.EMPTY),
taskOffsets.getOrDefault(memberId, MemberTaskOffsets.EMPTY)
)));
@@ -234,14 +241,14 @@ public class TargetAssignmentBuilder {
}
newGroupAssignment = assignor.assign(
new GroupSpecImpl(
- Collections.unmodifiableMap(memberSpecs),
+ Collections.unmodifiableMap(memberMetadataMap),
assignmentConfigs
),
new TopologyMetadata(metadataImage, topology.subtopologies().get())
);
} else {
newGroupAssignment = new GroupAssignment(
- memberSpecs.keySet().stream().collect(Collectors.toMap(x -> x, x -> MemberAssignment.empty())));
+ memberMetadataMap.keySet().stream().collect(Collectors.toMap(x -> x, x -> new MemberAssignment(Map.of(), Map.of()))));
}
// Compute delta from previous to new target assignment and create the
@@ -249,7 +256,7 @@ public class TargetAssignmentBuilder {
List<CoordinatorRecord> records = new ArrayList<>();
Map<String, org.apache.kafka.coordinator.group.streams.TasksTuple> newTargetAssignment = new HashMap<>();
- memberSpecs.keySet().forEach(memberId -> {
+ memberMetadataMap.keySet().forEach(memberId -> {
org.apache.kafka.coordinator.group.streams.TasksTuple oldMemberAssignment = targetAssignment.get(memberId);
org.apache.kafka.coordinator.group.streams.TasksTuple newMemberAssignment = newMemberAssignment(newGroupAssignment, memberId);
@@ -291,16 +298,25 @@ public class TargetAssignmentBuilder {
) {
MemberAssignment newMemberAssignment = newGroupAssignment.members().get(memberId);
if (newMemberAssignment != null) {
+ // Copy the maps returned by the assignor so the server does not keep a reference to
+ // maps the assignor is free to mutate afterwards.
return new TasksTuple(
- newMemberAssignment.activeTasks(),
- newMemberAssignment.standbyTasks(),
- newMemberAssignment.warmupTasks()
+ copyTasks(newMemberAssignment.activeTasks()),
+ copyTasks(newMemberAssignment.standbyTasks()),
+ // Warm-up tasks are not assigned by the assignor; they are decided during reconciliation.
+ Map.of()
);
} else {
return TasksTuple.EMPTY;
}
}
+ private static Map<String, Set<Integer>> copyTasks(Map<String, Set<Integer>> tasks) {
+ Map<String, Set<Integer>> copy = new HashMap<>();
+ tasks.forEach((subtopologyId, partitions) -> copy.put(subtopologyId, new HashSet<>(partitions)));
+ return copy;
+ }
+
/**
* The assignment result returned by {{@link TargetAssignmentBuilder#build()}}.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
index e07a3f1eca4..2286392bb00 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopologyMetadata.java
@@ -17,7 +17,7 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
-import org.apache.kafka.coordinator.group.streams.assignor.TopologyDescriber;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
import java.util.Collections;
@@ -27,7 +27,7 @@ import java.util.Objects;
import java.util.SortedMap;
/**
- * The topology metadata class is used by the {@link org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor} to get topic and
+ * The topology metadata class is used by the {@link org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor} to get topic and
* partition metadata for the topology that the streams group using.
*
* @param metadataImage The metadata image
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImpl.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImpl.java
index 7479fcae0fc..bdd07eeba61 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImpl.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImpl.java
@@ -16,21 +16,52 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+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;
+
+import java.util.Collection;
import java.util.Map;
import java.util.Objects;
/**
* The assignment specification for a streams group.
*
- * @param members The member metadata keyed by member ID.
- * @param assignmentConfigs Any configurations passed to the assignor.
+ * @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.
*/
-public record GroupSpecImpl(Map<String, AssignmentMemberSpec> members,
- Map<String, String> assignmentConfigs) implements GroupSpec {
+public record GroupSpecImpl(
+ Map<String, MemberMetadataAndStateImpl> members,
+ Map<String, String> configs
+) implements GroupSpec {
public GroupSpecImpl {
Objects.requireNonNull(members);
- Objects.requireNonNull(assignmentConfigs);
+ Objects.requireNonNull(configs);
+ }
+
+ @Override
+ public Collection<String> memberIds() {
+ return members.keySet();
+ }
+
+ @Override
+ public MemberAssignmentMetadata memberMetadata(String memberId) {
+ return requireMember(memberId);
+ }
+
+ @Override
+ public MemberAssignmentState memberAssignmentState(String memberId) {
+ return requireMember(memberId);
+ }
+
+ private MemberMetadataAndStateImpl requireMember(String memberId) {
+ MemberMetadataAndStateImpl member = members.get(memberId);
+ if (member == null) {
+ throw new IllegalArgumentException("Member Id " + memberId + " not found.");
+ }
+ return member;
}
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberAssignment.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberAssignment.java
deleted file mode 100644
index 2902e647382..00000000000
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberAssignment.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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 java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-/**
- * The task assignment for a streams group member.
- *
- * @param activeTasks The active tasks assigned to this member keyed by subtopologyId.
- * @param standbyTasks The standby tasks assigned to this member keyed by subtopologyId.
- * @param warmupTasks The warm-up tasks assigned to this member keyed by subtopologyId.
- */
-public record MemberAssignment(Map<String, Set<Integer>> activeTasks,
- Map<String, Set<Integer>> standbyTasks,
- Map<String, Set<Integer>> warmupTasks) {
-
- public MemberAssignment {
- Objects.requireNonNull(activeTasks);
- Objects.requireNonNull(standbyTasks);
- Objects.requireNonNull(warmupTasks);
- }
-
- public static MemberAssignment empty() {
- return new MemberAssignment(Map.of(), Map.of(), Map.of());
- }
-}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentMemberSpec.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndStateImpl.java
similarity index 54%
rename from group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentMemberSpec.java
rename to group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndStateImpl.java
index fa41b5a8a3b..ec04f4804f0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentMemberSpec.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndStateImpl.java
@@ -16,6 +16,9 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentMetadata;
+import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentState;
+
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
@@ -23,36 +26,39 @@ import java.util.Optional;
import java.util.Set;
/**
- * The assignment specification for a streams group member.
+ * Implementation of both the {@link MemberAssignmentMetadata} and the {@link MemberAssignmentState}
+ * interfaces for a streams group member.
*
- * @param instanceId The instance ID if provided.
- * @param rackId The rack ID if provided.
- * @param activeTasks Current target active tasks
- * @param standbyTasks Current target standby tasks
- * @param warmupTasks Current target warm-up tasks
- * @param processId The process ID.
- * @param clientTags The client tags for a rack-aware assignment.
- * @param taskOffsets The last received cumulative task offsets of assigned tasks or dormant tasks.
+ * @param instanceId The instance ID if provided.
+ * @param rackId The rack ID if provided.
+ * @param processId The process ID.
+ * @param clientTags The client tags for a rack-aware assignment.
+ * @param activeTasks Current active tasks.
+ * @param standbyTasks Current standby tasks.
+ * @param warmupTasks Current warm-up tasks.
+ * @param taskOffsets The last received cumulative task offsets of assigned tasks or dormant tasks.
+ * @param taskEndOffsets The last received cumulative task end offsets of assigned tasks or dormant tasks.
*/
-public record AssignmentMemberSpec(Optional<String> instanceId,
- Optional<String> rackId,
- Map<String, Set<Integer>> activeTasks,
- Map<String, Set<Integer>> standbyTasks,
- Map<String, Set<Integer>> warmupTasks,
- String processId,
- Map<String, String> clientTags,
- Map<TaskId, Long> taskOffsets,
- Map<TaskId, Long> taskEndOffsets
-) {
+public record MemberMetadataAndStateImpl(
+ Optional<String> instanceId,
+ Optional<String> rackId,
+ String processId,
+ Map<String, String> clientTags,
+ Map<String, Set<Integer>> activeTasks,
+ Map<String, Set<Integer>> standbyTasks,
+ Map<String, Set<Integer>> warmupTasks,
+ Map<String, Map<Integer, Long>> taskOffsets,
+ Map<String, Map<Integer, Long>> taskEndOffsets
+) implements MemberAssignmentMetadata, MemberAssignmentState {
- public AssignmentMemberSpec {
+ public MemberMetadataAndStateImpl {
Objects.requireNonNull(instanceId);
Objects.requireNonNull(rackId);
+ Objects.requireNonNull(processId);
+ clientTags = Collections.unmodifiableMap(Objects.requireNonNull(clientTags));
activeTasks = Collections.unmodifiableMap(Objects.requireNonNull(activeTasks));
standbyTasks = Collections.unmodifiableMap(Objects.requireNonNull(standbyTasks));
warmupTasks = Collections.unmodifiableMap(Objects.requireNonNull(warmupTasks));
- Objects.requireNonNull(processId);
- clientTags = Collections.unmodifiableMap(Objects.requireNonNull(clientTags));
taskOffsets = Collections.unmodifiableMap(Objects.requireNonNull(taskOffsets));
taskEndOffsets = Collections.unmodifiableMap(Objects.requireNonNull(taskEndOffsets));
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java
index 93827d29b1c..2f81b2bd547 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignor.java
@@ -16,6 +16,13 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+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;
+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.api.streams.assignor.TopologyDescriber;
+
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -56,13 +63,10 @@ public class MockAssignor implements TaskAssignor {
}
// Copy existing assignment and fill temporary data structures
- for (Map.Entry<String, AssignmentMemberSpec> memberEntry : groupSpec.members().entrySet()) {
- final String memberId = memberEntry.getKey();
- final AssignmentMemberSpec memberSpec = memberEntry.getValue();
-
- Map<String, Set<Integer>> activeTasks = new HashMap<>(memberSpec.activeTasks());
+ for (final String memberId : groupSpec.memberIds()) {
+ Map<String, Set<Integer>> activeTasks = new HashMap<>(groupSpec.memberAssignmentState(memberId).activeTasks());
- newTargetAssignment.put(memberId, new MemberAssignment(activeTasks, new HashMap<>(), new HashMap<>()));
+ newTargetAssignment.put(memberId, new MemberAssignment(activeTasks, new HashMap<>()));
for (Map.Entry<String, Set<Integer>> entry : activeTasks.entrySet()) {
final String subtopologyId = entry.getKey();
final Set<Integer> taskIds = entry.getValue();
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/ProcessState.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/ProcessState.java
index 330a0cc0da3..b06ff236f49 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/ProcessState.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/ProcessState.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignorException;
+
import java.util.AbstractMap;
import java.util.HashMap;
import java.util.HashSet;
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java
index 4828f0238b2..b1686b426aa 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignor.java
@@ -17,10 +17,19 @@
package org.apache.kafka.coordinator.group.streams.assignor;
+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;
+import org.apache.kafka.coordinator.group.api.streams.assignor.MemberAssignmentState;
+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.api.streams.assignor.TopologyDescriber;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -66,7 +75,7 @@ public class StickyTaskAssignor implements TaskAssignor {
assignStandby(statefulTasks);
}
- return buildGroupAssignment(groupSpec.members().keySet());
+ return buildGroupAssignment(groupSpec.memberIds());
}
private LinkedList<TaskId> taskIds(final TopologyDescriber topologyDescriber, final boolean isActive) {
@@ -85,8 +94,8 @@ public class StickyTaskAssignor implements TaskAssignor {
private void initialize(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber) {
localState = new LocalState();
localState.numStandbyReplicas =
- groupSpec.assignmentConfigs().isEmpty() ? 0
- : Integer.parseInt(groupSpec.assignmentConfigs().get("num.standby.replicas"));
+ groupSpec.configs().isEmpty() ? 0
+ : Integer.parseInt(groupSpec.configs().get("num.standby.replicas"));
// Helpers for computing active tasks per member, and tasks per member
localState.totalActiveTasks = 0;
@@ -98,25 +107,24 @@ public class StickyTaskAssignor implements TaskAssignor {
if (topologyDescriber.isStateful(subtopology))
localState.totalTasks += numberOfPartitions * localState.numStandbyReplicas;
}
- localState.totalMembersWithActiveTaskCapacity = groupSpec.members().size();
- localState.totalMembersWithTaskCapacity = groupSpec.members().size();
+ localState.totalMembersWithActiveTaskCapacity = groupSpec.memberIds().size();
+ localState.totalMembersWithTaskCapacity = groupSpec.memberIds().size();
localState.activeTasksPerMember = computeTasksPerMember(localState.totalActiveTasks, localState.totalMembersWithActiveTaskCapacity);
localState.totalTasksPerMember = computeTasksPerMember(localState.totalTasks, localState.totalMembersWithTaskCapacity);
localState.processIdToState = new HashMap<>(localState.totalMembersWithActiveTaskCapacity);
localState.activeTaskToPrevMember = new HashMap<>(localState.totalActiveTasks);
localState.standbyTaskToPrevMember = new HashMap<>(localState.numStandbyReplicas > 0 ? (localState.totalTasks - localState.totalActiveTasks) / localState.numStandbyReplicas : 0);
- for (final Map.Entry<String, AssignmentMemberSpec> memberEntry : groupSpec.members().entrySet()) {
- final String memberId = memberEntry.getKey();
- final String processId = memberEntry.getValue().processId();
+ for (final String memberId : groupSpec.memberIds()) {
+ final MemberAssignmentState memberAssignmentState = groupSpec.memberAssignmentState(memberId);
+ final String processId = groupSpec.memberMetadata(memberId).processId();
final Member member = new Member(processId, memberId);
- final AssignmentMemberSpec memberSpec = memberEntry.getValue();
localState.processIdToState.putIfAbsent(processId, new ProcessState(processId));
localState.processIdToState.get(processId).addMember(memberId);
// prev active tasks
- for (final Map.Entry<String, Set<Integer>> entry : memberSpec.activeTasks().entrySet()) {
+ for (final Map.Entry<String, Set<Integer>> entry : memberAssignmentState.activeTasks().entrySet()) {
final Set<Integer> partitionNoSet = entry.getValue();
for (final int partitionNo : partitionNoSet) {
localState.activeTaskToPrevMember.put(new TaskId(entry.getKey(), partitionNo), member);
@@ -124,7 +132,7 @@ public class StickyTaskAssignor implements TaskAssignor {
}
// prev standby tasks
- for (final Map.Entry<String, Set<Integer>> entry : memberSpec.standbyTasks().entrySet()) {
+ for (final Map.Entry<String, Set<Integer>> entry : memberAssignmentState.standbyTasks().entrySet()) {
final Set<Integer> partitionNoSet = entry.getValue();
for (final int partitionNo : partitionNoSet) {
final TaskId taskId = new TaskId(entry.getKey(), partitionNo);
@@ -135,7 +143,7 @@ public class StickyTaskAssignor implements TaskAssignor {
}
}
- private GroupAssignment buildGroupAssignment(final Set<String> members) {
+ private GroupAssignment buildGroupAssignment(final Collection<String> members) {
final Map<String, MemberAssignment> memberAssignments = new HashMap<>();
final Map<String, Set<TaskId>> activeTasksAssignments = localState.processIdToState.entrySet().stream()
@@ -162,7 +170,7 @@ public class StickyTaskAssignor implements TaskAssignor {
if (standbyTasksAssignments.containsKey(memberId)) {
standByTasks.putAll(toCompactedTaskIds(standbyTasksAssignments.get(memberId)));
}
- memberAssignments.put(memberId, new MemberAssignment(activeTasks, standByTasks, new HashMap<>()));
+ memberAssignments.put(memberId, new MemberAssignment(activeTasks, standByTasks));
}
return new GroupAssignment(memberAssignments);
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskId.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskId.java
index 979c042df00..fe1a14b105b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskId.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/TaskId.java
@@ -20,7 +20,7 @@ import java.util.Comparator;
import java.util.Objects;
/**
- * The identifier for a task
+ * The identifier for a task, consisting of the subtopology ID and the partition.
*
* @param subtopologyId The unique identifier of the subtopology.
* @param partition The partition of the input topics this task is processing.
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 185108b4d58..24fcb7a308f 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -97,6 +97,9 @@ import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAss
import org.apache.kafka.coordinator.group.api.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.api.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.api.assignor.PartitionAssignorException;
+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.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.classic.ClassicGroupMember;
import org.apache.kafka.coordinator.group.classic.ClassicGroupState;
@@ -148,8 +151,6 @@ import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil;
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.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignorException;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.MetadataProvenance;
@@ -11113,8 +11114,8 @@ public class GroupMetadataManagerTest {
// empty lists rather than borrowing another member's values).
StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
group.updateTaskOffsets(memberId1, new MemberTaskOffsets(
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 100L),
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 500L)
+ Map.of(subtopology1, Map.of(0, 100L)),
+ Map.of(subtopology1, Map.of(0, 500L))
));
List<StreamsGroupDescribeResponseData.DescribedGroup> described = context.sendStreamsGroupDescribe(List.of(groupId));
@@ -19056,8 +19057,8 @@ public class GroupMetadataManagerTest {
StreamsGroup group = context.groupMetadataManager.streamsGroup(groupId);
assertEquals(
new org.apache.kafka.coordinator.group.streams.MemberTaskOffsets(
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 10L),
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 20L)
+ Map.of(subtopology1, Map.of(0, 10L)),
+ Map.of(subtopology1, Map.of(0, 20L))
),
group.taskOffsets(memberId)
);
@@ -19082,8 +19083,8 @@ public class GroupMetadataManagerTest {
assertEquals(List.of(), result.records());
assertEquals(
new org.apache.kafka.coordinator.group.streams.MemberTaskOffsets(
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 12L),
- Map.of(new org.apache.kafka.coordinator.group.streams.assignor.TaskId(subtopology1, 0), 20L)
+ Map.of(subtopology1, Map.of(0, 12L)),
+ Map.of(subtopology1, Map.of(0, 20L))
),
group.taskOffsets(memberId)
);
@@ -19674,8 +19675,8 @@ public class GroupMetadataManagerTest {
)
.build();
- assignor.prepareGroupAssignment(new org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment(Map.of(
- memberId, org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment.empty()
+ assignor.prepareGroupAssignment(new org.apache.kafka.coordinator.group.api.streams.assignor.GroupAssignment(Map.of(
+ memberId, new MemberAssignment(Map.of(), Map.of())
)));
// Member joins the streams group.
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
index 0b8ea094c2d..c7a5ff48127 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java
@@ -60,6 +60,7 @@ import org.apache.kafka.coordinator.common.runtime.MockCoordinatorExecutor;
import org.apache.kafka.coordinator.common.runtime.MockCoordinatorTimer;
import org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import org.apache.kafka.coordinator.group.api.assignor.ShareGroupPartitionAssignor;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
import org.apache.kafka.coordinator.group.classic.ClassicGroup;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentKey;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
@@ -114,7 +115,6 @@ import org.apache.kafka.coordinator.group.streams.StreamsGroupBuilder;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import org.apache.kafka.coordinator.group.streams.TasksTupleWithEpochs;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
import org.apache.kafka.coordinator.group.streams.topics.InternalTopicManager;
import org.apache.kafka.server.authorizer.Authorizer;
import org.apache.kafka.server.common.ApiMessageAndVersion;
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsetsTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsetsTest.java
index 941c5c3cc97..8c1e2f305af 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsetsTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MemberTaskOffsetsTest.java
@@ -18,7 +18,6 @@ package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
import org.junit.jupiter.api.Test;
@@ -30,9 +29,11 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
public class MemberTaskOffsetsTest {
- private static StreamsGroupHeartbeatRequestData.TaskOffset taskOffset(final String subtopologyId,
- final int partition,
- final long offset) {
+ private static StreamsGroupHeartbeatRequestData.TaskOffset taskOffset(
+ final String subtopologyId,
+ final int partition,
+ final long offset
+ ) {
return new StreamsGroupHeartbeatRequestData.TaskOffset()
.setSubtopologyId(subtopologyId)
.setPartition(partition)
@@ -47,11 +48,11 @@ public class MemberTaskOffsetsTest {
);
assertEquals(
- Map.of(new TaskId("sub-1", 0), 10L, new TaskId("sub-1", 1), 20L),
+ Map.of("sub-1", Map.of(0, 10L, 1, 20L)),
result.taskOffsets()
);
assertEquals(
- Map.of(new TaskId("sub-1", 0), 15L, new TaskId("sub-1", 1), 25L),
+ Map.of("sub-1", Map.of(0, 15L, 1, 25L)),
result.taskEndOffsets()
);
}
@@ -76,8 +77,8 @@ public class MemberTaskOffsetsTest {
@Test
public void shouldRetainBothMapsWhenBothListsAreNull() {
MemberTaskOffsets previous = new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 10L),
- Map.of(new TaskId("sub-1", 0), 15L)
+ Map.of("sub-1", Map.of(0, 10L)),
+ Map.of("sub-1", Map.of(0, 15L))
);
assertEquals(previous, previous.update(null, null));
@@ -87,28 +88,28 @@ public class MemberTaskOffsetsTest {
@Test
public void shouldUpdateTaskOffsetsAndRetainTaskEndOffsetsWhenEndOffsetsNull() {
MemberTaskOffsets previous = new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 10L),
- Map.of(new TaskId("sub-1", 0), 15L)
+ Map.of("sub-1", Map.of(0, 10L)),
+ Map.of("sub-1", Map.of(0, 15L))
);
MemberTaskOffsets result = previous.update(List.of(taskOffset("sub-1", 0, 12L)), null);
- assertEquals(Map.of(new TaskId("sub-1", 0), 12L), result.taskOffsets());
+ assertEquals(Map.of("sub-1", Map.of(0, 12L)), result.taskOffsets());
// The end-offsets were not reported, so the previously reported values are retained.
- assertEquals(Map.of(new TaskId("sub-1", 0), 15L), result.taskEndOffsets());
+ assertEquals(Map.of("sub-1", Map.of(0, 15L)), result.taskEndOffsets());
}
@Test
public void shouldUpdateTaskEndOffsetsAndRetainTaskOffsetsWhenOffsetsNull() {
MemberTaskOffsets previous = new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 10L),
- Map.of(new TaskId("sub-1", 0), 15L)
+ Map.of("sub-1", Map.of(0, 10L)),
+ Map.of("sub-1", Map.of(0, 15L))
);
MemberTaskOffsets result = previous.update(null, List.of(taskOffset("sub-1", 0, 18L)));
// The offsets were not reported, so the previously reported values are retained.
- assertEquals(Map.of(new TaskId("sub-1", 0), 10L), result.taskOffsets());
- assertEquals(Map.of(new TaskId("sub-1", 0), 18L), result.taskEndOffsets());
+ assertEquals(Map.of("sub-1", Map.of(0, 10L)), result.taskOffsets());
+ assertEquals(Map.of("sub-1", Map.of(0, 18L)), result.taskEndOffsets());
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java
index 2b1f1fb5511..93bab5825e1 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java
@@ -16,12 +16,12 @@
*/
package org.apache.kafka.coordinator.group.streams;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupSpec;
-import org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignorException;
-import org.apache.kafka.coordinator.group.streams.assignor.TopologyDescriber;
+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;
+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.api.streams.assignor.TopologyDescriber;
import java.util.Map;
import java.util.Map.Entry;
@@ -49,7 +49,7 @@ public class MockTaskAssignor implements TaskAssignor {
entry -> {
TasksTuple tasksTuple = entry.getValue();
return new MemberAssignment(
- tasksTuple.activeTasks(), tasksTuple.standbyTasks(), tasksTuple.warmupTasks());
+ tasksTuple.activeTasks(), tasksTuple.standbyTasks());
})));
}
@@ -70,7 +70,7 @@ public class MockTaskAssignor implements TaskAssignor {
@Override
public GroupAssignment assign(final GroupSpec groupSpec, final TopologyDescriber topologyDescriber)
throws TaskAssignorException {
- assignmentConfigs = groupSpec.assignmentConfigs();
+ assignmentConfigs = groupSpec.configs();
return preparedGroupAssignment;
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
index 1eeec0bd02f..969e9c3d807 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupMemberTest.java
@@ -21,7 +21,6 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAss
import org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue.TaskIds;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.generated.StreamsGroupMemberMetadataValue.KeyValue;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
@@ -346,14 +345,12 @@ public class StreamsGroupMemberTest {
);
MemberTaskOffsets taskOffsets = new MemberTaskOffsets(
mkMap(
- mkEntry(new TaskId(SUBTOPOLOGY1, 0), 100L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 2), 120L),
- mkEntry(new TaskId(SUBTOPOLOGY2, 1), 200L)
+ mkEntry(SUBTOPOLOGY1, mkMap(mkEntry(0, 100L), mkEntry(2, 120L))),
+ mkEntry(SUBTOPOLOGY2, mkMap(mkEntry(1, 200L)))
),
mkMap(
- mkEntry(new TaskId(SUBTOPOLOGY1, 0), 150L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 2), 170L),
- mkEntry(new TaskId(SUBTOPOLOGY2, 1), 250L)
+ mkEntry(SUBTOPOLOGY1, mkMap(mkEntry(0, 150L), mkEntry(2, 170L))),
+ mkEntry(SUBTOPOLOGY2, mkMap(mkEntry(1, 250L)))
)
);
@@ -459,16 +456,12 @@ public class StreamsGroupMemberTest {
// TaskId, i.e. by subtopologyId then partition, for a deterministic response.
MemberTaskOffsets taskOffsets = new MemberTaskOffsets(
mkMap(
- mkEntry(new TaskId(SUBTOPOLOGY2, 1), 200L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 2), 120L),
- mkEntry(new TaskId(SUBTOPOLOGY2, 0), 190L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 0), 100L)
+ mkEntry(SUBTOPOLOGY2, mkMap(mkEntry(1, 200L), mkEntry(0, 190L))),
+ mkEntry(SUBTOPOLOGY1, mkMap(mkEntry(2, 120L), mkEntry(0, 100L)))
),
mkMap(
- mkEntry(new TaskId(SUBTOPOLOGY2, 0), 290L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 0), 150L),
- mkEntry(new TaskId(SUBTOPOLOGY2, 1), 250L),
- mkEntry(new TaskId(SUBTOPOLOGY1, 2), 170L)
+ mkEntry(SUBTOPOLOGY2, mkMap(mkEntry(0, 290L), mkEntry(1, 250L))),
+ mkEntry(SUBTOPOLOGY1, mkMap(mkEntry(0, 150L), mkEntry(2, 170L)))
)
);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
index 124520733da..82a5e8b7765 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java
@@ -48,7 +48,6 @@ import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyKey;
import org.apache.kafka.coordinator.group.generated.StreamsGroupTopologyValue;
import org.apache.kafka.coordinator.group.streams.StreamsGroup.StreamsGroupState;
import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.timeline.SnapshotRegistry;
@@ -120,8 +119,8 @@ public class StreamsGroupTest {
assertEquals(Map.of(), streamsGroup.taskOffsets());
MemberTaskOffsets offsets = new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 10L),
- Map.of(new TaskId("sub-1", 0), 20L)
+ Map.of("sub-1", Map.of(0, 10L)),
+ Map.of("sub-1", Map.of(0, 20L))
);
streamsGroup.updateTaskOffsets("member-id", offsets);
@@ -130,8 +129,8 @@ public class StreamsGroupTest {
// A new report replaces the previous one.
MemberTaskOffsets newerOffsets = new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 15L),
- Map.of(new TaskId("sub-1", 0), 25L)
+ Map.of("sub-1", Map.of(0, 15L)),
+ Map.of("sub-1", Map.of(0, 25L))
);
streamsGroup.updateTaskOffsets("member-id", newerOffsets);
assertEquals(newerOffsets, streamsGroup.taskOffsets("member-id"));
@@ -142,8 +141,8 @@ public class StreamsGroupTest {
StreamsGroup streamsGroup = createStreamsGroup("foo");
streamsGroup.updateMember(new StreamsGroupMember.Builder("member-id").build());
streamsGroup.updateTaskOffsets("member-id", new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 10L),
- Map.of(new TaskId("sub-1", 0), 20L)
+ Map.of("sub-1", Map.of(0, 10L)),
+ Map.of("sub-1", Map.of(0, 20L))
));
streamsGroup.removeMember("member-id");
@@ -1233,8 +1232,8 @@ public class StreamsGroupTest {
.build());
// Transient, unpersisted per-task offsets reported by the member; the describe path must surface them.
group.updateTaskOffsets("member1", new MemberTaskOffsets(
- Map.of(new TaskId("sub-1", 0), 5L),
- Map.of(new TaskId("sub-1", 0), 9L)
+ Map.of("sub-1", Map.of(0, 5L)),
+ Map.of("sub-1", Map.of(0, 9L))
));
snapshotRegistry.idempotentCreateSnapshot(1);
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
index 0edb840870c..e9f17f11a93 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java
@@ -22,14 +22,13 @@ import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.KRaftCoordinatorMetadataImage;
import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
+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.generated.StreamsGroupMemberMetadataValue;
import org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.TaskRole;
-import org.apache.kafka.coordinator.group.streams.assignor.AssignmentMemberSpec;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.streams.assignor.GroupSpecImpl;
-import org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskId;
+import org.apache.kafka.coordinator.group.streams.assignor.MemberMetadataAndStateImpl;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredTopology;
@@ -50,7 +49,7 @@ import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.apache.kafka.coordinator.group.Assertions.assertUnorderedRecordsEquals;
import static org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentMetadataRecord;
import static org.apache.kafka.coordinator.group.streams.StreamsCoordinatorRecordHelpers.newStreamsGroupTargetAssignmentRecord;
-import static org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.createAssignmentMemberSpec;
+import static org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.createMemberMetadataAndState;
import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasks;
import static org.apache.kafka.coordinator.group.streams.TaskAssignmentTestUtil.mkTasksTuple;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -87,45 +86,53 @@ public class TargetAssignmentBuilderTest {
}
@ParameterizedTest
- @EnumSource(TaskRole.class)
- public void testCreateAssignmentMemberSpec(TaskRole taskRole) {
+ // Active, standby and warm-up tasks all come from the member's current assignment, so this
+ // test varies which role the member's current tasks are in.
+ @EnumSource(value = TaskRole.class, names = {"ACTIVE", "STANDBY"})
+ public void testCreateMemberMetadataAndState(TaskRole taskRole) {
String fooSubtopologyId = Uuid.randomUuid().toString();
String barSubtopologyId = Uuid.randomUuid().toString();
final Map<String, String> clientTags = mkMap(mkEntry("tag1", "value1"), mkEntry("tag2", "value2"));
+
+ Map<String, Set<Integer>> activeTasks = taskRole == TaskRole.ACTIVE
+ ? Map.of(fooSubtopologyId, Set.of(1, 2, 3), barSubtopologyId, Set.of(1, 2, 3)) : Map.of();
+ Map<String, Set<Integer>> standbyTasks = taskRole == TaskRole.STANDBY
+ ? Map.of(fooSubtopologyId, Set.of(1, 2, 3), barSubtopologyId, Set.of(1, 2, 3)) : Map.of();
+
StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
.setRackId("rackId")
.setInstanceId("instanceId")
.setProcessId("processId")
.setClientTags(clientTags)
+ .setAssignedTasks(new TasksTupleWithEpochs(
+ taskRole == TaskRole.ACTIVE
+ ? Map.of(fooSubtopologyId, Map.of(1, 0, 2, 0, 3, 0), barSubtopologyId, Map.of(1, 0, 2, 0, 3, 0))
+ : Map.of(),
+ standbyTasks,
+ Map.of()))
.build();
- TasksTuple assignment = mkTasksTuple(taskRole,
- mkTasks(fooSubtopologyId, 1, 2, 3),
- mkTasks(barSubtopologyId, 1, 2, 3)
- );
-
- AssignmentMemberSpec assignmentMemberSpec = createAssignmentMemberSpec(
+ MemberMetadataAndStateImpl memberMetadata = createMemberMetadataAndState(
member,
- assignment,
MemberTaskOffsets.EMPTY
);
- assertEquals(new AssignmentMemberSpec(
+ assertEquals(new MemberMetadataAndStateImpl(
Optional.of("instanceId"),
Optional.of("rackId"),
- assignment.activeTasks(),
- assignment.standbyTasks(),
- assignment.warmupTasks(),
"processId",
clientTags,
+ activeTasks,
+ standbyTasks,
+ Map.of(),
Map.of(),
Map.of()
- ), assignmentMemberSpec);
+ ), memberMetadata);
}
@Test
- public void testCreateAssignmentMemberSpecPopulatesTaskOffsets() {
+ public void testCreateMemberMetadataAndStatePopulatesTaskOffsets() {
String fooSubtopologyId = Uuid.randomUuid().toString();
StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
@@ -133,19 +140,41 @@ public class TargetAssignmentBuilderTest {
.setInstanceId("instanceId")
.setProcessId("processId")
.setClientTags(Map.of())
+ .setAssignedTasks(TasksTupleWithEpochs.EMPTY)
.build();
- Map<TaskId, Long> taskOffsets = Map.of(new TaskId(fooSubtopologyId, 0), 10L);
- Map<TaskId, Long> taskEndOffsets = Map.of(new TaskId(fooSubtopologyId, 0), 20L);
+ Map<String, Map<Integer, Long>> taskOffsets = Map.of(fooSubtopologyId, Map.of(0, 10L));
+ Map<String, Map<Integer, Long>> taskEndOffsets = Map.of(fooSubtopologyId, Map.of(0, 20L));
- AssignmentMemberSpec assignmentMemberSpec = createAssignmentMemberSpec(
+ MemberMetadataAndStateImpl memberMetadata = createMemberMetadataAndState(
member,
- TasksTuple.EMPTY,
new MemberTaskOffsets(taskOffsets, taskEndOffsets)
);
- assertEquals(taskOffsets, assignmentMemberSpec.taskOffsets());
- assertEquals(taskEndOffsets, assignmentMemberSpec.taskEndOffsets());
+ assertEquals(taskOffsets, memberMetadata.taskOffsets());
+ assertEquals(taskEndOffsets, memberMetadata.taskEndOffsets());
+ }
+
+ @Test
+ public void testCreateMemberMetadataAndStateSourcesWarmupTasksFromCurrentAssignment() {
+ String fooSubtopologyId = Uuid.randomUuid().toString();
+
+ // Warm-up tasks are decided during reconciliation and stored in the member's current
+ // assignment; they must be sourced from there, not from the target assignment.
+ StreamsGroupMember member = new StreamsGroupMember.Builder("member-id")
+ .setInstanceId("instanceId")
+ .setRackId("rackId")
+ .setProcessId("processId")
+ .setClientTags(Map.of())
+ .setAssignedTasks(new TasksTupleWithEpochs(Map.of(), Map.of(), Map.of(fooSubtopologyId, Set.of(1, 2, 3))))
+ .build();
+
+ MemberMetadataAndStateImpl memberMetadata = createMemberMetadataAndState(
+ member,
+ MemberTaskOffsets.EMPTY
+ );
+
+ assertEquals(Map.of(fooSubtopologyId, Set.of(1, 2, 3)), memberMetadata.warmupTasks());
}
@Test
@@ -167,7 +196,9 @@ public class TargetAssignmentBuilderTest {
@ParameterizedTest
- @EnumSource(TaskRole.class)
+ // Warm-up tasks are not produced by the assignor (only active and standby), so they cannot appear
+ // in the resulting target assignment. See MemberAssignment.
+ @EnumSource(value = TaskRole.class, names = {"ACTIVE", "STANDBY"})
public void testAssignmentHasNotChanged(TaskRole taskRole) {
TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
"my-group",
@@ -221,7 +252,9 @@ public class TargetAssignmentBuilderTest {
@ParameterizedTest
- @EnumSource(TaskRole.class)
+ // Warm-up tasks are not produced by the assignor (only active and standby), so they cannot appear
+ // in the resulting target assignment. See MemberAssignment.
+ @EnumSource(value = TaskRole.class, names = {"ACTIVE", "STANDBY"})
public void testAssignmentSwapped(TaskRole taskRole) {
TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
"my-group",
@@ -288,7 +321,9 @@ public class TargetAssignmentBuilderTest {
@ParameterizedTest
- @EnumSource(TaskRole.class)
+ // Warm-up tasks are not produced by the assignor (only active and standby), so they cannot appear
+ // in the resulting target assignment. See MemberAssignment.
+ @EnumSource(value = TaskRole.class, names = {"ACTIVE", "STANDBY"})
public void testPartialAssignmentUpdate(TaskRole taskRole) {
TargetAssignmentBuilderTestContext context = new TargetAssignmentBuilderTestContext(
"my-group",
@@ -404,6 +439,7 @@ public class TargetAssignmentBuilderTest {
memberBuilder.setUserEndpoint(new StreamsGroupMemberMetadataValue.Endpoint().setHost("host").setPort(9090));
memberBuilder.setInstanceId(null);
memberBuilder.setRackId(null);
+ memberBuilder.setAssignedTasks(TasksTupleWithEpochs.EMPTY);
members.put(memberId, memberBuilder.build());
targetAssignment.put(memberId, targetTasks);
}
@@ -424,16 +460,15 @@ public class TargetAssignmentBuilderTest {
String memberId,
TasksTuple assignment
) {
- memberAssignments.put(memberId, new MemberAssignment(assignment.activeTasks(), assignment.standbyTasks(), assignment.warmupTasks()));
+ memberAssignments.put(memberId, new MemberAssignment(assignment.activeTasks(), assignment.standbyTasks()));
}
public org.apache.kafka.coordinator.group.streams.TargetAssignmentBuilder.TargetAssignmentResult build() {
// Prepare expected member specs.
- Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> memberMetadataMap = new HashMap<>();
members.forEach((memberId, member) ->
- memberSpecs.put(memberId, createAssignmentMemberSpec(
+ memberMetadataMap.put(memberId, createMemberMetadataAndState(
member,
- targetAssignment.getOrDefault(memberId, TasksTuple.EMPTY),
MemberTaskOffsets.EMPTY
)
));
@@ -444,7 +479,7 @@ public class TargetAssignmentBuilderTest {
TopologyMetadata topologyMetadata = new TopologyMetadata(metadataImage, subtopologies);
// Prepare the expected assignment spec.
- GroupSpecImpl groupSpec = new GroupSpecImpl(memberSpecs, new HashMap<>());
+ GroupSpecImpl groupSpec = new GroupSpecImpl(memberMetadataMap, new HashMap<>());
// We use `any` here to always return an assignment but use `verify` later on
// to ensure that the input was correct.
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImplTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImplTest.java
index be2aa696e8f..4d30558192b 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImplTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/GroupSpecImplTest.java
@@ -24,28 +24,32 @@ 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 {
- private Map<String, AssignmentMemberSpec> members;
+ private Map<String, MemberMetadataAndStateImpl> members;
+ private MemberMetadataAndStateImpl member;
private GroupSpecImpl groupSpec;
@BeforeEach
void setUp() {
members = new HashMap<>();
- members.put("test-member", new AssignmentMemberSpec(
+ member = new MemberMetadataAndStateImpl(
Optional.of("test-instance"),
Optional.of("test-rack"),
+ "test-process",
Map.of(),
Map.of(),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
- ));
+ );
+ members.put("test-member", member);
groupSpec = new GroupSpecImpl(
members,
@@ -54,8 +58,29 @@ public class GroupSpecImplTest {
}
@Test
- void testMembers() {
- assertEquals(members, groupSpec.members());
+ void testMemberIds() {
+ assertEquals(members.keySet(), groupSpec.memberIds());
+ }
+
+ @Test
+ void testMemberMetadata() {
+ assertEquals(member, groupSpec.memberMetadata("test-member"));
+ }
+
+ @Test
+ void testMemberAssignmentState() {
+ assertEquals(member, groupSpec.memberAssignmentState("test-member"));
+ }
+
+ @Test
+ void testMemberNotFound() {
+ assertThrows(IllegalArgumentException.class, () -> groupSpec.memberMetadata("unknown"));
+ assertThrows(IllegalArgumentException.class, () -> groupSpec.memberAssignmentState("unknown"));
+ }
+
+ @Test
+ void testConfigs() {
+ assertTrue(groupSpec.configs().isEmpty());
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java
index da73e27e8c2..a4d4d914aab 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java
@@ -16,6 +16,11 @@
*/
package org.apache.kafka.coordinator.group.streams.assignor;
+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.TaskAssignorException;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber;
+
import org.junit.jupiter.api.Test;
import java.util.HashMap;
@@ -60,25 +65,25 @@ public class MockAssignorTest {
@Test
public void testDoubleAssignment() {
- final AssignmentMemberSpec memberSpec1 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata1 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
- Map.of("test-subtopology", Set.of(0)),
+ "test-process",
Map.of(),
+ Map.of("test-subtopology", Set.of(0)),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
);
- final AssignmentMemberSpec memberSpec2 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata2 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
- Map.of("test-subtopology", Set.of(0)),
+ "test-process",
Map.of(),
+ Map.of("test-subtopology", Set.of(0)),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
@@ -86,7 +91,7 @@ public class MockAssignorTest {
TaskAssignorException ex = assertThrows(TaskAssignorException.class, () -> assignor.assign(
new GroupSpecImpl(
- Map.of("member1", memberSpec1, "member2", memberSpec2),
+ Map.of("member1", memberMetadata1, "member2", memberMetadata2),
new HashMap<>()
),
new TopologyDescriberImpl(5, List.of("test-subtopology"))
@@ -113,13 +118,13 @@ public class MockAssignorTest {
@Test
public void testSingleMember() {
- final AssignmentMemberSpec memberSpec = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "test-process",
Map.of(),
Map.of(),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
@@ -127,7 +132,7 @@ public class MockAssignorTest {
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(
- Map.of("test_member", memberSpec),
+ Map.of("test_member", memberMetadata),
new HashMap<>()
),
new TopologyDescriberImpl(4, List.of("test-subtopology"))
@@ -145,25 +150,25 @@ public class MockAssignorTest {
@Test
public void testTwoMembersTwoSubtopologies() {
- final AssignmentMemberSpec memberSpec1 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata1 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "test-process",
Map.of(),
Map.of(),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
);
- final AssignmentMemberSpec memberSpec2 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata2 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "test-process",
Map.of(),
Map.of(),
Map.of(),
- "test-process",
Map.of(),
Map.of(),
Map.of()
@@ -171,7 +176,7 @@ public class MockAssignorTest {
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(
- mkMap(mkEntry("test_member1", memberSpec1), mkEntry("test_member2", memberSpec2)),
+ mkMap(mkEntry("test_member1", memberMetadata1), mkEntry("test_member2", memberMetadata2)),
new HashMap<>()
),
new TopologyDescriberImpl(4, List.of("test-subtopology1", "test-subtopology2"))
@@ -198,38 +203,38 @@ public class MockAssignorTest {
@Test
public void testTwoMembersTwoSubtopologiesStickiness() {
- final AssignmentMemberSpec memberSpec1 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata1 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "test-process",
+ Map.of(),
mkMap(
mkEntry("test-subtopology1", new HashSet<>(List.of(0, 2, 3))),
mkEntry("test-subtopology2", new HashSet<>(List.of(0)))
),
Map.of(),
Map.of(),
- "test-process",
- Map.of(),
Map.of(),
Map.of()
);
- final AssignmentMemberSpec memberSpec2 = new AssignmentMemberSpec(
+ final MemberMetadataAndStateImpl memberMetadata2 = new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "test-process",
+ Map.of(),
mkMap(
mkEntry("test-subtopology1", new HashSet<>(List.of(1))),
mkEntry("test-subtopology2", new HashSet<>(List.of(3)))
),
Map.of(),
Map.of(),
- "test-process",
- Map.of(),
Map.of(),
Map.of()
);
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(
- mkMap(mkEntry("test_member1", memberSpec1), mkEntry("test_member2", memberSpec2)),
+ mkMap(mkEntry("test_member1", memberMetadata1), mkEntry("test_member2", memberMetadata2)),
new HashMap<>()
),
new TopologyDescriberImpl(4, List.of("test-subtopology1", "test-subtopology2"))
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java
index 161f3e6a5d8..f623f196f23 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java
@@ -17,6 +17,10 @@
package org.apache.kafka.coordinator.group.streams.assignor;
+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.TopologyDescriber;
+
import org.junit.jupiter.api.Test;
import org.mockito.internal.util.collections.Sets;
@@ -53,13 +57,13 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignOneActiveTaskToEachProcessWhenTaskCountSameAsProcessCount() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(
- mkMap(mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3)),
+ mkMap(mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3)),
new HashMap<>()
),
new TopologyDescriberImpl(3, false, List.of("test-subtopology"))
@@ -78,15 +82,15 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTopicGroupIdEvenlyAcrossClientsWithNoStandByTasks() {
- final AssignmentMemberSpec memberSpec11 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec12 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec21 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec22 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec31 = createAssignmentMemberSpec("process3");
- final AssignmentMemberSpec memberSpec32 = createAssignmentMemberSpec("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(mkEntry("member1_1", memberSpec11), mkEntry("member1_2", memberSpec12),
- mkEntry("member2_1", memberSpec21), mkEntry("member2_2", memberSpec22),
- mkEntry("member3_1", memberSpec31), mkEntry("member3_2", memberSpec32));
+ final MemberMetadataAndStateImpl memberMetadata11 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata12 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata21 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata22 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata31 = createMemberMetadata("process3");
+ final MemberMetadataAndStateImpl memberMetadata32 = createMemberMetadata("process3");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(mkEntry("member1_1", memberMetadata11), mkEntry("member1_2", memberMetadata12),
+ mkEntry("member2_1", memberMetadata21), mkEntry("member2_2", memberMetadata22),
+ mkEntry("member3_1", memberMetadata31), mkEntry("member3_2", memberMetadata32));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -106,15 +110,15 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTopicGroupIdEvenlyAcrossClientsWithStandByTasks() {
final Map<String, Set<Integer>> tasks = mkMap(mkEntry("test-subtopology1", Sets.newSet(0, 1, 2)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 2)));
- final AssignmentMemberSpec memberSpec11 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec12 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec21 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec22 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec31 = createAssignmentMemberSpec("process3");
- final AssignmentMemberSpec memberSpec32 = createAssignmentMemberSpec("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(mkEntry("member1_1", memberSpec11), mkEntry("member1_2", memberSpec12),
- mkEntry("member2_1", memberSpec21), mkEntry("member2_2", memberSpec22),
- mkEntry("member3_1", memberSpec31), mkEntry("member3_2", memberSpec32));
+ final MemberMetadataAndStateImpl memberMetadata11 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata12 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata21 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata22 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata31 = createMemberMetadata("process3");
+ final MemberMetadataAndStateImpl memberMetadata32 = createMemberMetadata("process3");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(mkEntry("member1_1", memberMetadata11), mkEntry("member1_2", memberMetadata12),
+ mkEntry("member2_1", memberMetadata21), mkEntry("member2_2", memberMetadata22),
+ mkEntry("member3_1", memberMetadata31), mkEntry("member3_2", memberMetadata32));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -135,9 +139,9 @@ public class StickyTaskAssignorTest {
@Test
public void shouldNotMigrateActiveTaskToOtherProcess() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
- AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- Map<String, AssignmentMemberSpec> members = mkMap(mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
+ MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -154,9 +158,9 @@ public class StickyTaskAssignorTest {
testMember1.activeTasks().get("test-subtopology").size() + testMember2.activeTasks().get("test-subtopology").size());
// flip the previous active tasks assignment around.
- memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
- members = mkMap(mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
+ members = mkMap(mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -175,11 +179,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldMigrateActiveTasksToNewProcessWithoutChangingAllAssignments() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 2))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 2))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -203,11 +207,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignBasedOnCapacity() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec21 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec22 = createAssignmentMemberSpec("process2");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2_1", memberSpec21), mkEntry("member2_2", memberSpec22));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata21 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata22 = createMemberMetadata("process2");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2_1", memberMetadata21), mkEntry("member2_2", memberMetadata22));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -230,10 +234,10 @@ public class StickyTaskAssignorTest {
final Map<String, Set<Integer>> activeTasks = mkMap(
mkEntry("test-subtopology1", Sets.newSet(0, 1, 2, 3, 4, 5)),
mkEntry("test-subtopology2", Sets.newSet(0)));
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", activeTasks, Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", activeTasks, Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -257,14 +261,14 @@ public class StickyTaskAssignorTest {
@Test
public void shouldKeepActiveTaskStickinessWhenMoreClientThanActiveTasks() {
- AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
- AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
- AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- AssignmentMemberSpec memberSpec4 = createAssignmentMemberSpec("process4");
- AssignmentMemberSpec memberSpec5 = createAssignmentMemberSpec("process5");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2),
- mkEntry("member3", memberSpec3), mkEntry("member4", memberSpec4), mkEntry("member5", memberSpec5));
+ MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
+ MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
+ MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ MemberMetadataAndStateImpl memberMetadata4 = createMemberMetadata("process4");
+ MemberMetadataAndStateImpl memberMetadata5 = createMemberMetadata("process5");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2),
+ mkEntry("member3", memberMetadata3), mkEntry("member4", memberMetadata4), mkEntry("member5", memberMetadata5));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -291,14 +295,14 @@ public class StickyTaskAssignorTest {
assertNull(testMember5.activeTasks().get("test-subtopology"));
// change up the assignment and make sure it is still sticky
- memberSpec1 = createAssignmentMemberSpec("process1");
- memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
- memberSpec3 = createAssignmentMemberSpec("process3");
- memberSpec4 = createAssignmentMemberSpec("process4", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
- memberSpec5 = createAssignmentMemberSpec("process5", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ memberMetadata1 = createMemberMetadata("process1");
+ memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
+ memberMetadata3 = createMemberMetadata("process3");
+ memberMetadata4 = createMemberMetadata("process4", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
+ memberMetadata5 = createMemberMetadata("process5", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2),
- mkEntry("member3", memberSpec3), mkEntry("member4", memberSpec4), mkEntry("member5", memberSpec5));
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2),
+ mkEntry("member3", memberMetadata3), mkEntry("member4", memberMetadata4), mkEntry("member5", memberMetadata5));
result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -327,11 +331,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksToClientWithPreviousStandbyTasks() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(2))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(1))));
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(0))));
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(2))));
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(1))));
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(0))));
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -354,10 +358,10 @@ public class StickyTaskAssignorTest {
@Test
public void shouldNotAssignStandbyTasksToClientWithPreviousStandbyTasksAndCurrentActiveTasks() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(0))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(1))));
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(0))));
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", Map.of(), mkMap(mkEntry("test-subtopology", Set.of(1))));
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, "1"))),
@@ -380,17 +384,17 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignBasedOnCapacityWhenMultipleClientHaveStandbyTasks() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
mkMap(mkEntry("test-subtopology", Set.of(0))),
mkMap(mkEntry("test-subtopology", Set.of(1))));
- final AssignmentMemberSpec memberSpec21 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata21 = createMemberMetadata("process2",
mkMap(mkEntry("test-subtopology", Set.of(2))),
mkMap(mkEntry("test-subtopology", Set.of(1))));
- final AssignmentMemberSpec memberSpec22 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata22 = createMemberMetadata("process2",
Map.of(), Map.of());
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1),
- mkEntry("member2_1", memberSpec21), mkEntry("member2_2", memberSpec22));
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1),
+ mkEntry("member2_1", memberMetadata21), mkEntry("member2_2", memberMetadata22));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -414,13 +418,13 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignStandbyTasksToDifferentClientThanCorrespondingActiveTaskIsAssignedTo() {
final Map<String, Set<Integer>> tasks = mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 3)));
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
- final AssignmentMemberSpec memberSpec4 = createAssignmentMemberSpec("process4", mkMap(mkEntry("test-subtopology", Set.of(3))), Map.of());
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2),
- mkEntry("member3", memberSpec3), mkEntry("member4", memberSpec4));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata4 = createMemberMetadata("process4", mkMap(mkEntry("test-subtopology", Set.of(3))), Map.of());
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2),
+ mkEntry("member3", memberMetadata3), mkEntry("member4", memberMetadata4));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -451,12 +455,12 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignMultipleReplicasOfStandbyTask() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2),
- mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Set.of(0))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Set.of(1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", mkMap(mkEntry("test-subtopology", Set.of(2))), Map.of());
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2),
+ mkEntry("member3", memberMetadata3));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -471,9 +475,9 @@ public class StickyTaskAssignorTest {
@Test
public void shouldNotAssignStandbyTaskReplicasWhenNoClientAvailableWithoutHavingTheTaskAssigned() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -486,12 +490,12 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignActiveAndStandbyTasks() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1),
- mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1),
+ mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -505,14 +509,14 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignAtLeastOneTaskToEachClientIfPossible() {
- final AssignmentMemberSpec memberSpec11 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec12 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec13 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1_1", memberSpec11), mkEntry("member1_2", memberSpec12), mkEntry("member1_3", memberSpec13),
- mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata11 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata12 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata13 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1_1", memberMetadata11), mkEntry("member1_2", memberMetadata12), mkEntry("member1_3", memberMetadata13),
+ mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -526,15 +530,15 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignEachActiveTaskToOneClientWhenMoreClientsThanTasks() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- final AssignmentMemberSpec memberSpec4 = createAssignmentMemberSpec("process4");
- final AssignmentMemberSpec memberSpec5 = createAssignmentMemberSpec("process5");
- final AssignmentMemberSpec memberSpec6 = createAssignmentMemberSpec("process6");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3),
- mkEntry("member4", memberSpec4), mkEntry("member5", memberSpec5), mkEntry("member6", memberSpec6));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ final MemberMetadataAndStateImpl memberMetadata4 = createMemberMetadata("process4");
+ final MemberMetadataAndStateImpl memberMetadata5 = createMemberMetadata("process5");
+ final MemberMetadataAndStateImpl memberMetadata6 = createMemberMetadata("process6");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3),
+ mkEntry("member4", memberMetadata4), mkEntry("member5", memberMetadata5), mkEntry("member6", memberMetadata6));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -547,15 +551,15 @@ public class StickyTaskAssignorTest {
@Test
public void shouldBalanceActiveAndStandbyTasksAcrossAvailableClients() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- final AssignmentMemberSpec memberSpec4 = createAssignmentMemberSpec("process4");
- final AssignmentMemberSpec memberSpec5 = createAssignmentMemberSpec("process5");
- final AssignmentMemberSpec memberSpec6 = createAssignmentMemberSpec("process6");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3),
- mkEntry("member4", memberSpec4), mkEntry("member5", memberSpec5), mkEntry("member6", memberSpec6));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ final MemberMetadataAndStateImpl memberMetadata4 = createMemberMetadata("process4");
+ final MemberMetadataAndStateImpl memberMetadata5 = createMemberMetadata("process5");
+ final MemberMetadataAndStateImpl memberMetadata6 = createMemberMetadata("process6");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3),
+ mkEntry("member4", memberMetadata4), mkEntry("member5", memberMetadata5), mkEntry("member6", memberMetadata6));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members,
@@ -570,11 +574,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignMoreTasksToClientWithMoreCapacity() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec21 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec22 = createAssignmentMemberSpec("process2");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2_1", memberSpec21), mkEntry("member2_2", memberSpec22));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata21 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata22 = createMemberMetadata("process2");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2_1", memberMetadata21), mkEntry("member2_2", memberMetadata22));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -587,12 +591,12 @@ public class StickyTaskAssignorTest {
@Test
public void shouldReBalanceTasksAcrossAllClientsWhenCapacityAndTaskCountTheSame() {
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 3))), Map.of());
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final AssignmentMemberSpec memberSpec4 = createAssignmentMemberSpec("process4");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3), mkEntry("member4", memberSpec4));
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 3))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final MemberMetadataAndStateImpl memberMetadata4 = createMemberMetadata("process4");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3), mkEntry("member4", memberMetadata4));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -607,11 +611,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldReBalanceTasksAcrossClientsWhenCapacityLessThanTaskCount() {
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 3))), Map.of());
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 3))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -625,11 +629,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldRebalanceTasksToClientsBasedOnCapacity() {
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 3, 2))), Map.of());
- final AssignmentMemberSpec memberSpec31 = createAssignmentMemberSpec("process3");
- final AssignmentMemberSpec memberSpec32 = createAssignmentMemberSpec("process3");
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member2", memberSpec2), mkEntry("member3_1", memberSpec31), mkEntry("member3_2", memberSpec32));
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 3, 2))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata31 = createMemberMetadata("process3");
+ final MemberMetadataAndStateImpl memberMetadata32 = createMemberMetadata("process3");
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member2", memberMetadata2), mkEntry("member3_1", memberMetadata31), mkEntry("member3_2", memberMetadata32));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -644,11 +648,11 @@ public class StickyTaskAssignorTest {
public void shouldMoveMinimalNumberOfTasksWhenPreviouslyAboveCapacityAndNewClientAdded() {
final Set<Integer> p1PrevTasks = Sets.newSet(0, 2);
final Set<Integer> p2PrevTasks = Sets.newSet(1, 3);
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", p1PrevTasks)), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", p2PrevTasks)), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", p1PrevTasks)), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", p2PrevTasks)), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -666,10 +670,10 @@ public class StickyTaskAssignorTest {
@Test
public void shouldNotMoveAnyTasksWhenNewTasksAdded() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(2, 3))), Map.of());
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(2, 3))), Map.of());
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -686,11 +690,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignNewTasksToNewClientWhenPreviousTasksAssignedToOldClients() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(2, 1))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 3))), Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(2, 1))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 3))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -710,20 +714,20 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksNotPreviouslyActiveToNewClient() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
mkMap(mkEntry("test-subtopology0", Sets.newSet(1)), mkEntry("test-subtopology1", Sets.newSet(2, 3))),
mkMap(mkEntry("test-subtopology0", Sets.newSet(0)), mkEntry("test-subtopology1", Sets.newSet(1)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2",
mkMap(mkEntry("test-subtopology0", Sets.newSet(0)), mkEntry("test-subtopology1", Sets.newSet(1)), mkEntry("test-subtopology2", Sets.newSet(2))),
mkMap(mkEntry("test-subtopology0", Sets.newSet(1, 2, 3)), mkEntry("test-subtopology1", Sets.newSet(0, 2, 3)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))));
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3",
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3",
mkMap(mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))),
mkMap(mkEntry("test-subtopology0", Sets.newSet(2)), mkEntry("test-subtopology1", Sets.newSet(2))));
- final AssignmentMemberSpec newMemberSpec = createAssignmentMemberSpec("process4",
+ final MemberMetadataAndStateImpl newMemberSpec = createMemberMetadata("process4",
Map.of(),
mkMap(mkEntry("test-subtopology0", Sets.newSet(0, 1, 2, 3)), mkEntry("test-subtopology1", Sets.newSet(0, 1, 2, 3)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 2, 3))));
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3), mkEntry("newMember", newMemberSpec));
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3), mkEntry("newMember", newMemberSpec));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -742,20 +746,20 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksNotPreviouslyActiveToMultipleNewClients() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
mkMap(mkEntry("test-subtopology0", Sets.newSet(1)), mkEntry("test-subtopology1", Sets.newSet(2, 3))),
mkMap(mkEntry("test-subtopology0", Sets.newSet(0)), mkEntry("test-subtopology1", Sets.newSet(1)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2",
mkMap(mkEntry("test-subtopology0", Sets.newSet(0)), mkEntry("test-subtopology1", Sets.newSet(1)), mkEntry("test-subtopology2", Sets.newSet(2))),
mkMap(mkEntry("test-subtopology0", Sets.newSet(1, 2, 3)), mkEntry("test-subtopology1", Sets.newSet(0, 2, 3)), mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))));
- final AssignmentMemberSpec bounce1 = createAssignmentMemberSpec("bounce1",
+ final MemberMetadataAndStateImpl bounce1 = createMemberMetadata("bounce1",
Map.of(),
mkMap(mkEntry("test-subtopology2", Sets.newSet(0, 1, 3))));
- final AssignmentMemberSpec bounce2 = createAssignmentMemberSpec("bounce2",
+ final MemberMetadataAndStateImpl bounce2 = createMemberMetadata("bounce2",
Map.of(),
mkMap(mkEntry("test-subtopology0", Sets.newSet(2, 3)), mkEntry("test-subtopology1", Sets.newSet(0))));
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("bounce_member1", bounce1), mkEntry("bounce_member2", bounce2));
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("bounce_member1", bounce1), mkEntry("bounce_member2", bounce2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -774,10 +778,10 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksToNewClient() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(1, 2))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(1, 2))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -789,11 +793,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksToNewClientWithoutFlippingAssignmentBetweenExistingClients() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(3, 4, 5))), Map.of());
- final AssignmentMemberSpec newMemberSpec = createAssignmentMemberSpec("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("newMember", newMemberSpec));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", mkMap(mkEntry("test-subtopology", Sets.newSet(3, 4, 5))), Map.of());
+ final MemberMetadataAndStateImpl newMemberSpec = createMemberMetadata("process3");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("newMember", newMemberSpec));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -815,11 +819,11 @@ public class StickyTaskAssignorTest {
@Test
public void shouldAssignTasksToNewClientWithoutFlippingAssignmentBetweenExistingAndBouncedClients() {
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 6))), Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2", Map.of(), mkMap(mkEntry("test-subtopology", Sets.newSet(3, 4, 5))));
- final AssignmentMemberSpec newMemberSpec = createAssignmentMemberSpec("newProcess");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("newMember", newMemberSpec));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1", mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1, 2, 6))), Map.of());
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2", Map.of(), mkMap(mkEntry("test-subtopology", Sets.newSet(3, 4, 5))));
+ final MemberMetadataAndStateImpl newMemberSpec = createMemberMetadata("newProcess");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("newMember", newMemberSpec));
GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, new HashMap<>()),
@@ -845,9 +849,9 @@ public class StickyTaskAssignorTest {
final int numClients = 5;
final int numStandbyReplicas = 2;
- Map<String, AssignmentMemberSpec> members = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> members = new HashMap<>();
for (int i = 0; i < numClients; i++) {
- members.put("member" + i, createAssignmentMemberSpec("process" + i));
+ members.put("member" + i, createMemberMetadata("process" + i));
}
GroupAssignment result = assignor.assign(
@@ -916,9 +920,9 @@ public class StickyTaskAssignorTest {
final int numClients = 7;
final int numStandbyReplicas = 1;
- Map<String, AssignmentMemberSpec> members = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> members = new HashMap<>();
for (int i = 0; i < numClients; i++) {
- members.put("member" + i, createAssignmentMemberSpec("process" + i));
+ members.put("member" + i, createMemberMetadata("process" + i));
}
GroupAssignment result = assignor.assign(
@@ -967,9 +971,9 @@ public class StickyTaskAssignorTest {
final int numClients = 3;
final int numStandbyReplicas = 5;
- Map<String, AssignmentMemberSpec> members = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> members = new HashMap<>();
for (int i = 0; i < numClients; i++) {
- members.put("member" + i, createAssignmentMemberSpec("process" + i));
+ members.put("member" + i, createMemberMetadata("process" + i));
}
GroupAssignment result = assignor.assign(
@@ -1011,9 +1015,9 @@ public class StickyTaskAssignorTest {
subtopologies.add("subtopology-" + i);
}
- Map<String, AssignmentMemberSpec> members = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> members = new HashMap<>();
for (int i = 0; i < numClients; i++) {
- members.put("member" + i, createAssignmentMemberSpec("process" + i));
+ members.put("member" + i, createMemberMetadata("process" + i));
}
GroupAssignment result = assignor.assign(
@@ -1044,8 +1048,8 @@ public class StickyTaskAssignorTest {
final int numTasks = 10;
final int numStandbyReplicas = 3;
- Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", createAssignmentMemberSpec("process1"))
+ Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", createMemberMetadata("process1"))
);
GroupAssignment result = assignor.assign(
@@ -1068,9 +1072,9 @@ public class StickyTaskAssignorTest {
final int numClients = 2;
final int numStandbyReplicas = 5; // More than available clients
- Map<String, AssignmentMemberSpec> members = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> members = new HashMap<>();
for (int i = 0; i < numClients; i++) {
- members.put("member" + i, createAssignmentMemberSpec("process" + i));
+ members.put("member" + i, createMemberMetadata("process" + i));
}
GroupAssignment result = assignor.assign(
@@ -1101,19 +1105,19 @@ public class StickyTaskAssignorTest {
public void shouldReassignTasksWhenNewNodeJoinsWithExistingActiveAndStandbyAssignments() {
// Initial setup: Node 1 has active tasks 0,1 and standby tasks 2,3
// Node 2 has active tasks 2,3 and standby tasks 0,1
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1))),
mkMap(mkEntry("test-subtopology", Sets.newSet(2, 3))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2",
mkMap(mkEntry("test-subtopology", Sets.newSet(2, 3))),
mkMap(mkEntry("test-subtopology", Sets.newSet(0, 1))));
// Node 3 joins as new client
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2), mkEntry("member3", memberSpec3));
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2), mkEntry("member3", memberMetadata3));
final GroupAssignment result = assignor.assign(
new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, "1"))),
@@ -1148,12 +1152,12 @@ public class StickyTaskAssignorTest {
@Test
public void shouldRangeAssignTasksWhenScalingUp() {
// Two clients, the second one is new
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
Map.of("test-subtopology1", Set.of(0, 1), "test-subtopology2", Set.of(0, 1)),
Map.of());
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
// Two subtopologies with 2 tasks each (4 tasks total) with standby replicas enabled
final GroupAssignment result = assignor.assign(
@@ -1196,10 +1200,10 @@ public class StickyTaskAssignorTest {
@Test
public void shouldRangeAssignTasksWhenStartingEmpty() {
// Two clients starting empty (no previous tasks)
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1");
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1), mkEntry("member2", memberSpec2));
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1");
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2");
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1), mkEntry("member2", memberMetadata2));
// Two subtopologies with 2 tasks each (4 tasks total) with standby replicas enabled
final GroupAssignment result = assignor.assign(
@@ -1257,18 +1261,18 @@ public class StickyTaskAssignorTest {
// Process1: active=[0], standby=[1] (previously had both active and standby tasks)
// Process2: active=[1] (had the active task that process1 had as standby)
// Process3: no previous tasks
- final AssignmentMemberSpec memberSpec1 = createAssignmentMemberSpec("process1",
+ final MemberMetadataAndStateImpl memberMetadata1 = createMemberMetadata("process1",
mkMap(mkEntry("test-subtopology", Sets.newSet(0))),
mkMap(mkEntry("test-subtopology", Sets.newSet(1))));
- final AssignmentMemberSpec memberSpec2 = createAssignmentMemberSpec("process2",
+ final MemberMetadataAndStateImpl memberMetadata2 = createMemberMetadata("process2",
mkMap(mkEntry("test-subtopology", Sets.newSet(1))),
Map.of());
- final AssignmentMemberSpec memberSpec3 = createAssignmentMemberSpec("process3");
+ final MemberMetadataAndStateImpl memberMetadata3 = createMemberMetadata("process3");
- final Map<String, AssignmentMemberSpec> members = mkMap(
- mkEntry("member1", memberSpec1),
- mkEntry("member2", memberSpec2),
- mkEntry("member3", memberSpec3));
+ final Map<String, MemberMetadataAndStateImpl> members = mkMap(
+ mkEntry("member1", memberMetadata1),
+ mkEntry("member2", memberMetadata2),
+ mkEntry("member3", memberMetadata3));
// We have 2 active tasks + 1 standby replica = 4 total tasks
// Quota per process = 4 tasks / 3 processes = 1.33 -> 2 tasks per process
@@ -1429,29 +1433,32 @@ public class StickyTaskAssignorTest {
return mergeAllStandbyTasks(result, result.members().keySet().toArray(memberIds));
}
- private AssignmentMemberSpec createAssignmentMemberSpec(final String processId) {
- return new AssignmentMemberSpec(
+ private MemberMetadataAndStateImpl createMemberMetadata(final String processId) {
+ return new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ processId,
Map.of(),
Map.of(),
Map.of(),
- processId,
Map.of(),
Map.of(),
Map.of());
}
- private AssignmentMemberSpec createAssignmentMemberSpec(final String processId, final Map<String, Set<Integer>> prevActiveTasks,
- final Map<String, Set<Integer>> prevStandbyTasks) {
- return new AssignmentMemberSpec(
+ private MemberMetadataAndStateImpl createMemberMetadata(
+ final String processId,
+ final Map<String, Set<Integer>> prevActiveTasks,
+ final Map<String, Set<Integer>> prevStandbyTasks
+ ) {
+ return new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ processId,
+ Map.of(),
prevActiveTasks,
prevStandbyTasks,
Map.of(),
- processId,
- Map.of(),
Map.of(),
Map.of());
}
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsAssignorBenchmarkUtils.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsAssignorBenchmarkUtils.java
index fa3324d012e..b1478498a3f 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsAssignorBenchmarkUtils.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsAssignorBenchmarkUtils.java
@@ -16,10 +16,10 @@
*/
package org.apache.kafka.jmh.assignor;
+import org.apache.kafka.coordinator.group.api.streams.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
-import org.apache.kafka.coordinator.group.streams.assignor.AssignmentMemberSpec;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupSpec;
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.ConfiguredInternalTopic;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
@@ -46,21 +46,21 @@ public class StreamsAssignorBenchmarkUtils {
Map<String, StreamsGroupMember> members,
Map<String, String> assignmentConfigs
) {
- Map<String, AssignmentMemberSpec> memberSpecs = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> memberSpecs = new HashMap<>();
// Prepare the member spec for all members.
for (Map.Entry<String, StreamsGroupMember> memberEntry : members.entrySet()) {
String memberId = memberEntry.getKey();
StreamsGroupMember member = memberEntry.getValue();
- memberSpecs.put(memberId, new AssignmentMemberSpec(
+ memberSpecs.put(memberId, new MemberMetadataAndStateImpl(
member.instanceId(),
member.rackId(),
+ member.processId(),
+ member.clientTags(),
Map.of(),
Map.of(),
Map.of(),
- member.processId(),
- member.clientTags(),
Map.of(),
Map.of()
));
diff --git a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java
index 22863fef5fa..b25c441cf82 100644
--- a/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java
+++ b/jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/StreamsStickyAssignorBenchmark.java
@@ -17,16 +17,16 @@
package org.apache.kafka.jmh.assignor;
import org.apache.kafka.coordinator.common.runtime.CoordinatorMetadataImage;
+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;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TaskAssignor;
+import org.apache.kafka.coordinator.group.api.streams.assignor.TopologyDescriber;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import org.apache.kafka.coordinator.group.streams.TopologyMetadata;
-import org.apache.kafka.coordinator.group.streams.assignor.AssignmentMemberSpec;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupAssignment;
-import org.apache.kafka.coordinator.group.streams.assignor.GroupSpec;
import org.apache.kafka.coordinator.group.streams.assignor.GroupSpecImpl;
-import org.apache.kafka.coordinator.group.streams.assignor.MemberAssignment;
+import org.apache.kafka.coordinator.group.streams.assignor.MemberMetadataAndStateImpl;
import org.apache.kafka.coordinator.group.streams.assignor.StickyTaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TaskAssignor;
-import org.apache.kafka.coordinator.group.streams.assignor.TopologyDescriber;
import org.apache.kafka.coordinator.group.streams.topics.ConfiguredSubtopology;
import org.openjdk.jmh.annotations.Benchmark;
@@ -132,34 +132,35 @@ public class StreamsStickyAssignorBenchmark {
GroupAssignment initialAssignment = new StickyTaskAssignor().assign(groupSpec, topologyDescriber);
Map<String, MemberAssignment> members = initialAssignment.members();
- Map<String, AssignmentMemberSpec> updatedMemberSpec = new HashMap<>();
+ Map<String, MemberMetadataAndStateImpl> updatedMemberSpec = new HashMap<>();
- for (Map.Entry<String, AssignmentMemberSpec> member : groupSpec.members().entrySet()) {
+ for (String memberId : groupSpec.memberIds()) {
MemberAssignment memberAssignment = members.getOrDefault(
- member.getKey(),
- new MemberAssignment(Map.of(), Map.of(), Map.of())
+ memberId,
+ new MemberAssignment(Map.of(), Map.of())
);
- updatedMemberSpec.put(member.getKey(), new AssignmentMemberSpec(
+ updatedMemberSpec.put(memberId, new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ groupSpec.memberMetadata(memberId).processId(),
+ Map.of(),
memberAssignment.activeTasks(),
memberAssignment.standbyTasks(),
- memberAssignment.warmupTasks(),
- member.getValue().processId(),
+ // Warm-up tasks are not assigned by the assignor; they are decided during reconciliation.
Map.of(),
Map.of(),
Map.of()
));
}
- updatedMemberSpec.put("newMember", new AssignmentMemberSpec(
+ updatedMemberSpec.put("newMember", new MemberMetadataAndStateImpl(
Optional.empty(),
Optional.empty(),
+ "process-newMember",
Map.of(),
Map.of(),
Map.of(),
- "process-newMember",
Map.of(),
Map.of(),
Map.of()