lianetm commented on code in PR #20055:
URL: https://github.com/apache/kafka/pull/20055#discussion_r2186107711
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -317,4 +451,29 @@ private ConsumerGroupMember computeNextAssignment(
.build();
}
}
+
+ /**
+ * Gets the set of topic IDs that the member is subscribed to.
+ *
+ * @return The set of topic IDs that the member is subscribed to.
+ */
+ private Set<Uuid> subscribedTopicIds() {
+ Set<String> subscriptions = member.subscribedTopicNames();
+ String subscribedTopicRegex = member.subscribedTopicRegex();
+ if (subscribedTopicRegex != null && !subscribedTopicRegex.isEmpty()) {
+ ResolvedRegularExpression resolvedRegularExpression =
resolvedRegularExpressions.get(subscribedTopicRegex);
+ if (resolvedRegularExpression != null) {
+ if (subscriptions.isEmpty()) {
+ subscriptions = resolvedRegularExpression.topics;
+ } else if (!resolvedRegularExpression.topics.isEmpty()) {
+ subscriptions = new UnionSet<>(subscriptions,
resolvedRegularExpression.topics);
+ }
+ } else {
+ // Treat an unresolved regex as matching no topics, to be
conservative.
+ }
+ }
+
+ TopicIds subscribedTopicIds = new TopicIds(subscriptions,
metadataImage.topics());
+ return subscribedTopicIds;
Review Comment:
```suggestion
return new TopicIds(subscriptions, metadataImage.topics());
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2243,21 +2243,30 @@ private
CoordinatorResult<ConsumerGroupHeartbeatResponseData, CoordinatorRecord>
// epoch 0 and that it is fully initialized.
boolean bumpGroupEpoch = group.groupEpoch() == 0;
- bumpGroupEpoch |= hasMemberSubscriptionChanged(
+ boolean subscribedTopicNamesChanged = hasMemberSubscriptionChanged(
groupId,
member,
updatedMember,
records
);
-
- bumpGroupEpoch |= maybeUpdateRegularExpressions(
+ UpdateRegularExpressionsResult updateRegularExpressionsResult =
maybeUpdateRegularExpressions(
context,
group,
member,
updatedMember,
records
);
+ // The subscription has changed when either the subscribed topic names
or subscribed topic
+ // regex has changed.
+ boolean hasSubscriptionChanged = subscribedTopicNamesChanged ||
updateRegularExpressionsResult.subscribedTopicRegexChanged;
Review Comment:
nit: this seems only needed further down on `maybeReconcile` , should we
move it closer?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/CurrentAssignmentBuilder.java:
##########
@@ -215,6 +284,64 @@ private boolean ownsRevokedPartitions(
return false;
}
+ /**
+ * Updates the current assignment, removing any partitions that are not
part of the subscribed topics.
+ * This method is a lot faster than running the full reconciliation logic
in computeNextAssignment.
+ *
+ * @param memberAssignedPartitions The assigned partitions of the member
to use.
+ * @return A new ConsumerGroupMember.
+ */
+ private ConsumerGroupMember updateCurrentAssignment(
+ Map<Uuid, Set<Integer>> memberAssignedPartitions
+ ) {
+ Set<Uuid> subscribedTopicIds = subscribedTopicIds();
+
+ // Reuse the original map if no topics need to be removed.
+ Map<Uuid, Set<Integer>> newAssignedPartitions =
memberAssignedPartitions;
+ Map<Uuid, Set<Integer>> newPartitionsPendingRevocation = new
HashMap<>(member.partitionsPendingRevocation());
+ for (Map.Entry<Uuid, Set<Integer>> entry :
memberAssignedPartitions.entrySet()) {
Review Comment:
I guess it would be common for regex case to end up with empty
`subscribedTopicIds` here (regex not resolved), but still we would iterate the
whole set of assigned partitions.
Would it make sense to treat that case separately to avoid iterating to
remove 1 by 1 if subscribedTopicIds.isEmpty?
--
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]
