KAFKA-15561: Client support for new SubscriptionPattern based subscription #15188

Phuc-Hong-Tran · 2024-01-13T11:40:34Z

Change:

Implement methods in AsyncKafkaConsumer that accept SubscriptionPattern to subscribe to topic(s).
Pass on the subscriptionPattern to SubscriptionState to use once server support RE2/J.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

Phuc-Hong-Tran · 2024-01-13T11:43:10Z

@lianetm, PTAL, thanks in advance.

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

lianetm · 2024-01-16T20:10:01Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManagerTest.java

@@ -86,7 +86,6 @@
 import static org.mockito.Mockito.when;

 public class OffsetsRequestManagerTest {
-


Nit: I find it's better to avoid changes in unrelated files, even if minor

Could you please revert this change?

lianetm

Hey @Phuc-Hong-Tran , thanks for the changes! I took a first pass, some high level comments:

This feature is not yet implemented on the broker, so there are a few areas that may need alignment. Would maybe make sense to wait for some progress in the broker and then give this another push.
This PR attempts to ensure that the new regex is included in the subscription state and that's definitely needed. But then we're missing the other half of the story: we should make sure that the regex is passed-on to the broker on the next heartbeat request (see HeartbeatRequestManager here)
We need to add some tests to cover the new changes. Some unit tests, and also integration tests, probably similar to the ones defined for the subscribe with Pattern (see PlainTextConsumerTest). Of course, at this point we would be blocked because the server does not support it yet.

Thanks!

Phuc-Hong-Tran · 2024-01-16T20:37:47Z

@lianetm, thanks for the comments, I will make sure to address those points in my next PR.

Regarding your point about passing the regex for HeartbeatRequestManager, I origninally included that in my code change, then I came across this PR #14956 and decided that we need to wait for the broker to implement new regex logic first.

lianetm · 2024-01-16T21:49:11Z

This is the task to closely follow https://issues.apache.org/jira/browse/KAFKA-14517, where the broker will support the new regex.

Phuc-Hong-Tran · 2024-01-16T22:39:26Z

Thanks @lianetm

cadonna

Thanks for the PR, @Phuc-Hong-Tran !

I did a first pass and left a couple of comments.

I could not find any unit tests. I also think there is something missing regarding getting the new pattern subscription into the heartbeat request.

I also have a more organizational question. We would like to get this ticket done as soon as possible. Do you have time to work on it?
If not, do not worry, I could take over and we would become co-authors. Let us please know what you think. The next release is coming up soon.

clients/src/main/java/org/apache/kafka/clients/consumer/Consumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

cadonna · 2024-01-31T10:58:56Z

This feature is not yet implemented on the broker, so there are a few areas that may need alignment. Would maybe make sense to wait for some progress in the broker and then give this another push.

This PR attempts to ensure that the new regex is included in the subscription state and that's definitely needed. But then we're missing the other half of the story: we should make sure that the regex is passed-on to the broker on the next heartbeat request (see HeartbeatRequestManager here)

I do not understand why we want to wait. We should make progress as much as possible even if the broker code is not there yet. In the worst case, we would get an error back from the broker until the broker-side is implemented. But that would be fine in my opinion.
@lianetm WDYT?

Phuc-Hong-Tran · 2024-01-31T11:00:14Z

@cadonna thanks for the comment, I can still finish one as the deadline required, there is no need to wait for the logic on broker to be finnished, though isn't it this one aiming for 3.8 release?

cadonna · 2024-01-31T11:06:59Z

@Phuc-Hong-Tran

though isn't it this one aiming for 3.8 release?

Yes, but we have time-based releases in Apache Kafka. That means that the deadline for the release will be somewhere in the beginning of April. We need to keep enough time before this data for testing.

Phuc-Hong-Tran · 2024-01-31T11:12:04Z

I understand, will get back to speed on this one

dajac · 2024-01-31T11:14:18Z

In my opinion, we should not merge the client side only after the server side is implemented. The reason is that we need to change the RPC (this is actually missing in this PR) and this should be driven by the server side work.

cadonna · 2024-01-31T11:15:46Z

@dajac OK, but we can implement and unit test everything up to the RPC, right?

Phuc-Hong-Tran · 2024-01-31T11:20:22Z

@dajac, when we're talking about the RPC, do we mean the field for the regex in ConsumerGroupHeartbeatRequest.json?

dajac · 2024-01-31T11:25:04Z

@cadonna Yes. We could for instance commit the RPC and then work independently on the client and the server. My only concern is that we usually discover issues while working on the server side. This is why I usually prefer to get the server code into a reasonable state first. In this case, the risk is low as we are talking about a single field.

Phuc-Hong-Tran · 2024-02-19T13:48:57Z

@cadonna @lianetm, since we're supporting for both subscribe method using java.util.regex.Pattern and SubscriptionPattern, I think we should throw a illegal heartbeat exeption when user try to use both method at the same time and inform the user to use once at a time, since the field SubscribedRegex is used for java.util.regex.Pattern as well as SubscriptionPattern. What do you guys think?

IMO, we must support the deprecated pattern subscriptions with java.util.regex.Pattern to ensure backwards compatibility, but we do not need to support mixed usage of java.util.regex.Pattern and Google regex patterns. I think this is a blind spot in the KIP. I propose to throw an IllegalStateException if subscribe(java.util.regex.Pattern) is called after subscribe(SubscriptionPattern) (and vice versa) without calling unsubscribe() in between. That is similar to the restrictions between pattern, topic, and partition subscriptions @lianetm linked above. I do not think it is worth to consider the edge case of mixed usage of the two pattern types. Does this make sense to you? \cc @dajac What do you as the original author of the KIP think? Should we update the KIP to make this clear?

@cadonna I would rather follow what we already do with subscribe today. The last one called takes precedence.

I have a question. The subscribe method that use Pattern override the subscription with topic(s) that match the Pattern. When user choose to use SubscriptionPattern, but already used Pattern beforehand, should we clear out the old subscription?

…est.json

cadonna · 2024-02-20T09:35:08Z

@cadonna @lianetm, since we're supporting for both subscribe method using java.util.regex.Pattern and SubscriptionPattern, I think we should throw a illegal heartbeat exeption when user try to use both method at the same time and inform the user to use once at a time, since the field SubscribedRegex is used for java.util.regex.Pattern as well as SubscriptionPattern. What do you guys think?

IMO, we must support the deprecated pattern subscriptions with java.util.regex.Pattern to ensure backwards compatibility, but we do not need to support mixed usage of java.util.regex.Pattern and Google regex patterns. I think this is a blind spot in the KIP. I propose to throw an IllegalStateException if subscribe(java.util.regex.Pattern) is called after subscribe(SubscriptionPattern) (and vice versa) without calling unsubscribe() in between. That is similar to the restrictions between pattern, topic, and partition subscriptions @lianetm linked above. I do not think it is worth to consider the edge case of mixed usage of the two pattern types. Does this make sense to you? \cc @dajac What do you as the original author of the KIP think? Should we update the KIP to make this clear?

@cadonna I would rather follow what we already do with subscribe today. The last one called takes precedence.

@dajac The javadocs of subscribe() say:

     * @throws IllegalStateException If {@code subscribe()} is called previously with topics, or assign is called
     *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
     *                               configured at-least one partition assignment strategy

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

Line 727 in f0087ac

    
                * @throws IllegalStateException If {@code subscribe()} is called previously with topics, or assign is called

One could argue that both subscriptions are pattern subscriptions, but they are quite different internally. I am wondering how complex it is to allow mixed usage. If it is not that complex, I agree with your proposal, otherwise I am doubting whether it is really worth it.

In any case the KIP is not clear about the expected behavior.

cadonna · 2024-02-20T10:07:10Z

I was wondering whether we should introduce a new error code to signal to the user that the regular expression is invalid. Otherwise, we would have to use the invalid request exception and it does not seem user friendly. @cadonna @lianetm What do you think about this?

@dajac Is it also possible to verify the validity of the regular expression client-side?
If we assume that the clients send valid regular expressions to the brokers, I think it would be OK to return an invalid request exception and log the error broker-side. Sending invalid regular expressions should than just be a mistake that happens during development of the clients and not something that happens during usage of the clients.

The benefit would be to find the mistakes in regular expressions without a request to the brokers.

The downside of it is that we need some way to validate the regular expressions client-side like the corresponding Google library in Java and I do not know what dependency are needed for clients in other languages.

Phuc-Hong-Tran · 2024-02-20T10:11:11Z

I was wondering whether we should introduce a new error code to signal to the user that the regular expression is invalid. Otherwise, we would have to use the invalid request exception and it does not seem user friendly. @cadonna @lianetm What do you think about this?

@dajac Is it also possible to verify the validity of the regular expression client-side?

If we assume that the clients send valid regular expressions to the brokers, I think it would be OK to return an invalid request exception and log the error broker-side. Sending invalid regular expressions should than just be a mistake that happens during development of the clients and not something that happens during usage of the clients.

The benefit would be to find the mistakes in regular expressions without a request to the brokers.

The downside of it is that we need some way to validate the regular expressions client-side like the corresponding Google library in Java and I do not know what dependency are needed for clients in other languages.

I don't think we can include the Google library in the client code. I saw the comment about it on the pull request for the implementation of the regex logic on the broker. Will find it again and quote it here.

Edit: here is the comment #14327 (review)

cadonna · 2024-02-20T10:18:41Z

I was wondering whether we should introduce a new error code to signal to the user that the regular expression is invalid. Otherwise, we would have to use the invalid request exception and it does not seem user friendly. @cadonna @lianetm What do you think about this?

@dajac Is it also possible to verify the validity of the regular expression client-side?
If we assume that the clients send valid regular expressions to the brokers, I think it would be OK to return an invalid request exception and log the error broker-side. Sending invalid regular expressions should than just be a mistake that happens during development of the clients and not something that happens during usage of the clients.
The benefit would be to find the mistakes in regular expressions without a request to the brokers.
The downside of it is that we need some way to validate the regular expressions client-side like the corresponding Google library in Java and I do not know what dependency are needed for clients in other languages.

I don't think we can include the Google library in the client code. I saw the comment about it on the pull request for the implementation of the regex logic on the broker. Will find it again and quote it here.

Edit: here is the comment #14327 (review)

Ah, I see! Thanks for the link!
Nevertheless, there it says the resolution of the regular expression regarding the existing topics on the brokers. It does not say anything about checking whether the regular expression is valid.
I definitely agree that finding topics that match the regular expression is something that needs to be strictly done on the brokers.

Phuc-Hong-Tran · 2024-02-20T10:23:00Z

@cadonna I'll see if there is a way to go around with not using the Google library to check regex validity (finger-crossed!)

cadonna

Thanks for the updates @Phuc-Hong-Tran !

Here my feedback!

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

cadonna · 2024-02-20T12:04:06Z

clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json

+    { "name": "SubscribedTopicRegex", "type": "string", "versions": "1+", "nullableVersions": "1+", "default": "null",
+      "about": "null if it didn't change since the last heartbeat; the subscribed topic regex otherwise" },


Didn't we say we want to commit this change in a separate PR?
#15188 (comment)

Will change this to a separate PR

cadonna · 2024-02-20T12:05:18Z

...nts/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManagerTest.java

@@ -86,7 +86,6 @@
 import static org.mockito.Mockito.when;

 public class OffsetsRequestManagerTest {
-


Could you please revert this change?

clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java

lianetm · 2024-02-23T16:14:42Z

Regarding the validation of the regex, I lean towards having a new error, like @dajac suggested. Just to clearly tell the user that it is using an invalid regex, without overcomplicating the experience with an InvalidRequest that would have the user wondering what exactly went wrong, probably hinting a a much larger surface area, since it indicates that something wen't wrong in the client/server interaction.

This means I think it's better to validate it on the client side too. I could be missing something, but to my understanding, the point of avoiding the new regex engine on the client was not to avoid the dependency itself, it was more a position we still keep, that the broker is the sole responsible for resolving the topics to include in a subscription when a regex is used. We would still keep that principle if we introduce the dependency on the client just to validate the regex.

Co-authored-by: Bruno Cadonna <cadonna@apache.org>

This reverts commit 53fb53d.

…o SubscriptionPattern, refactored subscribeInternal for AsyncKafkaConsumer

Phuc-Hong-Tran · 2024-03-13T02:41:46Z

Regex validity check will be included in the next pull request, I'll try to get it done by this weekend.

This reverts commit 443739a.

…ubscriptionPattern

github-actions · 2024-06-13T03:33:31Z

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

AndrewJSchofield · 2024-07-23T08:25:44Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * The pattern matching will be done on the broker against topics existing at the time of check
+     * and matching topic(s) will be returned to the client.
+     * <p>
+     * This is a short-hand for {@link #subscribe(SubscriptionPattern, ConsumerRebalanceListener)}, which


Should be "shorthand".

AndrewJSchofield · 2024-07-23T08:26:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+
+    /**
+     * Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
+     * The pattern matching will be done on the broker against topics existing at the time of check


Better would be "at the time of the check and matching topics will be returned".

AndrewJSchofield · 2024-07-23T08:27:01Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -754,6 +752,49 @@ public void subscribe(Pattern pattern) {
        delegate.subscribe(pattern);
    }

+    /**
+     * Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
+     * The pattern matching will be done on the broker against topics existing at the time of check


Better would be "at the time of the check and matching topics will be returned". No need to (s) which is a bit messy.

AndrewJSchofield · 2024-07-23T08:29:30Z

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

@@ -141,6 +141,16 @@ public synchronized void subscribe(Pattern pattern) {
        subscribe(pattern, Optional.empty());
    }

+    @Override
+    public void subscribe(SubscriptionPattern pattern, ConsumerRebalanceListener callback) {
+


It would be very easy to adapt the implementation of subscribe(Pattern pattern, ConsumerRebalanceListener callback) for this rather than leaving it blank.

Agreed. This is necessary for full coverage of the client-side changes.

@AndrewJSchofield, just a question, why do we need to adapt the implementation of subscribe(Pattern pattern, ConsumerRebalanceListener callback) while the SubscriptionPattern is to be resolved on the broker, not locally like Pattern. If we go that route the MockConsumer won't really simulate how the AsyncConsumer works

I don't mind about the details of the implementation here, but I expect you do need a mocked implemented of this new method in order to complete the PR.

AndrewJSchofield · 2024-07-23T08:29:37Z

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

+
+    @Override
+    public void subscribe(SubscriptionPattern pattern) {
+


AndrewJSchofield · 2024-07-23T08:30:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

+        } else {
+            this.pattern = pattern;
+        }
+


nit: Extraneous blank line

AndrewJSchofield · 2024-07-23T08:31:34Z

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

+public class SubscriptionPattern {
+    final private String pattern;
+    public SubscriptionPattern(final String pattern) {
+        if (pattern == null || pattern.equals("")) {


I suggest using String.isEmpty() rather than equals("").

Or use org.apache.kafka.common.utils.Utils.isBlank() which also catches the whitespace case:

Suggested change

if (pattern == null || pattern.equals("")) {

if (Utils.isBlank(pattern)) {

AndrewJSchofield · 2024-07-23T08:33:41Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java

@@ -553,7 +553,7 @@ public ConsumerGroupHeartbeatRequestData buildRequestData() {
                }
            } else {
                // SubscribedTopicRegex - only sent if has changed since the last heartbeat
-                //                      - not supported yet
+


Why not go ahead and implement the SentFields code and call data.setSubscribedTopicRegex()?

AndrewJSchofield · 2024-07-23T08:36:08Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

    /* the pattern user has requested */
    private Pattern subscribedPattern;

+    /* RE2J compatible regex */
+    private SubscriptionPattern subscriptionPattern;


I agree with @cadonna. It seems very odd to me having subscribedPattern and subscriptionPattern. I understand that there are old patterns and new patterns, but really they are achieving the same thing. Hiding the difference inside subscribedPattern seems sensible to me.

AndrewJSchofield · 2024-07-23T08:37:22Z

clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json

@@ -13,6 +13,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+//Version 1 added SubscribedTopicRegex to the request for KIP-848


I would expect you to actually define v1 including the extra field, and also v1 of the response (which I suppose is structurally identical to v0).

kirktrue

Thanks for the PR, @Phuc-Hong-Tran!

Let a few comments, mostly minor.

kirktrue · 2024-08-27T14:21:33Z

clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java

@@ -141,6 +141,16 @@ public synchronized void subscribe(Pattern pattern) {
        subscribe(pattern, Optional.empty());
    }

+    @Override
+    public void subscribe(SubscriptionPattern pattern, ConsumerRebalanceListener callback) {
+


Agreed. This is necessary for full coverage of the client-side changes.

kirktrue · 2024-08-27T14:23:33Z

clients/src/main/java/org/apache/kafka/clients/consumer/SubscriptionPattern.java

+public class SubscriptionPattern {
+    final private String pattern;
+    public SubscriptionPattern(final String pattern) {
+        if (pattern == null || pattern.equals("")) {


Or use org.apache.kafka.common.utils.Utils.isBlank() which also catches the whitespace case:

Suggested change

if (pattern == null || pattern.equals("")) {

if (Utils.isBlank(pattern)) {

kirktrue · 2024-08-27T14:27:14Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/LegacyKafkaConsumer.java

@@ -495,6 +496,16 @@ public void subscribe(Pattern pattern) {
        subscribeInternal(pattern, Optional.empty());
    }

+    @Override
+    public void subscribe(SubscriptionPattern pattern, ConsumerRebalanceListener callback) {
+        throw new IllegalArgumentException("Operation not supported in the classic group protocol");


Throwing an IllegalArgumentException seems just a tiny bit weird. What about UnsupportedOperationException instead?

kirktrue · 2024-08-27T14:29:43Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java

+    public synchronized SubscriptionPattern subscriptionPattern() {
+        return this.subscriptionPattern;
+    }
+
+    public synchronized Pattern subscribedPattern() {
+        return this.subscribedPattern;
+    }
+


Super nit: I've seen committers request that we drop unnecessary this qualifiers:

Suggested change

public synchronized SubscriptionPattern subscriptionPattern() {

return this.subscriptionPattern;

}

public synchronized Pattern subscribedPattern() {

return this.subscribedPattern;

}

public synchronized SubscriptionPattern subscriptionPattern() {

return subscriptionPattern;

}

public synchronized Pattern subscribedPattern() {

return subscribedPattern;

}

kirktrue · 2024-10-08T22:58:16Z

@Phuc-Hong-Tran—I wanted to make sure that you weren't held up waiting for input from others. After the number of review comments gets to a certain size, it can be hard to tell where we stand. Let me know if you need us to review or respond to any outstanding questions. Thanks!

phuchong.tran added 3 commits January 9, 2024 23:13

created SubscriptionPattern for new regex-subscription functionalities

61037a8

added logic in HeartBeatRequestManager to send topic regex to broker

423b6a2

added functions that use SubscriptionPattern to subscribe to topic

a90da13

phuchong.tran added 3 commits January 14, 2024 00:11

fixed checkStyle errors in SubscriptionPattern

4df71e9

Retrigger CI

819c5ec

Re-trigger tests

8e8e6c6