dajac commented on code in PR #17694:
URL: https://github.com/apache/kafka/pull/17694#discussion_r1831260391
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -342,6 +348,46 @@ private void removeStaticMember(ConsumerGroupMember
oldMember) {
}
}
+ /**
+ * Update the resolved regular expression.
+ *
+ * @param regex The regular expression.
+ * @param newResolvedRegularExpression The regular expression's metadata.
+ */
+ public void updateResolvedRegularExpression(
+ String regex,
+ ResolvedRegularExpression newResolvedRegularExpression
+ ) {
+ removeResolvedRegularExpression(regex);
+ if (newResolvedRegularExpression != null) {
+ resolvedRegularExpressions.put(regex,
newResolvedRegularExpression);
+ newResolvedRegularExpression.topics.forEach(topicName ->
subscribedTopicNames.compute(topicName, Utils::incValue));
Review Comment:
We have two data structures:
1) `subscribedRegularExpressions` which is updated when the member
subscribed regex is updated. Hence here we account for the change immediately.
2) `resolvedRegularExpressions` which is updated when the regular is
resolved. At the same time, we update `subscribedTopicNames` to take into
account the subscribed topics via the regex. We can only do this when the regex
is resolved though.
--
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]
