squah-confluent commented on code in PR #20055:
URL: https://github.com/apache/kafka/pull/20055#discussion_r2443289637
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/share/ShareGroupAssignmentBuilder.java:
##########
@@ -83,11 +125,38 @@ public ShareGroupMember build() {
// when the member is updated.
return new ShareGroupMember.Builder(member)
.setState(MemberState.STABLE)
- .setAssignedPartitions(targetAssignment.partitions())
+ // If we have client-side assignors, the latest target
assignment may not
+ // be consistent with the latest subscribed topics, so we must
always
+ // filter the assigned partitions to ensure they are
consistent with the
+ // subscribed topics.
+
.setAssignedPartitions(filterAssignedPartitions(targetAssignment.partitions(),
member.subscribedTopicNames()))
.updateMemberEpoch(targetAssignmentEpoch)
.build();
+ } else if (hasSubscriptionChanged) {
+ return new ShareGroupMember.Builder(member)
+
.setAssignedPartitions(filterAssignedPartitions(targetAssignment.partitions(),
member.subscribedTopicNames()))
+ .build();
+ } else {
+ return member;
}
+ }
- return member;
+ private Map<Uuid, Set<Integer>> filterAssignedPartitions(
+ Map<Uuid, Set<Integer>> partitions,
+ Set<String> subscribedTopicNames
+ ) {
+ TopicIds subscribedTopicIds = new
TopicIds(member.subscribedTopicNames(), metadataImage);
Review Comment:
Hi @majialoong, this looks like an oversight. Please feel free to submit a
PR to address the issue!
--
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]
