MINOR: Make share group state partition metadata unmodifiable (#23888)

Wrap InitMapValue's partitions and
ShareGroupStatePartitionMetadataInfo's collections in unmodifiable
views, so that the share group state partition metadata cannot be
modified through them. Add withAddedPartitions and withRemovedPartitions
to InitMapValue to derive new values.

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.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Reviewers: David Jacot <dajac@apache.org>

Author: squah-confluent

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

@@ -3335,26 +3335,21 @@ static Map<Uuid, InitMapValue> combineInitMaps(
         Map<Uuid, InitMapValue> initialized,
         Map<Uuid, InitMapValue> initializing
     ) {
-        Map<Uuid, InitMapValue> finalInitMap = new HashMap<>();
         Set<Uuid> combinedTopicIdSet = new HashSet<>(initialized.keySet());
+        combinedTopicIdSet.addAll(initializing.keySet());
 
-        Set<Uuid> initializingSet = initializing.keySet();
-
-        combinedTopicIdSet.addAll(initializingSet);
+        Map<Uuid, InitMapValue> finalInitMap = new HashMap<>();
 
         for (Uuid topicId : combinedTopicIdSet) {
             InitMapValue initializedValue = initialized.get(topicId);
-            Set<Integer> 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<Void, CoordinatorRecord> initializeShareGroupState(
         // record has up-to-date information.
         Map<Uuid, InitMapValue> 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<Uuid, InitMapValue> 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<Uuid, Set<Integer>> entry : topicPartitionMap.entrySet()) {
+        for (Map.Entry<Uuid, InitMapValue> entry : currentMap.initializingTopics().entrySet()) {
             Uuid topicId = entry.getKey();
-            if (finalInitializingMap.containsKey(topicId)) {
-                Set<Integer> 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<Void, CoordinatorRecord> uninitializeShareGroupState(
             if (!topicPartitionMap.containsKey(topicId)) {
                 finalInitializingTopics.put(entry.getKey(), entry.getValue());
             } else {
-                Set<Integer> 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<Uuid, InitMapValue> attachInitValue(Map<Uuid, Set<Integer>> initMap)
     public Map<Uuid, Set<Integer>> initializedShareGroupPartitions(
         String groupId
     ) {
-        Map<Uuid, Set<Integer>> 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());
     }
 
     /**
@@ -9485,10 +9474,7 @@ public List<DeleteShareGroupStateRequestData.DeleteStateData> sharePartitionsEli
             return deleteShareGroupStateRequestTopicsData;
         }
 
-        Map<Uuid, InitMapValue> initializedTopics = new HashMap<>();
-        currentMap.initializedTopics().forEach((topicId, initValue) -> {
-            initializedTopics.put(topicId, new InitMapValue(initValue.name(), new HashSet<>(initValue.partitions()), initValue.timestamp()));
-        });
+        Map<Uuid, InitMapValue> initializedTopics = new HashMap<>(currentMap.initializedTopics());
         Set<Uuid> deletingTopics = new HashSet<>(currentMap.deletingTopics());
 
         requestData.topics().forEach(topic -> {

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.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -82,6 +84,11 @@ public record ShareGroupStatePartitionMetadataInfo(
         Map<Uuid, InitMapValue> initializedTopics,
         Set<Uuid> deletingTopics
     ) {
+        public ShareGroupStatePartitionMetadataInfo {
+            initializingTopics = Collections.unmodifiableMap(initializingTopics);
+            initializedTopics = Collections.unmodifiableMap(initializedTopics);
+            deletingTopics = Collections.unmodifiableSet(deletingTopics);
+        }
     }
 
     /**
@@ -97,6 +104,29 @@ public record InitMapValue(
         Set<Integer> partitions,
         long timestamp
     ) {
+        public InitMapValue {
+            partitions = Collections.unmodifiableSet(partitions);
+        }
+
+        /**
+         * @param added The partitions to add.
+         * @return A copy of this value with the partitions added.
+         */
+        public InitMapValue withAddedPartitions(Set<Integer> added) {
+            Set<Integer> 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(Set<Integer> removed) {
+            Set<Integer> newPartitions = new HashSet<>(partitions);
+            newPartitions.removeAll(removed);
+            return new InitMapValue(name, newPartitions, timestamp);
+        }
     }
 
     /**

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

@@ -30221,7 +30221,7 @@ public void testShareGroupInitializingClearsCommonDeleting() {
             )
         );
 
-        assertEquals(expectedRecords, records);
+        assertRecordsEquals(expectedRecords, records);
     }
 
     @Test

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");