lianetm commented on code in PR #17463:
URL: https://github.com/apache/kafka/pull/17463#discussion_r1819718899
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3575,6 +3600,65 @@ public void replay(
}
}
+ /**
+ * Replays ConsumerGroupRegexKey/Value to update the hard state of
+ * the consumer group related to a regular expression. It updates
+ * the resolution of the regular expression with the value, or deletes
+ * the regular expression if the record value is null.
+ *
+ * @param key A ConsumerGroupRegexKey record.
+ * @param value A ConsumerGroupRegexValue record.
+ */
+ public void replay(
+ ConsumerGroupRegexKey key,
+ ConsumerGroupRegexValue value
+ ) {
+
+ String groupId = key.groupId();
+ String regex = key.regex();
+
+ try {
+ getOrMaybeCreatePersistedConsumerGroup(groupId, value != null);
+ } catch (GroupIdNotFoundException ex) {
+ // If the group does not exist anymore do not load regex.
+ // TODO: validate if this could be a case: group does exist
anymore but regex record being replayed.
+ return;
+ }
+
+ if (value != null) {
+ updateGroupRegex(
+ new
ConsumerGroupRegex.RegexKey.Builder().updateWith(key).build(),
+ new Resolution.Builder().updateWith(value).build()
+ );
Review Comment:
yes, makes total sense, done and it does look much cleaner. The regexManger
only cares about the async operation (resolves, writes records), info stored in
the group (replayed from records). Makes sense?
--
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]
