gabriellefu commented on code in PR #22970:
URL: https://github.com/apache/kafka/pull/22970#discussion_r3693153936
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/MemberMetadataAndStateImpl.java:
##########
@@ -56,11 +57,23 @@ public record MemberMetadataAndStateImpl(
Objects.requireNonNull(rackId);
Objects.requireNonNull(processId);
clientTags =
Collections.unmodifiableMap(Objects.requireNonNull(clientTags));
- activeTasks =
Collections.unmodifiableMap(Objects.requireNonNull(activeTasks));
- standbyTasks =
Collections.unmodifiableMap(Objects.requireNonNull(standbyTasks));
- warmupTasks =
Collections.unmodifiableMap(Objects.requireNonNull(warmupTasks));
- taskOffsets =
Collections.unmodifiableMap(Objects.requireNonNull(taskOffsets));
- taskEndOffsets =
Collections.unmodifiableMap(Objects.requireNonNull(taskEndOffsets));
+ // These collections belong to the coordinator (the member's current
assignment and the offsets reported in
+ // the last heartbeat) and are only unmodifiable at the outer level
there, so wrap the nested ones as well.
+ activeTasks = unmodifiableTasks(activeTasks);
+ standbyTasks = unmodifiableTasks(standbyTasks);
+ warmupTasks = unmodifiableTasks(warmupTasks);
+ taskOffsets = unmodifiableOffsets(taskOffsets);
+ taskEndOffsets = unmodifiableOffsets(taskEndOffsets);
+ }
+
+ private static Map<String, Set<Integer>> unmodifiableTasks(Map<String,
Set<Integer>> tasks) {
+ return Objects.requireNonNull(tasks).entrySet().stream()
+ .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry ->
Collections.unmodifiableSet(entry.getValue())));
Review Comment:
change the function to make it unmodifiable and deepcopy()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
