You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dongnuo123 (via GitHub)" <gi...@apache.org> on 2025/06/04 19:48:32 UTC
[PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
dongnuo123 opened a new pull request, #19904:
URL: https://github.com/apache/kafka/pull/19904
Consumers can subscribe to an RE2J SubscriptionPattern that will be resolved and maintained on the server-side (KIP-848). Currently, those regexes are refreshed on the coordinator when a consumer subscribes to a new regex, or if there is a new topic metadata image (to ensure regex resolution stays up-to-date with existing topics)
But with [KAFKA-18813](https://issues.apache.org/jira/browse/KAFKA-18813), the topics matching a regex are filtered based on ACLs. This generates a new situation, as regexes resolution do not stay up-to-date as topics become visible (ACLs added/delete).
This patch introduces time-based refresh for the subscribed regex by
- Adding `group.consumer.regex.refresh.interval.ms` config that controls the refresh interval.
- Periodically scheduling regex refresh task to the coordinator executor. The task is cancelled when the consumer group is no longer subscribed to any regex.
- Creating a cache in `GroupMetadataManager` that keeps track of the latest context of the member used to filter unauthorized topics.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2140187447
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -3076,6 +3076,29 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
sendAndReceiveRegexHeartbeat(response, listenerName, None)
}
+ @Test
+ def testConsumerGroupHeartbeatWithRegexWithTopicDescribeAclAddedAndRemoved(): Unit = {
+ createTopicWithBrokerPrincipal(topic)
+ val allowAllOpsAcl = new AccessControlEntry(clientPrincipalString, WILDCARD_HOST, ALL, ALLOW)
+ addAndVerifyAcls(Set(allowAllOpsAcl), groupResource)
+
+ val memberId = Uuid.randomUuid.toString;
+ var response = sendAndReceiveFirstRegexHeartbeat(memberId, listenerName)
+ TestUtils.tryUntilNoAssertionError() {
+ response = sendAndReceiveRegexHeartbeat(response, listenerName, Some(0), true)
+ }
+
+ addAndVerifyAcls(topicDescribeAcl(topicResource), topicResource)
+ TestUtils.tryUntilNoAssertionError(waitTime = 75000) {
Review Comment:
Yeah I agree it's a bit annoying. The refresh interval config lower limit is set to be 60s to avoid too frequent refresh, but the integration test seems to use system time so there's no way to advance time manually. Or we decrease the lower limit back to 10s
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2129826048
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -20542,6 +20553,112 @@ barTopicName, computeTopicHash(barTopicName, metadataImage)
);
}
+ @Test
Review Comment:
Added an IT to authorizer integration tests
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2140324934
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -3076,6 +3076,29 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
sendAndReceiveRegexHeartbeat(response, listenerName, None)
}
+ @Test
+ def testConsumerGroupHeartbeatWithRegexWithTopicDescribeAclAddedAndRemoved(): Unit = {
+ createTopicWithBrokerPrincipal(topic)
+ val allowAllOpsAcl = new AccessControlEntry(clientPrincipalString, WILDCARD_HOST, ALL, ALLOW)
+ addAndVerifyAcls(Set(allowAllOpsAcl), groupResource)
+
+ val memberId = Uuid.randomUuid.toString;
+ var response = sendAndReceiveFirstRegexHeartbeat(memberId, listenerName)
+ TestUtils.tryUntilNoAssertionError() {
+ response = sendAndReceiveRegexHeartbeat(response, listenerName, Some(0), true)
+ }
+
+ addAndVerifyAcls(topicDescribeAcl(topicResource), topicResource)
+ TestUtils.tryUntilNoAssertionError(waitTime = 75000) {
Review Comment:
Since it is an internal config, we can reduce it to 10s.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2135951958
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -204,6 +205,11 @@ public class GroupCoordinatorConfig {
ConsumerGroupMigrationPolicy.DOWNGRADE + ": only downgrade from consumer group to classic group is enabled, " +
ConsumerGroupMigrationPolicy.DISABLED + ": neither upgrade nor downgrade is enabled.";
+ public static final String CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_CONFIG = "group.consumer.regex.batch.refresh.max.interval.ms";
+ public static final String CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DOC = "The interval at which the group coordinator will refresh " +
+ "the topics matching the group subscribed regexes. This is only applicable to consumer groups using the consumer group protocol. ";
+ public static final int CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DEFAULT = 10 * 60 * 1000; // 10 minutes
+
Review Comment:
Yes it makes sense. I wanted to differentiate it with `REGEX_BATCH_REFRESH_MIN_INTERVAL_MS` so regex.refresh.interval.ms also works
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2127826516
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -8537,6 +8659,32 @@ static String consumerGroupSyncKey(String groupId, String memberId) {
return "sync-" + groupId + "-" + memberId;
}
+ /**
+ * Generate a regex refresh key for the timer.
+ *
+ * Package private for testing.
+ *
+ * @param groupId The group id.
+ *
Review Comment:
nit: extra blank
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -204,6 +204,11 @@ public class GroupCoordinatorConfig {
ConsumerGroupMigrationPolicy.DOWNGRADE + ": only downgrade from consumer group to classic group is enabled, " +
ConsumerGroupMigrationPolicy.DISABLED + ": neither upgrade nor downgrade is enabled.";
+ public static final String CONSUMER_GROUP_REGEX_REFRESH_INTERVAL_CONFIG = "group.consumer.regex.refresh.interval.ms";
+ public static final String CONSUMER_GROUP_REGEX_REFRESH_INTERVAL_DOC = "The interval at which the group coordinator will atomatically" +
Review Comment:
What is the 'atomatically' mean?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -8537,6 +8659,32 @@ static String consumerGroupSyncKey(String groupId, String memberId) {
return "sync-" + groupId + "-" + memberId;
}
+ /**
+ * Generate a regex refresh key for the timer.
+ *
+ * Package private for testing.
+ *
+ * @param groupId The group id.
+ *
+ * @return the regex timeout refresh key.
+ */
+ static String regexTimeoutRefreshKeyForTimer(String groupId) {
+ return "regex-timeout-refresh-for-timer-" + groupId;
+ }
+
+ /**
+ * Generate a regex refresh key for the executor.
+ *
+ * Package private for testing.
+ *
+ * @param groupId The group id.
+ *
Review Comment:
nit: extra blank
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -20542,6 +20553,112 @@ barTopicName, computeTopicHash(barTopicName, metadataImage)
);
}
+ @Test
Review Comment:
Should we add IT for this?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2138803472
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -3093,7 +3116,7 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
// member permissions while computing assignments.
var member2Response = sendAndReceiveFirstRegexHeartbeat("memberWithLimitedAccess", listenerName)
member1Response = sendAndReceiveRegexHeartbeat(member1Response, interBrokerListenerName, Some(1))
- member1Response = sendAndReceiveRegexHeartbeat(member1Response, interBrokerListenerName, None, fullRequest = true)
+ member1Response = sendAndReceiveRegexHeartbeat(member1Response, interBrokerListenerName, Some(1), fullRequest = true)
Review Comment:
When full request is used, the assignment in the response should be set
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #19904:
URL: https://github.com/apache/kafka/pull/19904#issuecomment-2943063987
@dongnuo123 Thanks for the patch. I have a general comment on the approach. Have you considered adding a new time based condition to [maybeUpdateRegularExpressions](https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java#L3072)? It seems a bit simpler but I may have missed something.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2131654014
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -204,6 +205,11 @@ public class GroupCoordinatorConfig {
ConsumerGroupMigrationPolicy.DOWNGRADE + ": only downgrade from consumer group to classic group is enabled, " +
ConsumerGroupMigrationPolicy.DISABLED + ": neither upgrade nor downgrade is enabled.";
+ public static final String CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_CONFIG = "group.consumer.regex.batch.refresh.max.interval.ms";
+ public static final String CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DOC = "The interval at which the group coordinator will refresh " +
+ "the topics matching the group subscribed regexes. This is only applicable to consumer groups using the consumer group protocol. ";
+ public static final int CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DEFAULT = 10 * 60 * 1000; // 10 minutes
+
Review Comment:
We cannot add a new config without a KIP now. We will soon do a KIP to improve regular expressions, I suggest to include it in that one. In the mean time, we can keep it internal, mainly for testing purposes. What do you think? If you agree, let's add a comment to state that it is internal for testing.
Regarding the name, I would go with something simpler: `group.consumer.regex.refresh.interval.ms`. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -308,6 +314,7 @@ public class GroupCoordinatorConfig {
.define(CONSUMER_GROUP_MAX_SIZE_CONFIG, INT, CONSUMER_GROUP_MAX_SIZE_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_MAX_SIZE_DOC)
.define(CONSUMER_GROUP_ASSIGNORS_CONFIG, LIST, CONSUMER_GROUP_ASSIGNORS_DEFAULT, null, MEDIUM, CONSUMER_GROUP_ASSIGNORS_DOC)
.define(CONSUMER_GROUP_MIGRATION_POLICY_CONFIG, STRING, CONSUMER_GROUP_MIGRATION_POLICY_DEFAULT, ConfigDef.CaseInsensitiveValidString.in(Utils.enumOptions(ConsumerGroupMigrationPolicy.class)), MEDIUM, CONSUMER_GROUP_MIGRATION_POLICY_DOC)
+ .define(CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_CONFIG, INT, CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DEFAULT, atLeast(REGEX_BATCH_REFRESH_MIN_INTERVAL_MS + 1), MEDIUM, CONSUMER_GROUP_REGEX_BATCH_REFRESH_MAX_INTERVAL_MS_DOC)
Review Comment:
nit: Should we require at least 1 minutes to avoid having too many refreshes?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3134,11 +3137,11 @@ private boolean maybeUpdateRegularExpressions(
// 2. The last refresh is older than 10s. If the group does not have any regular
// expressions but the current member just brought a new one, we should continue.
long lastRefreshTimeMs = group.lastResolvedRegularExpressionRefreshTimeMs();
- if (time.milliseconds() <= lastRefreshTimeMs + REGEX_BATCH_REFRESH_INTERVAL_MS) {
+ if (time.milliseconds() <= lastRefreshTimeMs + REGEX_BATCH_REFRESH_MIN_INTERVAL_MS) {
Review Comment:
nit: Let's extract time into a var (e.g. currentTimeMs) as we reuse it now.
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -3076,6 +3076,23 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
sendAndReceiveRegexHeartbeat(response, listenerName, None)
}
+ @Test
+ def testConsumerGroupHeartbeatWithRegexWithGrantedTopicDescribeAcl(): Unit = {
Review Comment:
Should we add a test in which we start with an ACL, subscribe to the topic, get it, remove the ACL, and finally verify that it is removed after the refresh timeout?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2129830186
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -204,6 +204,11 @@ public class GroupCoordinatorConfig {
ConsumerGroupMigrationPolicy.DOWNGRADE + ": only downgrade from consumer group to classic group is enabled, " +
ConsumerGroupMigrationPolicy.DISABLED + ": neither upgrade nor downgrade is enabled.";
+ public static final String CONSUMER_GROUP_REGEX_REFRESH_INTERVAL_CONFIG = "group.consumer.regex.refresh.interval.ms";
+ public static final String CONSUMER_GROUP_REGEX_REFRESH_INTERVAL_DOC = "The interval at which the group coordinator will atomatically" +
Review Comment:
Thanks for the catch! Yes it's a typo. I wanted to say "automatically"
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #19904:
URL: https://github.com/apache/kafka/pull/19904#discussion_r2139938164
##########
core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala:
##########
@@ -3076,6 +3076,29 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest {
sendAndReceiveRegexHeartbeat(response, listenerName, None)
}
+ @Test
+ def testConsumerGroupHeartbeatWithRegexWithTopicDescribeAclAddedAndRemoved(): Unit = {
+ createTopicWithBrokerPrincipal(topic)
+ val allowAllOpsAcl = new AccessControlEntry(clientPrincipalString, WILDCARD_HOST, ALL, ALLOW)
+ addAndVerifyAcls(Set(allowAllOpsAcl), groupResource)
+
+ val memberId = Uuid.randomUuid.toString;
+ var response = sendAndReceiveFirstRegexHeartbeat(memberId, listenerName)
+ TestUtils.tryUntilNoAssertionError() {
+ response = sendAndReceiveRegexHeartbeat(response, listenerName, Some(0), true)
+ }
+
+ addAndVerifyAcls(topicDescribeAcl(topicResource), topicResource)
+ TestUtils.tryUntilNoAssertionError(waitTime = 75000) {
Review Comment:
Could we reduce the time? 60s is pretty long for a test. Could we use 10s and wait 15s?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18961: Time-based refresh for server-side RE2J regex [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #19904:
URL: https://github.com/apache/kafka/pull/19904
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org