mjsax commented on code in PR #22970:
URL: https://github.com/apache/kafka/pull/22970#discussion_r3671145851
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilderTest.java:
##########
@@ -177,6 +179,42 @@ public void
testCreateMemberMetadataAndStateSourcesWarmupTasksFromCurrentAssignm
assertEquals(Map.of(fooSubtopologyId, Set.of(1, 2, 3)),
memberMetadata.warmupTasks());
}
+ @Test
+ public void
testCreateMemberMetadataAndStateReturnsUnmodifiableCollections() {
Review Comment:
I think we can actually remove this test? It seems to replicate what we
already get from `MemberMetadataAndStateImplTest` so no need to duplicate test
coverage. We reach the same `MemberMetadataAndStateImpl` code just from our
more layer around it -- doesn't really add anything.
We should just move the client-tag test coverage that we only have here, to
the other test.
##########
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())));
+ }
+
+ private static Map<String, Map<Integer, Long>>
unmodifiableOffsets(Map<String, Map<Integer, Long>> offsets) {
+ return Objects.requireNonNull(offsets).entrySet().stream()
+ .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry ->
Collections.unmodifiableMap(entry.getValue())));
Review Comment:
```suggestion
.collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry
-> Map.copyOf(entry.getValue())));
```
##########
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:
```suggestion
.collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry
-> Set.copyOf(entry.getValue())));
```
--
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]
