This is an automated email from the ASF dual-hosted git repository.
squah-confluent 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 4861297e909 KAFKA-20292 [2/N]: Prepare to split
TargetAssignmentBuilders (#22486)
4861297e909 is described below
commit 4861297e9099c1e9603dcd57fd25ffb95552b19c
Author: Sean Quah <[email protected]>
AuthorDate: Sun Jun 28 00:13:27 2026 +0100
KAFKA-20292 [2/N]: Prepare to split TargetAssignmentBuilders (#22486)
To accommodate asynchronous assignments, such as those from client-side
assignors and assignors offloaded to background threads, we want to
split the TargetAssignmentBuilders into two: one builder for building
the target assignment and another for building the target assignment
records. Client-side assignors will only use the second builder.
Both builders require an up-to-date view of group members at the time
they are run. In the non-offloaded case, this is the same view. However,
the view needs to include the unwritten member operations from the
ongoing heartbeat request. Currently the operations are applied within
the TargetAssignmentBuilders. To avoid duplicating the logic once the
TargetAssignmentBuilders are split, we would like to lift it out and
pass the TargetAssignmentBuilders the updated view of members and
assignments.
Add UpdatedMembersAndTargetAssignmentView, which provides updated views
of members and target assignments after unwritten member joins, updates
and leaves. Future commits will update the consumer, share and streams
group assignment paths to use the new class.
Reviewers: David Jacot <[email protected]>
---
.../UpdatedMembersAndTargetAssignmentView.java | 123 +++++++++++++
.../UpdatedMembersAndTargetAssignmentViewTest.java | 191 +++++++++++++++++++++
2 files changed, 314 insertions(+)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentView.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentView.java
new file mode 100644
index 00000000000..5494c69854e
--- /dev/null
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentView.java
@@ -0,0 +1,123 @@
+/*
+ * 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.util;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A view of a group's members, static members, and target assignment after
unwritten membership
+ * changes have been applied.
+ *
+ * @param <M> The member type.
+ * @param <A> The member's target assignment type.
+ */
+public class UpdatedMembersAndTargetAssignmentView<M, A> {
+
+ /**
+ * The group members.
+ */
+ private final OverlayMap<String, M> members;
+
+ /**
+ * The static group members.
+ */
+ private final OverlayMap<String, String> staticMembers;
+
+ /**
+ * The target assignment per member id.
+ */
+ private final OverlayMap<String, A> targetAssignment;
+
+ /**
+ * @param members The group members. Must not be modified during
the lifetime of the view.
+ * @param staticMembers The static group members. Must not be modified
during the lifetime of the view.
+ * @param targetAssignment The target assignment per member id. Must not
be modified during the lifetime of the view.
+ */
+ public UpdatedMembersAndTargetAssignmentView(
+ Map<String, M> members,
+ Map<String, String> staticMembers,
+ Map<String, A> targetAssignment
+ ) {
+ this.members = new OverlayMap<>(Objects.requireNonNull(members));
+ this.staticMembers = new
OverlayMap<>(Objects.requireNonNull(staticMembers));
+ this.targetAssignment = new
OverlayMap<>(Objects.requireNonNull(targetAssignment));
+ }
+
+ /**
+ * @return The group members after updates.
+ */
+ public Map<String, M> members() {
+ return Collections.unmodifiableMap(members);
+ }
+
+ /**
+ * @return The static group members after updates.
+ */
+ public Map<String, String> staticMembers() {
+ return Collections.unmodifiableMap(staticMembers);
+ }
+
+ /**
+ * @return The target assignment per member id after updates.
+ */
+ public Map<String, A> targetAssignment() {
+ return Collections.unmodifiableMap(targetAssignment);
+ }
+
+ /**
+ * Adds or updates a member. If the member is static and there is a
different existing static
+ * member for the same instance id, the previous static member's target
assignment is moved to
+ * the new member and the previous static member is removed from the view.
+ *
+ * @param memberId The member id.
+ * @param instanceId The instance id of the member, or {@code null} if the
member is not static.
+ * @param member The member to add or update.
+ */
+ public void addOrUpdateMember(String memberId, String instanceId, M
member) {
+ members.put(memberId, member);
+ if (instanceId != null) {
+ String previousMemberId = staticMembers.put(instanceId, memberId);
+ if (previousMemberId != null &&
!memberId.equals(previousMemberId)) {
+ // A static member is being replaced. Move the assignment to
the new member.
+ A memberAssignment = targetAssignment.get(previousMemberId);
+ if (memberAssignment != null) {
+ targetAssignment.put(memberId, memberAssignment);
+ }
+
+ // Remove the previous member.
+ members.remove(previousMemberId);
+ targetAssignment.remove(previousMemberId);
+ }
+ }
+ }
+
+ /**
+ * Removes a member.
+ *
+ * @param memberId The member id.
+ * @param instanceId The instance id of the member, or {@code null} if the
member is not static.
+ */
+ public void removeMember(String memberId, String instanceId) {
+ members.remove(memberId);
+ if (instanceId != null &&
memberId.equals(staticMembers.get(instanceId))) {
+ staticMembers.remove(instanceId);
+ }
+ targetAssignment.remove(memberId);
+ }
+}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentViewTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentViewTest.java
new file mode 100644
index 00000000000..84be50efca0
--- /dev/null
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/util/UpdatedMembersAndTargetAssignmentViewTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class UpdatedMembersAndTargetAssignmentViewTest {
+
+ /**
+ * Creates an {@link UpdatedMembersAndTargetAssignmentView} with two
members, one static and one
+ * non-static.
+ */
+ private static UpdatedMembersAndTargetAssignmentView<String, String>
createView() {
+ return new UpdatedMembersAndTargetAssignmentView<>(
+ Map.of(
+ "member-1", "Member1",
+ "member-2", "Member2"
+ ),
+ Map.of(
+ "instance-id", "member-2"
+ ),
+ Map.of(
+ "member-1", "Assignment-member-1",
+ "member-2", "Assignment-member-2"
+ )
+ );
+ }
+
+ @Test
+ public void testAddMember() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.addOrUpdateMember("member-3", null, "Member3");
+
+ assertEquals(Map.of(
+ "member-1", "Member1",
+ "member-2", "Member2",
+ "member-3", "Member3"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-2"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-2", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testAddStaticMember() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.addOrUpdateMember("member-3", "instance-id-2", "Member3");
+
+ assertEquals(Map.of(
+ "member-1", "Member1",
+ "member-2", "Member2",
+ "member-3", "Member3"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-2",
+ "instance-id-2", "member-3"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-2", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testReplaceMember() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.addOrUpdateMember("member-1", null, "Member1-updated");
+
+ assertEquals(Map.of(
+ "member-1", "Member1-updated",
+ "member-2", "Member2"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-2"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-2", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testReplaceStaticMemberWithSameMemberId() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.addOrUpdateMember("member-2", "instance-id", "Member2-updated");
+
+ assertEquals(Map.of(
+ "member-1", "Member1",
+ "member-2", "Member2-updated"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-2"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-2", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testReplaceStaticMemberWithDifferentMemberId() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.addOrUpdateMember("member-3", "instance-id", "Member3");
+
+ assertEquals(Map.of(
+ "member-1", "Member1",
+ "member-3", "Member3"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-3"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-3", "Assignment-member-2"
+ ), view.targetAssignment());
+
+ // Removing the previous static member does not change the new static
member's assignment.
+ view.removeMember("member-2", "instance-id");
+
+ assertEquals(Map.of(
+ "member-1", "Member1",
+ "member-3", "Member3"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-3"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1",
+ "member-3", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testRemoveMember() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.removeMember("member-1", null);
+
+ assertEquals(Map.of(
+ "member-2", "Member2"
+ ), view.members());
+ assertEquals(Map.of(
+ "instance-id", "member-2"
+ ), view.staticMembers());
+ assertEquals(Map.of(
+ "member-2", "Assignment-member-2"
+ ), view.targetAssignment());
+ }
+
+ @Test
+ public void testRemoveStaticMember() {
+ UpdatedMembersAndTargetAssignmentView<String, String> view =
createView();
+
+ view.removeMember("member-2", "instance-id");
+
+ assertEquals(Map.of(
+ "member-1", "Member1"
+ ), view.members());
+ assertEquals(Map.of(), view.staticMembers());
+ assertEquals(Map.of(
+ "member-1", "Assignment-member-1"
+ ), view.targetAssignment());
+ }
+}
