jeffkbkim commented on code in PR #17694:
URL: https://github.com/apache/kafka/pull/17694#discussion_r1831080180
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -342,6 +348,47 @@ private void removeStaticMember(ConsumerGroupMember
oldMember) {
}
}
+ /**
+ * Update the regular expression.
+ *
+ * @param regex The regular expression.
+ * @param newResolvedRegularExpression The regular expression's metadata.
+ */
+ public void updateRegularExpression(
+ String regex,
+ ResolvedRegularExpression newResolvedRegularExpression
+ ) {
+ removeRegularExpression(regex);
+ if (newResolvedRegularExpression != null) {
+ resolvedRegularExpressions.put(regex,
newResolvedRegularExpression);
+ newResolvedRegularExpression.topics.forEach(topicName ->
subscribedTopicNames.compute(topicName, Utils::incValue));
+ }
+ }
+
+ /**
+ * Remove the regular expression.
+ *
+ * @param regex The regular expression.
+ */
+ public void removeRegularExpression(
+ String regex
+ ) {
+ ResolvedRegularExpression oldResolvedRegularExpression =
resolvedRegularExpressions.remove(regex);
+ if (oldResolvedRegularExpression != null) {
+ oldResolvedRegularExpression.topics.forEach(topicName ->
subscribedTopicNames.compute(topicName, Utils::decValue));
+ }
+ }
+
+ /**
+ * @return An optional containing the resolved regular expression
corresponding to the provided regex
+ * or an empty optional.
+ */
+ public Optional<ResolvedRegularExpression> regularExpression(
Review Comment:
nit: resolvedRegularExpression and javadocs missing param
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java:
##########
@@ -342,6 +348,47 @@ private void removeStaticMember(ConsumerGroupMember
oldMember) {
}
}
+ /**
+ * Update the regular expression.
+ *
+ * @param regex The regular expression.
+ * @param newResolvedRegularExpression The regular expression's metadata.
+ */
+ public void updateRegularExpression(
Review Comment:
nit: how's updateResolvedRegularExpression and
removeResolvedRegularExpression? and
"Update the resolved regular expression."
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -15086,6 +15087,63 @@ public void testShareGroupDescribeOnConsumerGroup() {
assertEquals(expected, actual);
}
+ @Test
+ public void testReplayConsumerGroupRegularExpression() {
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+
+ ResolvedRegularExpression regex = new ResolvedRegularExpression(
Review Comment:
nit: resolvedRegularExpression there are 2 more instances in other tests
##########
group-coordinator/src/main/resources/common/message/ConsumerGroupRegularExpressionValue.json:
##########
@@ -0,0 +1,27 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// KIP-848 is in development. This schema is subject to
non-backwards-compatible changes.
+{
+ "type": "data",
+ "name": "ConsumerGroupRegularExpressionValue",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "Topics", "versions": "0+", "type": "[]string" },
Review Comment:
should we include topic ids here? i am wondering if we need to have ids when
passing this information to the target assignment builder.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ResolvedRegularExpression.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern.consumer;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * The metadata associated with a regular expression in a Consumer Group.
+ */
+public class ResolvedRegularExpression {
+ /**
+ * The set of resolved topics.
+ */
+ public final Set<String> topics;
+
+ /**
+ * The version of the metadata image used to resolve the topics.
+ */
+ public final long version;
+
+ /**
+ * The timestamp at the time of the resolution.
+ */
+ public final long timestamp;
Review Comment:
will timestamp be only used as a reference when looking into the regex
record or will it actually be used in some way in the code?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ResolvedRegularExpression.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern.consumer;
+
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * The metadata associated with a regular expression in a Consumer Group.
+ */
+public class ResolvedRegularExpression {
+ /**
+ * The set of resolved topics.
+ */
+ public final Set<String> topics;
+
+ /**
+ * The version of the metadata image used to resolve the topics.
Review Comment:
could you refer me to the class that generates this metadata image version?
--
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]
