squah-confluent commented on code in PR #23972:
URL: https://github.com/apache/kafka/pull/23972#discussion_r4402258211
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/assignor/SimpleAssignorTest.java:
##########
@@ -760,6 +760,57 @@ public void testAssignWithUnsubscribedTopicHeterogeneous()
{
assertEveryPartitionGetsAssignment(6, computedAssignment);
}
+ @Test
+ public void testAssignWithTopicUnsubscribedByAllMembersHeterogeneous() {
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(TOPIC_1_UUID, TOPIC_1_NAME, 4)
+ .addTopic(TOPIC_2_UUID, TOPIC_2_NAME, 2)
+ .addTopic(TOPIC_3_UUID, TOPIC_3_NAME, 2)
+ .buildCoordinatorMetadataImage();
+ SubscribedTopicDescriberImpl subscribedTopicMetadata = new
SubscribedTopicDescriberImpl(
+ metadataImage
+ );
+
+ // No member subscribes to topic 2 anymore, but both members still
have a partition of it
+ Map<String, MemberSubscriptionAndAssignmentImpl> members = new
HashMap<>();
+ members.put(MEMBER_A, new MemberSubscriptionAndAssignmentImpl(
+ Optional.empty(),
+ Optional.empty(),
+ Set.of(TOPIC_1_UUID),
+ new Assignment(mkAssignment(
+ mkTopicAssignment(TOPIC_1_UUID, 0),
+ mkTopicAssignment(TOPIC_2_UUID, 0)))
+ ));
+ members.put(MEMBER_B, new MemberSubscriptionAndAssignmentImpl(
+ Optional.empty(),
+ Optional.empty(),
+ Set.of(TOPIC_1_UUID, TOPIC_3_UUID),
+ new Assignment(mkAssignment(
+ mkTopicAssignment(TOPIC_1_UUID, 1),
+ mkTopicAssignment(TOPIC_2_UUID, 1)))
+ ));
+
+ GroupSpec groupSpec = new GroupSpecImpl(metadataImage, members,
HETEROGENEOUS);
Review Comment:
Thanks for fixing this bug. Just fyi, I'm doing some refactoring of the
`GroupSpec`s and if #23896 merges first, we should be using the
`ShareGroupSpecImpl` class here
--
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]
