dajac commented on code in PR #23742:
URL: https://github.com/apache/kafka/pull/23742#discussion_r4346948394
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/ModernGroup.java:
##########
@@ -279,7 +279,7 @@ public Map<Uuid, Map<Integer, String>>
invertedTargetAssignment() {
public void updateTargetAssignment(String memberId, Assignment
newTargetAssignment) {
updateInvertedTargetAssignment(
memberId,
- targetAssignment.getOrDefault(memberId, new Assignment(Map.of())),
+ targetAssignment.getOrDefault(memberId,
Assignment.copyOf(Map.of())),
Review Comment:
I think that this code is gone in trunk but, if not, we should use
`Assignment.EMPTY`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/MemberSubscriptionAndAssignmentImpl.java:
##########
@@ -101,4 +101,5 @@ public String toString() {
", memberAssignment=" + memberAssignment +
')';
}
+
Review Comment:
nit: Let's revert this one.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/UniformHomogeneousAssignmentBuilderTest.java:
##########
@@ -461,8 +445,14 @@ public void
testReassignmentWhenOneMemberAddedAfterInitialAssignmentWithTwoMembe
subscribedTopicMetadata
);
- assertAssignment(expectedAssignment, computedAssignment);
- checkValidityAndBalance(members, computedAssignment);
+ // Each existing member gives up exactly one partition to the new
member.
+ // Which partitions move depends on iteration order.
+ assertReassignment(
+ members,
+ computedAssignment,
+ Map.of(topic1Uuid, Set.of(0, 1, 2), topic2Uuid, Set.of(0, 1, 2)),
+ Map.of(memberA, 2, memberB, 2, memberC, 0)
+ );
Review Comment:
Why do we need this change? It seems unrelated.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/Assignment.java:
##########
@@ -98,10 +133,10 @@ public static Assignment fromRecord(
public static Assignment fromRecord(
ShareGroupTargetAssignmentMemberValue record
) {
- return new Assignment(
+ return Assignment.of(
record.topicPartitions().stream().collect(Collectors.toMap(
ShareGroupTargetAssignmentMemberValue.TopicPartition::topicId,
- topicPartitions -> new
HashSet<>(topicPartitions.partitions())))
+ topicPartitions -> Set.copyOf(topicPartitions.partitions())))
Review Comment:
It seems that this is slightly less efficient than `unmodifiableSet(new
HashSet())`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -1271,7 +1271,7 @@ public static ConsumerGroup fromClassicGroup(
))
)
.build();
- consumerGroup.updateTargetAssignment(newMember.memberId(), new
Assignment(assignedPartitions));
+ consumerGroup.updateTargetAssignment(newMember.memberId(),
Assignment.copyOf(assignedPartitions));
Review Comment:
A defensive copy does not seem necessary here because we create the
assignment here. Should we rather ensure that `assignedPartitions` is immutable
when created and then transfer the ownership?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -274,6 +276,75 @@
public class GroupMetadataManagerTest {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testMutableAssignorOutputBecomesImmutableInputAfterReplay(boolean shareGroup) {
+ String groupId = "group";
+ String memberId = Uuid.randomUuid().toString();
+ Uuid topicId = Uuid.randomUuid();
+ Set<Integer> partitionIds = new HashSet<>(Set.of(0, 1));
+ Map<Uuid, Set<Integer>> partitions = new HashMap<>();
+ partitions.put(topicId, partitionIds);
+
+ MockPartitionAssignor assignor = spy(new
MockPartitionAssignor("range"));
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
List.of(assignor))
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG,
0)
+
.withConfig(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, 0)
+ .withShareGroupAssignor(assignor)
+ .withMetadataImage(new MetadataImageBuilder()
+ .addTopic(topicId, "foo", 2)
+ .buildCoordinatorMetadataImage())
+ .build();
+
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ memberId, new MemberAssignmentImpl(partitions)
+ )));
+
+ // The first join creates and replays records from the mutable output.
+ // A second join triggers assignment using the replayed target
assignment.
+ for (String joiningMemberId : List.of(memberId,
Uuid.randomUuid().toString())) {
+ List<CoordinatorRecord> records;
+ CoordinatorRecord expectedAssignmentRecord;
+ if (shareGroup) {
+ records = context.shareGroupHeartbeat(new
ShareGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(joiningMemberId)
+ .setMemberEpoch(0)
+ .setSubscribedTopicNames(List.of("foo"))).records();
+ expectedAssignmentRecord =
GroupCoordinatorRecordHelpers.newShareGroupTargetAssignmentRecord(
+ groupId, memberId, partitions);
+ } else {
+ records = context.consumerGroupHeartbeat(new
ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setMemberId(joiningMemberId)
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo"))
+ .setTopicPartitions(List.of())).records();
+ expectedAssignmentRecord =
GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(
+ groupId, memberId, partitions);
+ }
+ if (joiningMemberId.equals(memberId)) {
+ assertTrue(records.contains(expectedAssignmentRecord));
+ }
+ }
+
+ ArgumentCaptor<GroupSpec> groupSpecCaptor =
ArgumentCaptor.forClass(GroupSpec.class);
Review Comment:
Do we really need this? Could we just look up the target assignment after
the replay?
--
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]
