chickenchickenlove commented on code in PR #23742:
URL: https://github.com/apache/kafka/pull/23742#discussion_r4370857020
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/AssignmentTest.java:
##########
@@ -17,72 +17,140 @@
package org.apache.kafka.coordinator.group.modern;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.coordinator.group.assignor.AssignorHelpers;
import
org.apache.kafka.coordinator.group.generated.ConsumerGroupTargetAssignmentMemberValue;
+import
org.apache.kafka.coordinator.group.generated.ShareGroupTargetAssignmentMemberValue;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
import static
org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
public class AssignmentTest {
@Test
public void testPartitionsCannotBeNull() {
- assertThrows(NullPointerException.class, () -> new Assignment(null));
+ assertThrows(NullPointerException.class, () -> Assignment.of(null));
+ assertThrows(NullPointerException.class, () ->
Assignment.copyOf(null));
}
@Test
- public void testPartitionsImmutable() {
- // Assignments are used as input to assignors, which expect to receive
immutable
- // assignment maps, otherwise they will be modified in place.
+ public void testCopyOfPartitionsImmutable() {
+ Uuid topicId = Uuid.randomUuid();
Map<Uuid, Set<Integer>> partitions = new HashMap<>();
- partitions.put(Uuid.randomUuid(), Set.of(1, 2, 3));
- Assignment assignment = new Assignment(partitions);
- assertTrue(AssignorHelpers.isImmutableMap(assignment.partitions()));
+ partitions.put(topicId, new HashSet<>(Set.of(0, 1)));
+
+ Assignment assignment = Assignment.copyOf(partitions);
+
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
+ assertEquals(Map.of(topicId, Set.of(0, 1)), partitions);
+ }
+
+ @Test
+ public void testCopyOfIsolatedFromSourceSet() {
+ Uuid topicId = Uuid.randomUuid();
+ Set<Integer> partitionIds = new HashSet<>(Set.of(0, 1));
+ Map<Uuid, Set<Integer>> partitions = new HashMap<>();
+ partitions.put(topicId, partitionIds);
+
+ Assignment assignment = Assignment.copyOf(partitions);
+
+ partitionIds.remove(0);
+
+ assertEquals(Map.of(topicId, Set.of(0, 1)), assignment.partitions());
+ }
+
+ @Test
+ public void testCopyOfIsolatedFromSourceMap() {
+ Uuid topicId = Uuid.randomUuid();
+ Map<Uuid, Set<Integer>> partitions = new HashMap<>();
+ partitions.put(topicId, new HashSet<>(Set.of(0, 1)));
+
+ Assignment assignment = Assignment.copyOf(partitions);
+
+ partitions.clear();
+
+ assertEquals(Map.of(topicId, Set.of(0, 1)), assignment.partitions());
+ }
+
+ @Test
+ public void testOfReusesImmutablePartitionSets() {
+ Uuid topicId = Uuid.randomUuid();
+ Map<Uuid, Set<Integer>> partitions =
mkAssignment(mkTopicAssignment(topicId, 0, 1));
+
+ Assignment assignment = Assignment.of(partitions);
+
+ assertEquals(partitions, assignment.partitions());
+ assertSame(partitions.get(topicId),
assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
}
@Test
public void testAttributes() {
Map<Uuid, Set<Integer>> partitions = mkAssignment(
mkTopicAssignment(Uuid.randomUuid(), 1, 2, 3)
);
- Assignment assignment = new Assignment(partitions);
+ Assignment assignment = Assignment.copyOf(partitions);
assertEquals(partitions, assignment.partitions());
}
- @Test
- public void testFromTargetAssignmentRecord() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testFromTargetAssignmentRecord(boolean shareGroup) {
Uuid topicId1 = Uuid.randomUuid();
Uuid topicId2 = Uuid.randomUuid();
+ List<Integer> partitionIds1 = new ArrayList<>(List.of(1, 2, 3));
+ List<Integer> partitionIds2 = new ArrayList<>(List.of(4, 5, 6));
- List<ConsumerGroupTargetAssignmentMemberValue.TopicPartition>
partitions = new ArrayList<>();
- partitions.add(new
ConsumerGroupTargetAssignmentMemberValue.TopicPartition()
- .setTopicId(topicId1)
- .setPartitions(Arrays.asList(1, 2, 3)));
- partitions.add(new
ConsumerGroupTargetAssignmentMemberValue.TopicPartition()
- .setTopicId(topicId2)
- .setPartitions(Arrays.asList(4, 5, 6)));
-
- ConsumerGroupTargetAssignmentMemberValue record = new
ConsumerGroupTargetAssignmentMemberValue()
- .setTopicPartitions(partitions);
-
- Assignment assignment = Assignment.fromRecord(record);
+ Assignment assignment;
+ if (shareGroup) {
+ ShareGroupTargetAssignmentMemberValue record = new
ShareGroupTargetAssignmentMemberValue()
+ .setTopicPartitions(new ArrayList<>(List.of(
+ new ShareGroupTargetAssignmentMemberValue.TopicPartition()
+ .setTopicId(topicId1)
+ .setPartitions(partitionIds1),
+ new ShareGroupTargetAssignmentMemberValue.TopicPartition()
+ .setTopicId(topicId2)
+ .setPartitions(partitionIds2)
+ )));
+ assignment = Assignment.fromRecord(record);
+ record.topicPartitions().clear();
+ } else {
+ ConsumerGroupTargetAssignmentMemberValue record = new
ConsumerGroupTargetAssignmentMemberValue()
+ .setTopicPartitions(new ArrayList<>(List.of(
+ new
ConsumerGroupTargetAssignmentMemberValue.TopicPartition()
+ .setTopicId(topicId1)
+ .setPartitions(partitionIds1),
+ new
ConsumerGroupTargetAssignmentMemberValue.TopicPartition()
+ .setTopicId(topicId2)
+ .setPartitions(partitionIds2)
+ )));
+ assignment = Assignment.fromRecord(record);
+ record.topicPartitions().clear();
+ }
+ partitionIds1.clear();
+ partitionIds2.clear();
Review Comment:
My Fault, agreed..! (Sorry for the inconvenience)
I've reverted it in the next commit
--
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]
