squah-confluent commented on code in PR #23742:
URL: https://github.com/apache/kafka/pull/23742#discussion_r4347780761
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilder.java:
##########
@@ -158,10 +158,13 @@ private Assignment newMemberAssignment(
String memberId
) {
MemberAssignment newMemberAssignment =
newGroupAssignment.members().get(memberId);
- if (newMemberAssignment != null) {
- return new Assignment(newMemberAssignment.partitions());
- } else {
+ if (newMemberAssignment == null) {
return Assignment.EMPTY;
}
+
+ // The output is only read by the coordinator. Assignor inputs are
+ // reconstructed from target assignment records during replay.
+ // So, we don't need defensive copy here.
+ return Assignment.of(newMemberAssignment.partitions());
Review Comment:
Are we trying to guard against
1) a badly written custom assignor updating the sets after it has returned
them, or
2) coordinator code reaching into the assignment and updating the
`Set<Integer>`?
The reasoning works for 2) but not for 1). When assignor offload is enabled,
the record-building step can be delayed until after another group invokes the
custom assignor.
##########
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()));
Review Comment:
We've lost the `AssignorHelpers.isImmutableMap` assert which is the
particular behavior that the existing uniform assignors depend upon. The write
operations throwing does not imply that `AssignorHelpers.isImmutableMap` will
return `true`. `Map.copyOf` return values are immutable but aren't counted by
`AssignorHelpers.isImmutableMap` for example.
##########
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());
+ }
Review Comment:
nit: There's a lot of overhead in these tests just for one line of actual
interesting setup. We could combine the two.
##########
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();
assertEquals(mkAssignment(
mkTopicAssignment(topicId1, 1, 2, 3),
mkTopicAssignment(topicId2, 4, 5, 6)
), assignment.partitions());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId1).remove(1));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId2).remove(4));
Review Comment:
We've already tested this in the immutability tests above.
```suggestion
```
##########
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));
Review Comment:
We don't assert equality in the `copyOf` immutability test, where it is
arguably more likely to break.
Shall we add equality tests to `copyOf`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilderTest.java:
##########
@@ -170,20 +177,91 @@ public void testAssignment() {
.assign(groupSpec, new
SubscribedTopicDescriberImpl(metadataImage));
Map<String, Assignment> expectedAssignment = new HashMap<>();
- expectedAssignment.put("member-1", new Assignment(mkAssignment(
+ expectedAssignment.put("member-1", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
- expectedAssignment.put("member-2", new Assignment(mkAssignment(
+ expectedAssignment.put("member-2", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 3, 4, 5)
)));
- expectedAssignment.put("member-3", new Assignment(mkAssignment(
+ expectedAssignment.put("member-3", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 6),
mkTopicAssignment(barTopicId, 6)
)));
assertEquals(expectedAssignment, result.targetAssignment());
assertEquals(new TargetAssignmentMetadata(20, 12345L),
result.targetAssignmentMetadata());
}
+
+ @Test
+ public void testUnchangedAssignmentReusesPartitionSets() {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(previousAssignment.partitions())
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(previousAssignment.partitions().get(topicId),
assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAssignmentReusesAssignorResultPartitionSets(boolean
unmodifiableMap) {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+ Set<Integer> partitionIds = new HashSet<>(Set.of(0, 1));
+ Map<Uuid, Set<Integer>> partitions = new HashMap<>();
+ partitions.put(topicId, partitionIds);
+
+ Map<Uuid, Set<Integer>> returnedPartitions = unmodifiableMap
+ ? Collections.unmodifiableMap(partitions)
+ : partitions;
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(returnedPartitions)
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(partitionIds, assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ }
+
+ @Test
+ public void testMissingMemberAssignmentReturnsEmpty() {
+ Assignment previousAssignment = Assignment.of(mkAssignment(
+ mkTopicAssignment(Uuid.randomUuid(), 0, 1)
+ ));
+
+ assertSame(Assignment.EMPTY, buildTargetAssignment(previousAssignment,
Map.of()));
+ }
+
+ private Assignment buildTargetAssignment(
+ Assignment previousAssignment,
+ Map<String, MemberAssignment> memberAssignments
+ ) {
+ GroupSpec groupSpec = new GroupSpecImpl(
+ Map.of("member", new MemberSubscriptionAndAssignmentImpl(
+ Optional.empty(),
+ Optional.empty(),
+ previousAssignment.partitions().keySet(),
+ previousAssignment
+ )),
+ HOMOGENEOUS,
+ Map.of()
+ );
+ PartitionAssignor assignor = mock(PartitionAssignor.class);
+ when(assignor.assign(any(), any())).thenReturn(new
GroupAssignment(memberAssignments));
+
+ return new TargetAssignmentBuilder(20, assignor)
+ .withTime(new MockTime())
+ .withMetadataImage(new
MetadataImageBuilder().buildCoordinatorMetadataImage())
+ .withGroupSpec(groupSpec)
+ .build()
+ .targetAssignment()
+ .get("member");
+ }
Review Comment:
It is inconsistent that the new tests use this helper method and the others
do not.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilderTest.java:
##########
@@ -170,20 +177,91 @@ public void testAssignment() {
.assign(groupSpec, new
SubscribedTopicDescriberImpl(metadataImage));
Map<String, Assignment> expectedAssignment = new HashMap<>();
- expectedAssignment.put("member-1", new Assignment(mkAssignment(
+ expectedAssignment.put("member-1", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
- expectedAssignment.put("member-2", new Assignment(mkAssignment(
+ expectedAssignment.put("member-2", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 3, 4, 5)
)));
- expectedAssignment.put("member-3", new Assignment(mkAssignment(
+ expectedAssignment.put("member-3", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 6),
mkTopicAssignment(barTopicId, 6)
)));
assertEquals(expectedAssignment, result.targetAssignment());
assertEquals(new TargetAssignmentMetadata(20, 12345L),
result.targetAssignmentMetadata());
}
+
+ @Test
+ public void testUnchangedAssignmentReusesPartitionSets() {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(previousAssignment.partitions())
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(previousAssignment.partitions().get(topicId),
assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
+ }
Review Comment:
Isn't this fully implied by the next test?
##########
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:
Do we really need to test this? It's incredibly unlikely we'd ever reach
into a record and modify its collections in place. They're also `List`s instead
of `Set`s.
##########
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);
Review Comment:
nit: This assert seems trivially true? We would have to really mess up the
implementation of `clear()` and `remove()` above somehow to break it.
```suggestion
```
##########
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));
Review Comment:
These assertions don't test a property of `of` method but rather the input
maps and sets.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilderTest.java:
##########
@@ -170,20 +177,91 @@ public void testAssignment() {
.assign(groupSpec, new
SubscribedTopicDescriberImpl(metadataImage));
Map<String, Assignment> expectedAssignment = new HashMap<>();
- expectedAssignment.put("member-1", new Assignment(mkAssignment(
+ expectedAssignment.put("member-1", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
- expectedAssignment.put("member-2", new Assignment(mkAssignment(
+ expectedAssignment.put("member-2", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 3, 4, 5)
)));
- expectedAssignment.put("member-3", new Assignment(mkAssignment(
+ expectedAssignment.put("member-3", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 6),
mkTopicAssignment(barTopicId, 6)
)));
assertEquals(expectedAssignment, result.targetAssignment());
assertEquals(new TargetAssignmentMetadata(20, 12345L),
result.targetAssignmentMetadata());
}
+
+ @Test
+ public void testUnchangedAssignmentReusesPartitionSets() {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(previousAssignment.partitions())
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(previousAssignment.partitions().get(topicId),
assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAssignmentReusesAssignorResultPartitionSets(boolean
unmodifiableMap) {
Review Comment:
Do we really need the `unmodifiableMap` case?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TargetAssignmentBuilderTest.java:
##########
@@ -170,20 +177,91 @@ public void testAssignment() {
.assign(groupSpec, new
SubscribedTopicDescriberImpl(metadataImage));
Map<String, Assignment> expectedAssignment = new HashMap<>();
- expectedAssignment.put("member-1", new Assignment(mkAssignment(
+ expectedAssignment.put("member-1", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 1, 2),
mkTopicAssignment(barTopicId, 1, 2)
)));
- expectedAssignment.put("member-2", new Assignment(mkAssignment(
+ expectedAssignment.put("member-2", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 3, 4, 5),
mkTopicAssignment(barTopicId, 3, 4, 5)
)));
- expectedAssignment.put("member-3", new Assignment(mkAssignment(
+ expectedAssignment.put("member-3", Assignment.of(mkAssignment(
mkTopicAssignment(fooTopicId, 6),
mkTopicAssignment(barTopicId, 6)
)));
assertEquals(expectedAssignment, result.targetAssignment());
assertEquals(new TargetAssignmentMetadata(20, 12345L),
result.targetAssignmentMetadata());
}
+
+ @Test
+ public void testUnchangedAssignmentReusesPartitionSets() {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(previousAssignment.partitions())
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(previousAssignment.partitions().get(topicId),
assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().get(topicId).remove(0));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testAssignmentReusesAssignorResultPartitionSets(boolean
unmodifiableMap) {
+ Uuid topicId = Uuid.randomUuid();
+ Assignment previousAssignment =
Assignment.of(mkAssignment(mkTopicAssignment(topicId, 0, 1)));
+ Set<Integer> partitionIds = new HashSet<>(Set.of(0, 1));
+ Map<Uuid, Set<Integer>> partitions = new HashMap<>();
+ partitions.put(topicId, partitionIds);
+
+ Map<Uuid, Set<Integer>> returnedPartitions = unmodifiableMap
+ ? Collections.unmodifiableMap(partitions)
+ : partitions;
+ Assignment assignment = buildTargetAssignment(previousAssignment,
Map.of(
+ "member", new MemberAssignmentImpl(returnedPartitions)
+ ));
+
+ assertEquals(previousAssignment, assignment);
+ assertSame(partitionIds, assignment.partitions().get(topicId));
+ assertThrows(UnsupportedOperationException.class, () ->
assignment.partitions().clear());
+ }
+
+ @Test
+ public void testMissingMemberAssignmentReturnsEmpty() {
+ Assignment previousAssignment = Assignment.of(mkAssignment(
+ mkTopicAssignment(Uuid.randomUuid(), 0, 1)
+ ));
+
+ assertSame(Assignment.EMPTY, buildTargetAssignment(previousAssignment,
Map.of()));
+ }
Review Comment:
nit: Could we move this up just after `testAssignment` and call it
`testAssignmentWithMissingMemberAssignment` or something like that?
--
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]
