From bf331b24f721d494bfcaa2f071fafddce76a0e04 Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Tue, 20 Oct 2026 18:20:00 +0100 Subject: [PATCH 1/5] MINOR: Make InitMapValue's partitions immutable InitMapValue now copies its partitions with Set.copyOf, so that the share group state partition metadata can be shared without defensive copies, such as with assignors offloaded to another thread. Add withAddedPartitions and withRemovedPartitions to derive new values, and a constructor taking a Collection for the partition lists in records. Use them in GroupMetadataManager and remove the copies that are no longer needed. combineInitMaps reuses a value when only one of the maps has the topic, and initializeShareGroupState no longer deep copies the initializing topics to remove partitions in place. initializedShareGroupPartitions and sharePartitionsEligibleForOffsetDeletion no longer copy the partitions. The partitions of the immutable sets are iterated in an unspecified order, so tests now compare them regardless of order. Co-Authored-By: Claude Opus 5.5 --- .../group/GroupMetadataManager.java | 68 ++++++++----------- .../group/modern/share/ShareGroup.java | 29 ++++++++ .../GroupCoordinatorRecordHelpersTest.java | 7 +- .../group/GroupMetadataManagerTest.java | 13 ++-- 4 files changed, 67 insertions(+), 50 deletions(-) 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 11ffc392366f4..73b04e9db57cf 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 @@ -3335,26 +3335,21 @@ static Map combineInitMaps( Map initialized, Map initializing ) { - Map finalInitMap = new HashMap<>(); Set combinedTopicIdSet = new HashSet<>(initialized.keySet()); + combinedTopicIdSet.addAll(initializing.keySet()); - Set initializingSet = initializing.keySet(); - - combinedTopicIdSet.addAll(initializingSet); + Map finalInitMap = new HashMap<>(); for (Uuid topicId : combinedTopicIdSet) { InitMapValue initializedValue = initialized.get(topicId); - Set finalPartitions = initializedValue != null ? new HashSet<>(initializedValue.partitions()) : new HashSet<>(); - long timestamp = initializedValue != null ? initializedValue.timestamp() : -1; - String name = initializedValue != null ? initializedValue.name() : "UNKNOWN"; - - if (initializingSet.contains(topicId)) { - InitMapValue initializingValue = initializing.get(topicId); - finalPartitions.addAll(initializingValue.partitions()); - timestamp = initializingValue.timestamp(); - name = initializingValue.name(); + InitMapValue initializingValue = initializing.get(topicId); + if (initializingValue == null) { + finalInitMap.put(topicId, initializedValue); + } else if (initializedValue == null) { + finalInitMap.put(topicId, initializingValue); + } else { + finalInitMap.put(topicId, initializingValue.withAddedPartitions(initializedValue.partitions())); } - finalInitMap.putIfAbsent(topicId, new InitMapValue(name, finalPartitions, timestamp)); } return finalInitMap; @@ -5913,18 +5908,16 @@ public CoordinatorResult initializeShareGroupState( // record has up-to-date information. Map finalInitializedMap = combineInitMaps(currentMap.initializedTopics(), enrichedTopicPartitionMap); - // Fetch initializing info from state metadata. + // Fetch initializing info from state metadata, and remove any entries which are already initialized. Map finalInitializingMap = new HashMap<>(); - currentMap.initializingTopics().forEach((k, v) -> finalInitializingMap.put(k, new InitMapValue(v.name(), new HashSet<>(v.partitions()), v.timestamp()))); - - // Remove any entries which are already initialized. - for (Map.Entry> entry : topicPartitionMap.entrySet()) { + for (Map.Entry entry : currentMap.initializingTopics().entrySet()) { Uuid topicId = entry.getKey(); - if (finalInitializingMap.containsKey(topicId)) { - Set partitions = finalInitializingMap.get(topicId).partitions(); - partitions.removeAll(entry.getValue()); - if (partitions.isEmpty()) { - finalInitializingMap.remove(topicId); + if (!topicPartitionMap.containsKey(topicId)) { + finalInitializingMap.put(topicId, entry.getValue()); + } else { + InitMapValue remaining = entry.getValue().withRemovedPartitions(topicPartitionMap.get(topicId)); + if (!remaining.partitions().isEmpty()) { + finalInitializingMap.put(topicId, remaining); } } } @@ -5966,10 +5959,9 @@ public CoordinatorResult uninitializeShareGroupState( if (!topicPartitionMap.containsKey(topicId)) { finalInitializingTopics.put(entry.getKey(), entry.getValue()); } else { - Set partitions = new HashSet<>(entry.getValue().partitions()); - partitions.removeAll(topicPartitionMap.get(topicId)); - if (!partitions.isEmpty()) { - finalInitializingTopics.put(entry.getKey(), new InitMapValue(entry.getValue().name(), partitions, entry.getValue().timestamp())); + InitMapValue remaining = entry.getValue().withRemovedPartitions(topicPartitionMap.get(topicId)); + if (!remaining.partitions().isEmpty()) { + finalInitializingTopics.put(entry.getKey(), remaining); } } } @@ -6017,16 +6009,13 @@ private Map attachInitValue(Map> initMap) public Map> initializedShareGroupPartitions( String groupId ) { - Map> resultMap = new HashMap<>(); - ShareGroupStatePartitionMetadataInfo currentMap = shareGroupStatePartitionMetadata.get(groupId); - if (currentMap != null) { - currentMap.initializedTopics().forEach((topicId, initMapValue) -> { - resultMap.put(topicId, new HashSet<>(initMapValue.partitions())); - }); + + if (currentMap == null) { + return Map.of(); } - return resultMap; + return stripInitValue(currentMap.initializedTopics()); } /** @@ -6816,13 +6805,13 @@ public void replay( value.initializingTopics().stream() .map(topicPartitionInfo -> Map.entry( topicPartitionInfo.topicId(), - new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp))) + new InitMapValue(topicPartitionInfo.topicName(), topicPartitionInfo.partitions(), timestamp))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), value.initializedTopics().stream() .map(topicPartitionInfo -> Map.entry( topicPartitionInfo.topicId(), - new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp))) + new InitMapValue(topicPartitionInfo.topicName(), topicPartitionInfo.partitions(), timestamp))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), value.deletingTopics().stream() @@ -9485,10 +9474,7 @@ public List sharePartitionsEli return deleteShareGroupStateRequestTopicsData; } - Map initializedTopics = new HashMap<>(); - currentMap.initializedTopics().forEach((topicId, initValue) -> { - initializedTopics.put(topicId, new InitMapValue(initValue.name(), new HashSet<>(initValue.partitions()), initValue.timestamp())); - }); + Map initializedTopics = new HashMap<>(currentMap.initializedTopics()); Set deletingTopics = new HashSet<>(currentMap.deletingTopics()); requestData.topics().forEach(topic -> { diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java index 1029f32ae3d52..e6fff74ad4506 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java @@ -32,6 +32,8 @@ import org.apache.kafka.timeline.SnapshotRegistry; import org.apache.kafka.timeline.TimelineObject; +import java.util.Collection; +import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; @@ -97,6 +99,33 @@ public record InitMapValue( Set partitions, long timestamp ) { + public InitMapValue { + partitions = Set.copyOf(partitions); + } + + public InitMapValue(String name, Collection partitions, long timestamp) { + this(name, Set.copyOf(partitions), timestamp); + } + + /** + * @param added The partitions to add. + * @return A copy of this value with the partitions added. + */ + public InitMapValue withAddedPartitions(Collection added) { + Set newPartitions = new HashSet<>(partitions); + newPartitions.addAll(added); + return new InitMapValue(name, newPartitions, timestamp); + } + + /** + * @param removed The partitions to remove. + * @return A copy of this value with the partitions removed. + */ + public InitMapValue withRemovedPartitions(Collection removed) { + Set newPartitions = new HashSet<>(partitions); + newPartitions.removeAll(removed); + return new InitMapValue(name, newPartitions, timestamp); + } } /** diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java index 0f941fa139135..d33e341e438e9 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java @@ -60,7 +60,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -215,9 +214,7 @@ public void testNewShareGroupPartitionMetadataRecord() { Uuid topicId1 = Uuid.randomUuid(); String topicName2 = "t2"; Uuid topicId2 = Uuid.randomUuid(); - Set partitions = new LinkedHashSet<>(); - partitions.add(0); - partitions.add(1); + Set partitions = Set.of(0, 1); CoordinatorRecord expectedRecord = CoordinatorRecord.record( new ShareGroupStatePartitionMetadataKey() @@ -229,7 +226,7 @@ public void testNewShareGroupPartitionMetadataRecord() { new ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo() .setTopicId(topicId1) .setTopicName(topicName1) - .setPartitions(List.of(0, 1)) + .setPartitions(List.copyOf(partitions)) ) ) .setDeletingTopics( 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 899c45693fb35..b421aebae2e84 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 @@ -19352,7 +19352,7 @@ public void testShareGroupDeleteTombstones() { ); List records = new ArrayList<>(); context.groupMetadataManager.createGroupTombstoneRecordsAndCancelTimers("share-group-id", records); - assertEquals(expectedRecords, records); + assertRecordsEquals(expectedRecords, records); } @Test @@ -29129,6 +29129,7 @@ topicId2, new InitMapValue(topicName2, Set.of(0, 1), 1) context.groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(groupId, requestData, errorTopicResponseList, records); assertTrue(errorTopicResponseList.isEmpty()); + result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29244,6 +29245,7 @@ topicId2, new InitMapValue(topicName2, Set.of(0, 1), 1) context.groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(groupId, requestData, errorTopicResponseList, records); assertTrue(errorTopicResponseList.isEmpty()); + result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29326,6 +29328,7 @@ public void testSharePartitionsEligibleForOffsetDeletionErrorTopics() { ), errorTopicResponseList ); + result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29410,6 +29413,7 @@ public void testSharePartitionsEligibleForOffsetDeletionUninitializedTopics() { ), errorTopicResponseList ); + result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29501,6 +29505,7 @@ public void testSharePartitionsEligibleForOffsetDeletionUninitializedAndErrorTop ), errorTopicResponseList ); + result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -30221,7 +30226,7 @@ public void testShareGroupInitializingClearsCommonDeleting() { ) ); - assertEquals(expectedRecords, records); + assertRecordsEquals(expectedRecords, records); } @Test @@ -30594,8 +30599,8 @@ public void testMaybeCleanupShareGroupStateInitDeletedTopicsPresent() { ) ); - CoordinatorResult expectedResult = new CoordinatorResult<>(expectedRecords); - assertEquals(expectedResult, context.groupMetadataManager.maybeCleanupShareGroupState(Set.of(t1Id, t2Id, t6Id))); + CoordinatorResult result = context.groupMetadataManager.maybeCleanupShareGroupState(Set.of(t1Id, t2Id, t6Id)); + assertRecordsEquals(expectedRecords, result.records()); } @Test From 096acb8051b7738cca0df52fb7b4d8555db8f97e Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Wed, 21 Oct 2026 11:44:03 +0100 Subject: [PATCH 2/5] fixup: Wrap InitMapValue's partitions in an unmodifiable set instead of copying them Co-Authored-By: Claude Opus 5.5 --- .../kafka/coordinator/group/GroupMetadataManager.java | 4 ++-- .../coordinator/group/modern/share/ShareGroup.java | 7 ++----- .../group/GroupCoordinatorRecordHelpersTest.java | 7 +++++-- .../coordinator/group/GroupMetadataManagerTest.java | 11 +++-------- 4 files changed, 12 insertions(+), 17 deletions(-) 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 73b04e9db57cf..61a2b3c29b25d 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 @@ -6805,13 +6805,13 @@ public void replay( value.initializingTopics().stream() .map(topicPartitionInfo -> Map.entry( topicPartitionInfo.topicId(), - new InitMapValue(topicPartitionInfo.topicName(), topicPartitionInfo.partitions(), timestamp))) + new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), value.initializedTopics().stream() .map(topicPartitionInfo -> Map.entry( topicPartitionInfo.topicId(), - new InitMapValue(topicPartitionInfo.topicName(), topicPartitionInfo.partitions(), timestamp))) + new InitMapValue(topicPartitionInfo.topicName(), new HashSet<>(topicPartitionInfo.partitions()), timestamp))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)), value.deletingTopics().stream() diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java index e6fff74ad4506..aed3ce3379872 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java @@ -33,6 +33,7 @@ import org.apache.kafka.timeline.TimelineObject; import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Locale; @@ -100,11 +101,7 @@ public record InitMapValue( long timestamp ) { public InitMapValue { - partitions = Set.copyOf(partitions); - } - - public InitMapValue(String name, Collection partitions, long timestamp) { - this(name, Set.copyOf(partitions), timestamp); + partitions = Collections.unmodifiableSet(partitions); } /** diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java index d33e341e438e9..0f941fa139135 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpersTest.java @@ -60,6 +60,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -214,7 +215,9 @@ public void testNewShareGroupPartitionMetadataRecord() { Uuid topicId1 = Uuid.randomUuid(); String topicName2 = "t2"; Uuid topicId2 = Uuid.randomUuid(); - Set partitions = Set.of(0, 1); + Set partitions = new LinkedHashSet<>(); + partitions.add(0); + partitions.add(1); CoordinatorRecord expectedRecord = CoordinatorRecord.record( new ShareGroupStatePartitionMetadataKey() @@ -226,7 +229,7 @@ public void testNewShareGroupPartitionMetadataRecord() { new ShareGroupStatePartitionMetadataValue.TopicPartitionsInfo() .setTopicId(topicId1) .setTopicName(topicName1) - .setPartitions(List.copyOf(partitions)) + .setPartitions(List.of(0, 1)) ) ) .setDeletingTopics( 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 b421aebae2e84..a89ba8c20f7bf 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 @@ -19352,7 +19352,7 @@ public void testShareGroupDeleteTombstones() { ); List records = new ArrayList<>(); context.groupMetadataManager.createGroupTombstoneRecordsAndCancelTimers("share-group-id", records); - assertRecordsEquals(expectedRecords, records); + assertEquals(expectedRecords, records); } @Test @@ -29129,7 +29129,6 @@ topicId2, new InitMapValue(topicName2, Set.of(0, 1), 1) context.groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(groupId, requestData, errorTopicResponseList, records); assertTrue(errorTopicResponseList.isEmpty()); - result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29245,7 +29244,6 @@ topicId2, new InitMapValue(topicName2, Set.of(0, 1), 1) context.groupMetadataManager.sharePartitionsEligibleForOffsetDeletion(groupId, requestData, errorTopicResponseList, records); assertTrue(errorTopicResponseList.isEmpty()); - result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29328,7 +29326,6 @@ public void testSharePartitionsEligibleForOffsetDeletionErrorTopics() { ), errorTopicResponseList ); - result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29413,7 +29410,6 @@ public void testSharePartitionsEligibleForOffsetDeletionUninitializedTopics() { ), errorTopicResponseList ); - result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -29505,7 +29501,6 @@ public void testSharePartitionsEligibleForOffsetDeletionUninitializedAndErrorTop ), errorTopicResponseList ); - result.forEach(topic -> topic.partitions().sort(Comparator.comparingInt(DeleteShareGroupStateRequestData.PartitionData::partition))); assertEquals(expectedResult, result); assertRecordsEquals(expectedRecords, records); } @@ -30599,8 +30594,8 @@ public void testMaybeCleanupShareGroupStateInitDeletedTopicsPresent() { ) ); - CoordinatorResult result = context.groupMetadataManager.maybeCleanupShareGroupState(Set.of(t1Id, t2Id, t6Id)); - assertRecordsEquals(expectedRecords, result.records()); + CoordinatorResult expectedResult = new CoordinatorResult<>(expectedRecords); + assertEquals(expectedResult, context.groupMetadataManager.maybeCleanupShareGroupState(Set.of(t1Id, t2Id, t6Id))); } @Test From 1189796b6b2dfbfd4f7a9c53e0fb2bbed5cfcb37 Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Wed, 21 Oct 2026 11:49:02 +0100 Subject: [PATCH 3/5] MINOR: Make ShareGroupStatePartitionMetadataInfo's collections unmodifiable Co-Authored-By: Claude Opus 5.5 --- .../kafka/coordinator/group/modern/share/ShareGroup.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java index aed3ce3379872..b444a56c4d1a1 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java @@ -85,6 +85,11 @@ public record ShareGroupStatePartitionMetadataInfo( Map initializedTopics, Set deletingTopics ) { + public ShareGroupStatePartitionMetadataInfo { + initializingTopics = Collections.unmodifiableMap(initializingTopics); + initializedTopics = Collections.unmodifiableMap(initializedTopics); + deletingTopics = Collections.unmodifiableSet(deletingTopics); + } } /** From 674eb422acb67ae8132ce99f463f87700faf5d1c Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Wed, 21 Oct 2026 14:54:43 +0100 Subject: [PATCH 4/5] fixup: Take a Set in InitMapValue's withAddedPartitions and withRemovedPartitions Co-Authored-By: Claude Opus 5.5 --- .../kafka/coordinator/group/modern/share/ShareGroup.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java index b444a56c4d1a1..a4eeb62b801b4 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java @@ -32,7 +32,6 @@ import org.apache.kafka.timeline.SnapshotRegistry; import org.apache.kafka.timeline.TimelineObject; -import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -113,7 +112,7 @@ public record InitMapValue( * @param added The partitions to add. * @return A copy of this value with the partitions added. */ - public InitMapValue withAddedPartitions(Collection added) { + public InitMapValue withAddedPartitions(Set added) { Set newPartitions = new HashSet<>(partitions); newPartitions.addAll(added); return new InitMapValue(name, newPartitions, timestamp); @@ -123,7 +122,7 @@ public InitMapValue withAddedPartitions(Collection added) { * @param removed The partitions to remove. * @return A copy of this value with the partitions removed. */ - public InitMapValue withRemovedPartitions(Collection removed) { + public InitMapValue withRemovedPartitions(Set removed) { Set newPartitions = new HashSet<>(partitions); newPartitions.removeAll(removed); return new InitMapValue(name, newPartitions, timestamp); From e9c0c9dbb19f35b54061c3c97d8a6584dcac5f9b Mon Sep 17 00:00:00 2001 From: Sean Quah Date: Wed, 21 Oct 2026 14:59:02 +0100 Subject: [PATCH 5/5] fixup: Test InitMapValue's withAddedPartitions and withRemovedPartitions Co-Authored-By: Claude Opus 5.5 --- .../group/modern/share/ShareGroupTest.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupTest.java index f03c24e0f53e1..e426f68000d4d 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupTest.java @@ -30,6 +30,7 @@ import org.apache.kafka.coordinator.group.Group; import org.apache.kafka.coordinator.group.modern.Assignment; import org.apache.kafka.coordinator.group.modern.MemberState; +import org.apache.kafka.coordinator.group.modern.share.ShareGroup.InitMapValue; import org.apache.kafka.coordinator.group.modern.share.ShareGroup.ShareGroupState; import org.apache.kafka.timeline.SnapshotRegistry; @@ -53,6 +54,18 @@ public class ShareGroupTest { + @Test + public void testInitMapValueWithAddedPartitions() { + InitMapValue initMapValue = new InitMapValue("foo", Set.of(0, 1), 1L); + assertEquals(new InitMapValue("foo", Set.of(0, 1, 2), 1L), initMapValue.withAddedPartitions(Set.of(1, 2))); + } + + @Test + public void testInitMapValueWithRemovedPartitions() { + InitMapValue initMapValue = new InitMapValue("foo", Set.of(0, 1), 1L); + assertEquals(new InitMapValue("foo", Set.of(0), 1L), initMapValue.withRemovedPartitions(Set.of(1, 2))); + } + @Test public void testType() { ShareGroup shareGroup = createShareGroup("foo");