From a81daf5e159ddacdb459ae8d5a241739002528e7 Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Tue, 2 Dec 2025 17:09:58 +0000 Subject: [PATCH 1/2] KAFKA-19955: Fix performance regression in server-side assignors The server-side assignors are designed to re-use the previous assignment maps where possible and clone them for modification lazily. #20097 introduced a bug where the assignments would always be re-wrapped in an unmodifiable map and so we would end up cloning them repeatedly in the assignors when multiple changes need to be made to a member assignment. Fix the bug by removing the unmodifiable map wrapping. --- .../group/modern/MemberAssignmentImpl.java | 3 +- .../modern/MemberAssignmentImplTest.java | 49 +++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImplTest.java diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java index 1cbad4b783b90..9a17de81938f9 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java @@ -19,7 +19,6 @@ import org.apache.kafka.common.Uuid; import org.apache.kafka.coordinator.group.api.assignor.MemberAssignment; -import java.util.Collections; import java.util.Map; import java.util.Objects; import java.util.Set; @@ -31,7 +30,7 @@ */ public record MemberAssignmentImpl(Map> partitions) implements MemberAssignment { public MemberAssignmentImpl { - partitions = Collections.unmodifiableMap(Objects.requireNonNull(partitions)); + Objects.requireNonNull(partitions); } /** diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImplTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImplTest.java new file mode 100644 index 0000000000000..fb0d3042ed44e --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImplTest.java @@ -0,0 +1,49 @@ +/* + * 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.modern; + +import org.apache.kafka.common.Uuid; + +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; + +public class MemberAssignmentImplTest { + + @Test + public void testPartitionsMutable() { + // We depend on the the map inside MemberAssignmentImpl remaining mutable in the server-side + // assignors, otherwise we end up deep copying the map unnecessarily. + Uuid topicId1 = Uuid.randomUuid(); + Uuid topicId2 = Uuid.randomUuid(); + + HashMap> partitions = new HashMap<>(); + partitions.put(topicId1, new HashSet<>()); + + MemberAssignmentImpl memberAssignment = new MemberAssignmentImpl(partitions); + + // The map should remain mutable. + assertDoesNotThrow(() -> memberAssignment.partitions().put(topicId2, Set.of(3, 4, 5))); + + // The sets inside the map should remain mutable. + assertDoesNotThrow(() -> memberAssignment.partitions().get(topicId1).add(3)); + } +} From 89e2c27f614cec5a72c510591377805855c30e6d Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Tue, 2 Dec 2025 17:30:44 +0000 Subject: [PATCH 2/2] fixup: add note to javadoc --- .../kafka/coordinator/group/modern/MemberAssignmentImpl.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java index 9a17de81938f9..251912036f5f0 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberAssignmentImpl.java @@ -27,6 +27,8 @@ * The partition assignment for a modern group member. * * @param partitions The partitions assigned to this member keyed by topicId. + * The map will not be made immutable, since the server-side assignors rely on + * being able to mutate the map while building new assignments. */ public record MemberAssignmentImpl(Map> partitions) implements MemberAssignment { public MemberAssignmentImpl {