Skip to content

MINOR: Make share group state partition metadata unmodifiable #23888

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
squah-confluent merged 5 commits into from
Oct 21, 2026
Merged

MINOR: Make share group state partition metadata unmodifiable #23888

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
bf331b2
MINOR: Make InitMapValue's partitions immutable
squah-confluent Oct 20, 2026
096acb8
fixup: Wrap InitMapValue's partitions in an unmodifiable set instead …
squah-confluent Oct 21, 2026
1189796
MINOR: Make ShareGroupStatePartitionMetadataInfo's collections unmodi…
squah-confluent Oct 21, 2026
674eb42
fixup: Take a Set in InitMapValue's withAddedPartitions and withRemov…
squah-confluent Oct 21, 2026
e9c0c9d
fixup: Test InitMapValue's withAddedPartitions and withRemovedPartitions
squah-confluent Oct 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 25 additions & 39 deletions group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
}
}
Expand Down Expand Up @@ -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);
}
}
}
Expand Down Expand Up @@ -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());
}

/**
Expand Down Expand Up @@ -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 -> {
Expand Down
30 changes: 30 additions & 0 deletions ...coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
}

/**
Expand All @@ -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);
Comment on lines +116 to +118

@dajac dajac Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is a tad wasteful to allocate an HashSet here to then copy it in InitMapValue. Have you considered using unmodifiableSet?

@squah-confluent squah-confluent Oct 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you mean replacing Set.copyOf with Collections.unmodifiableSet in the constructor or passing an unmodifiableSet wrapper in to the existing constructor?

@dajac dajac Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Either way is fine. I don't really have a preference and I don't think that we have established a pattern (yet).

@squah-confluent squah-confluent Oct 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prefer copyOf in the constructor since it's truly immutable and it's not possible to misuse by holding on to the original collection and modifying it. But it does end up with more copies overall.

I think our original convention was to use Collections.unmodifiableMap/Set in constructors, so I'll go with that. Though the streams code is pulling a different way and going with copyOf lately...

@squah-confluent squah-confluent Oct 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The current version of the PR uses unmodifiableMap/Set.

}

/**
* @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);
Comment on lines +126 to +128

@dajac dajac Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto.

}
}

/**
Expand Down
2 changes: 1 addition & 1 deletion ...oordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30221,7 +30221,7 @@ public void testShareGroupInitializingClearsCommonDeleting() {
)
);

assertEquals(expectedRecords, records);
assertRecordsEquals(expectedRecords, records);
}

@Test
Expand Down
13 changes: 13 additions & 0 deletions ...dinator/src/test/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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");
Expand Down
Loading