You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/10/21 15:42:57 UTC
[PR] KAFKA-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
FrankYang0529 opened a new pull request, #17569:
URL: https://github.com/apache/kafka/pull/17569
Move `subscriptionState` subscription from the app thread to background thread.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17569:
URL: https://github.com/apache/kafka/pull/17569#issuecomment-2447503919
Also, @FrankYang0529 could you please remove the WIP from the PR title and merge trunk latest changes? Thanks! Almost there
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822939447
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
+ } catch (TimeoutException e) {
+ return false;
+ } finally {
+ timer.update();
+ }
Review Comment:
Hi @lianetm, thanks for review. I would prefer to keep it here, because we need to return `false` if there is exception. If we use another function to encapsulating it, we need to compare the return value and decide whether to do further operation.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822954928
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
+ } catch (TimeoutException e) {
+ return false;
+ } finally {
+ timer.update();
+ }
Review Comment:
good point, ok to keep it then.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #17569:
URL: https://github.com/apache/kafka/pull/17569
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818888151
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
Review Comment:
Yes, add a try catch block for exceptions.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818887708
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
Review Comment:
I also updated related comments with `on the next poll`. Thanks for the suggestion.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822930007
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
Hi @lianetm, thanks for the great question. I will think about the improvement. May I take the issue? Thanks.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17569:
URL: https://github.com/apache/kafka/pull/17569#issuecomment-2446002041
Hi @lianetm, thanks for the review. I addressed all comments.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818885881
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ subscriptions.subscribe(event.pattern(), event.listener());
+ metadata.requestUpdateForNewTopics();
+ updatePatternSubscription(metadata.fetch());
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that checks whether there is new assignment which matches subscribed pattern.
+ * If there is new assignment, this will make the consumer send the updated subscription.
+ */
+ private void process(final UpdateSubscriptionMetadataEvent event) {
Review Comment:
Yes, I updated it to `UpdatePatternSubscriptionEvent`. Thanks for the suggestion.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818887149
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
Review Comment:
Yes, moved the check to the beginning.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1816635997
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1610,9 +1589,10 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ applicationEventHandler.addAndGet(new UpdateSubscriptionMetadataEvent(calculateDeadlineMs(timer)));
Review Comment:
this change means that now we can throw `TimeoutException` here, which didn't happen before. The behaviour before this change (and in the classic consumer) is that if time runs out in the poll loop, we just return empty records. So should we handle the timeout here and just return false? (similar to how we do in other operations within poll, like updateFetchPositions)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
Review Comment:
nit: just to be accurate, this will just notify the `membershipMgr` about the new subscription (so the member will join the group or just send the new subscription on the next poll)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
Review Comment:
this could throw errors I expect (ie.`IllegalStateException` if we mix subscription types), so should we just catch and completeExceptionally? (maybe generalize and do all actions except for the check manager.isPresent within a try, catch all Exceptions and completeExceptionally, to play safe)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicPatternSubscriptionChangeEvent.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+public class TopicPatternSubscriptionChangeEvent extends SubscriptionChangeEvent {
Review Comment:
could you add the java doc for this new event? maybe simply saying that it is the event triggered when the consumer subscribes to a pattern (also for the TopicSubscriptionChangeEvent...triggered when the consumer subscribes to a list of topics)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ subscriptions.subscribe(event.pattern(), event.listener());
Review Comment:
similar to comment above about errors that could be thrown here and should completeExceptionally
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -233,15 +240,80 @@ public void testPollEvent() {
}
@Test
- public void testSubscriptionChangeEvent() {
- SubscriptionChangeEvent event = new SubscriptionChangeEvent();
+ public void testTopicSubscriptionChangeEvent() {
Review Comment:
could we add a negative test for the case I mentioned above, where this event fails because of mixed subscription types? The expectation (after addressing the comment above), is that the event should completeExceptionally with the same exception that `subscriptionState.subscribe` throws, `IllegalStateException` "Subscription to topics, partitions and pattern are mutually exclusive"
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -472,4 +517,33 @@ private void process(final SeekUnvalidatedEvent event) {
event.future().completeExceptionally(e);
}
}
+
+ /**
+ * This function evaluates the regex that the consumer subscribed to
+ * against the list of topic names from metadata, and updates
+ * the list of topics in subscription state accordingly
+ *
+ * @param cluster Cluster from which we get the topics
+ */
+ private void updatePatternSubscription(Cluster cluster) {
+ final Set<String> topicsToSubscribe = cluster.topics().stream()
+ .filter(subscriptions::matchesSubscribedPattern)
+ .collect(Collectors.toSet());
+ if (subscriptions.subscribeFromPattern(topicsToSubscribe)) {
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
+ if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
+ log.warn("Group membership manager not present when processing a subscribe event");
+ return;
+ }
Review Comment:
similar to comment above, I think we should perform this basic check before taking any action here
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ subscriptions.subscribe(event.pattern(), event.listener());
+ metadata.requestUpdateForNewTopics();
+ updatePatternSubscription(metadata.fetch());
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that checks whether there is new assignment which matches subscribed pattern.
+ * If there is new assignment, this will make the consumer send the updated subscription.
+ */
+ private void process(final UpdateSubscriptionMetadataEvent event) {
Review Comment:
this event is truly about updating the pattern subscription, would something like that be a clearer name? (I know we just inherited the name, but trying to improve it since we're here)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,49 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
Review Comment:
just for consistency, shouldn't this subcribe go after checking `consumerHeartbeatRequestManager.isPresent()`? (honestly I expect there is no way to get here without a hbReqMgr given that all group-related API calls validate group ID, but if we want the extra check we should do it at the beginning, before taking any action)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/UpdateSubscriptionMetadataEvent.java:
##########
@@ -0,0 +1,24 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+public class UpdateSubscriptionMetadataEvent extends CompletableApplicationEvent<Void> {
Review Comment:
let's add java doc to say this is triggered as part of the consumer poll loop to update the pattern subscription if metadata changed.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -491,48 +490,12 @@ public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
}
};
+ completeTopicSubscriptionChangeEventSuccessfully();
consumer.subscribe(Collections.singletonList(topicName), listener);
consumer.poll(Duration.ZERO);
assertTrue(callbackExecuted.get());
}
- @Test
- public void testSubscriptionRegexEvalOnPollOnlyIfMetadataChanges() {
Review Comment:
I don't see we got coverage for this in the AppEventProcessorTest, could we please add a test there? (to validate that we only eval the regex if the metadata has changed)
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818886442
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1610,9 +1589,10 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ applicationEventHandler.addAndGet(new UpdateSubscriptionMetadataEvent(calculateDeadlineMs(timer)));
Review Comment:
Updated it to catch the `TimeoutException`. Thanks.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818891701
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -491,48 +490,12 @@ public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
}
};
+ completeTopicSubscriptionChangeEventSuccessfully();
consumer.subscribe(Collections.singletonList(topicName), listener);
consumer.poll(Duration.ZERO);
assertTrue(callbackExecuted.get());
}
- @Test
- public void testSubscriptionRegexEvalOnPollOnlyIfMetadataChanges() {
Review Comment:
I create a new case `testUpdatePatternSubscriptionEventOnlyTakesEffectWhenMetadataHasNewVersion` to include same metadata version and newer metadata version cases.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818889466
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicPatternSubscriptionChangeEvent.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+public class TopicPatternSubscriptionChangeEvent extends SubscriptionChangeEvent {
Review Comment:
Yes, add java doc for all new events. Thanks.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1818890712
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -233,15 +240,80 @@ public void testPollEvent() {
}
@Test
- public void testSubscriptionChangeEvent() {
- SubscriptionChangeEvent event = new SubscriptionChangeEvent();
+ public void testTopicSubscriptionChangeEvent() {
Review Comment:
Yes, add new test cases `testTopicSubscriptionChangeEventWithIllegalSubscriptionState` and `testTopicPatternSubscriptionChangeEventWithIllegalSubscriptionState`.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1819083411
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,57 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
- * it is already a member.
+ * it is already a member on the next poll.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
- requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
+ // Join the group if not already part of it, or just send the new subscription to the broker on the next poll.
+ requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member on the next poll.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ try {
+ subscriptions.subscribe(event.pattern(), event.listener());
+ metadata.requestUpdateForNewTopics();
+ updatePatternSubscription(metadata.fetch());
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Process event that checks whether there is new assignment which matches subscribed pattern.
Review Comment:
just to be accurate, it's not checking assignments, it checks topics in metadata, and could end up adding new ones but also removing (ie. if topics matching regex are deleted). So maybe better to say something like: ..this will re-evaluate the subscribed regular expression using the latest topics from metadata, only if metadata changed. This will make the consumer.....
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicSubscriptionChangeEvent.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Application event indicating that a user calls the subscribe API for new topics.
+ * This will make the consumer join a consumer group if not part of it yet,
+ * or just send the updated subscription to the broker if it's already a member of the group on the next poll.
Review Comment:
re-arrange as above?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -233,15 +240,120 @@ public void testPollEvent() {
}
@Test
- public void testSubscriptionChangeEvent() {
- SubscriptionChangeEvent event = new SubscriptionChangeEvent();
+ public void testTopicSubscriptionChangeEvent() {
+ Set<String> topics = Set.of("topic1", "topic2");
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicSubscriptionChangeEvent event = new TopicSubscriptionChangeEvent(topics, listener, 12345);
setupProcessor(true);
+ when(subscriptionState.subscribe(topics, listener)).thenReturn(true);
+ when(metadata.requestUpdateForNewTopics()).thenReturn(1);
when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
processor.process(event);
+
+ verify(subscriptionState).subscribe(topics, listener);
+ verify(metadata).requestUpdateForNewTopics();
+ assertEquals(1, processor.metadataVersionSnapshot());
+ verify(membershipManager).onSubscriptionUpdated();
+ // verify member state doesn't transition to JOINING.
+ verify(membershipManager, never()).onConsumerPoll();
+ assertDoesNotThrow(() -> event.future().get());
+ }
+
+ @Test
+ public void testTopicSubscriptionChangeEventWithIllegalSubscriptionState() {
+ subscriptionState = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST);
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicSubscriptionChangeEvent event = new TopicSubscriptionChangeEvent(
+ Set.of("topic1", "topic2"), listener, 12345);
+
+ subscriptionState.subscribe(Pattern.compile("topic.*"), listener);
+ setupProcessor(true);
+ when(metadata.requestUpdateForNewTopics()).thenReturn(1);
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ processor.process(event);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
+ assertInstanceOf(IllegalStateException.class, e.getCause());
Review Comment:
should we add an assertion on the e.getMessage() too? would make sure we got the state exception we want, and it also makes the test intention really explicit.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -233,15 +240,120 @@ public void testPollEvent() {
}
@Test
- public void testSubscriptionChangeEvent() {
- SubscriptionChangeEvent event = new SubscriptionChangeEvent();
+ public void testTopicSubscriptionChangeEvent() {
+ Set<String> topics = Set.of("topic1", "topic2");
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicSubscriptionChangeEvent event = new TopicSubscriptionChangeEvent(topics, listener, 12345);
setupProcessor(true);
+ when(subscriptionState.subscribe(topics, listener)).thenReturn(true);
+ when(metadata.requestUpdateForNewTopics()).thenReturn(1);
when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
processor.process(event);
+
+ verify(subscriptionState).subscribe(topics, listener);
+ verify(metadata).requestUpdateForNewTopics();
+ assertEquals(1, processor.metadataVersionSnapshot());
+ verify(membershipManager).onSubscriptionUpdated();
+ // verify member state doesn't transition to JOINING.
+ verify(membershipManager, never()).onConsumerPoll();
+ assertDoesNotThrow(() -> event.future().get());
+ }
+
+ @Test
+ public void testTopicSubscriptionChangeEventWithIllegalSubscriptionState() {
+ subscriptionState = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST);
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicSubscriptionChangeEvent event = new TopicSubscriptionChangeEvent(
+ Set.of("topic1", "topic2"), listener, 12345);
+
+ subscriptionState.subscribe(Pattern.compile("topic.*"), listener);
+ setupProcessor(true);
+ when(metadata.requestUpdateForNewTopics()).thenReturn(1);
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ processor.process(event);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
+ assertInstanceOf(IllegalStateException.class, e.getCause());
+ }
+
+ @Test
+ public void testTopicPatternSubscriptionChangeEvent() {
+ Pattern pattern = Pattern.compile("topic.*");
+ Set<String> topics = Set.of("topic.1", "topic.2");
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicPatternSubscriptionChangeEvent event = new TopicPatternSubscriptionChangeEvent(pattern, listener, 12345);
+
+ setupProcessor(true);
+
+ Cluster cluster = mock(Cluster.class);
+ when(metadata.fetch()).thenReturn(cluster);
+ when(cluster.topics()).thenReturn(topics);
+ when(subscriptionState.matchesSubscribedPattern("topic.1")).thenReturn(true);
+ when(subscriptionState.matchesSubscribedPattern("topic.2")).thenReturn(true);
+ when(subscriptionState.subscribeFromPattern(topics)).thenReturn(true);
+ when(metadata.requestUpdateForNewTopics()).thenReturn(1);
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ processor.process(event);
+
+ verify(subscriptionState).subscribe(pattern, listener);
+ verify(subscriptionState).subscribeFromPattern(topics);
+ verify(metadata, times(2)).requestUpdateForNewTopics();
+ assertEquals(1, processor.metadataVersionSnapshot());
verify(membershipManager).onSubscriptionUpdated();
// verify member state doesn't transition to JOINING.
verify(membershipManager, never()).onConsumerPoll();
+ assertDoesNotThrow(() -> event.future().get());
+ }
+
+ @Test
+ public void testTopicPatternSubscriptionChangeEventWithIllegalSubscriptionState() {
+ subscriptionState = new SubscriptionState(new LogContext(), OffsetResetStrategy.EARLIEST);
+ Optional<ConsumerRebalanceListener> listener = Optional.of(new MockRebalanceListener());
+ TopicPatternSubscriptionChangeEvent event = new TopicPatternSubscriptionChangeEvent(
+ Pattern.compile("topic.*"), listener, 12345);
+
+ setupProcessor(true);
+
+ subscriptionState.subscribe(Set.of("topic.1", "topic.2"), listener);
+ processor.process(event);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
+ assertInstanceOf(IllegalStateException.class, e.getCause());
Review Comment:
similar to above about checking message
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/TopicPatternSubscriptionChangeEvent.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+/**
+ * Application event indicating that a user calls the subscribe API for a new pattern.
+ * This will make the consumer join a consumer group if not part of it yet,
+ * or just send the updated subscription to the broker if it's already a member of the group on the next poll.
Review Comment:
nit for readability: what about maybe removing "if it's already a member of the group" ? The "on the next poll' it's important but lands a bit out of place in the end
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1610,9 +1589,14 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
+ } catch (TimeoutException e) {
+ return false;
+ }
processBackgroundEvents();
+ timer.update();
Review Comment:
should this go in a `finally` clause to ensure it happens even if we timeout updating subscription?
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1814265847
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1610,7 +1589,7 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ applicationEventHandler.addAndGet(new UpdateSubscriptionMetadataEvent(calculateDeadlineMs(timer)));
Review Comment:
I think the update of the metadata is expected to be a non-blocking operation which takes the latest information it has available. I am not convinced this should be an `addAndGet` as a result.
If this does now start applying the timer to two operations, I would do a `timer.update()` before the `updateFetchPositions`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1721,13 +1697,7 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
}
fetchBuffer.retainAll(currentTopicPartitions);
- log.info("Subscribed to topic(s): {}", String.join(", ", topics));
- if (subscriptions.subscribe(new HashSet<>(topics), listener))
- this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
-
- // Trigger subscribe event to effectively join the group if not already part of it,
- // or just send the new subscription to the broker.
- applicationEventHandler.add(new SubscriptionChangeEvent());
+ applicationEventHandler.addAndGet(new TopicSubscriptionChangeEvent(new HashSet<>(topics), listener, calculateDeadlineMs(time.timer(defaultApiTimeoutMs))));
Review Comment:
Please split this long line.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/UpdateSubscriptionMetadataEvent.java:
##########
@@ -0,0 +1,24 @@
+/*
+ * 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.clients.consumer.internals.events;
+
+public class UpdateSubscriptionMetadataEvent extends CompletableApplicationEvent<Void> {
Review Comment:
nit: Extra space after `extends`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,47 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ log.info("Subscribed to topic(s): {}", String.join(", ", event.topics()));
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ log.info("Subscribed to pattern: '{}'", event.pattern());
Review Comment:
Same here for logging in `AsyncKafkaConsumer`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1688,10 +1667,7 @@ private void subscribeInternal(Pattern pattern, Optional<ConsumerRebalanceListen
if (pattern == null || pattern.toString().isEmpty())
throw new IllegalArgumentException("Topic pattern to subscribe to cannot be " + (pattern == null ?
"null" : "empty"));
- log.info("Subscribed to pattern: '{}'", pattern);
- subscriptions.subscribe(pattern, listener);
- metadata.requestUpdateForNewTopics();
- updatePatternSubscription(metadata.fetch());
+ applicationEventHandler.addAndGet(new TopicPatternSubscriptionChangeEvent(pattern, listener, calculateDeadlineMs(time.timer(defaultApiTimeoutMs))));
Review Comment:
Please split this long line.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,47 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ log.info("Subscribed to topic(s): {}", String.join(", ", event.topics()));
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
+
+ // Join the group if not already part of it, or just send the new subscription to the broker.
requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ log.info("Subscribed to pattern: '{}'", event.pattern());
+ subscriptions.subscribe(event.pattern(), event.listener());
+ metadata.requestUpdateForNewTopics();
+ updatePatternSubscription(metadata.fetch());
+ event.future().complete(null);
+ }
+
+ private void process(final UpdateSubscriptionMetadataEvent event) {
Review Comment:
Comment on this method please.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -239,16 +251,47 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
* it is already a member.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
+ log.info("Subscribed to topic(s): {}", String.join(", ", event.topics()));
Review Comment:
I'd prefer to leave this log line in `AsyncKafkaConsumer`.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1814850940
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1610,7 +1589,7 @@ private void updateLastSeenEpochIfNewer(TopicPartition topicPartition, OffsetAnd
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ applicationEventHandler.addAndGet(new UpdateSubscriptionMetadataEvent(calculateDeadlineMs(timer)));
Review Comment:
Hi @AndrewJSchofield, thanks for the review. To align with old behavior, I think we may need to use `addAndGet` here. I add a `timer.update()` before the `updateFetchPositions`.
In `ClassicKafkaConsumer`, it does a blocking operation to update assignment metadata and a non-blocking operation to wait for joining group.
https://github.com/apache/kafka/blob/553e6b4c6d912b3b619b5a1a5263b616b6d3f62e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java#L608-L621
In `ApplicationEventProcessor#process(UpdateSubscriptionMetadataEvent)`, we do a similar behavior.
https://github.com/apache/kafka/blob/76a64a8af60e4035219e016014f2f82534267d1a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java#L285-L297
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822849105
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
+ } catch (TimeoutException e) {
+ return false;
+ } finally {
+ timer.update();
+ }
Review Comment:
nit: what about encapsulating this in a `maybeUpdatePatternSubscription(timer)`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -248,16 +260,57 @@ private void process(final ListOffsetsEvent event) {
}
/**
- * Process event that indicates that the subscription changed. This will make the
+ * Process event that indicates that the subscription topics changed. This will make the
* consumer join the group if it is not part of it yet, or send the updated subscription if
- * it is already a member.
+ * it is already a member on the next poll.
*/
- private void process(final SubscriptionChangeEvent ignored) {
+ private void process(final TopicSubscriptionChangeEvent event) {
if (!requestManagers.consumerHeartbeatRequestManager.isPresent()) {
log.warn("Group membership manager not present when processing a subscribe event");
+ event.future().complete(null);
return;
}
- requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+
+ try {
+ if (subscriptions.subscribe(event.topics(), event.listener()))
+ this.metadataVersionSnapshot = metadata.requestUpdateForNewTopics();
+
+ // Join the group if not already part of it, or just send the new subscription to the broker on the next poll.
+ requestManagers.consumerHeartbeatRequestManager.get().membershipManager().onSubscriptionUpdated();
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Process event that indicates that the subscription topic pattern changed. This will make the
+ * consumer join the group if it is not part of it yet, or send the updated subscription if
+ * it is already a member on the next poll.
+ */
+ private void process(final TopicPatternSubscriptionChangeEvent event) {
+ try {
+ subscriptions.subscribe(event.pattern(), event.listener());
+ metadata.requestUpdateForNewTopics();
+ updatePatternSubscription(metadata.fetch());
+ event.future().complete(null);
+ } catch (Exception e) {
+ event.future().completeExceptionally(e);
+ }
+ }
+
+ /**
+ * Process event that re-evaluate the subscribed regular expression using the latest topics from metadata.
+ * Only if metadata changed, this will make the consumer send the updated subscription on the next poll.
Review Comment:
```suggestion
* Process event that re-evaluates the subscribed regular expression using the latest topics from metadata, only if metadata changed.
* This will make the consumer send the updated subscription on the next poll.
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
This bit got me thinking if there is maybe room for improvement here.... I would leave it as it is in this PR, just to keep the behaviour of the classic (poll loop iteration evaluates the regex if metadata changes and does not carry on until it finishes the eval), and to not "hide" such a change in this PR, but I will file a jira to think if we could improve performance here, and make this non-blocking for the new consumer, given that the eval already takes place in a diff thread
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17569:
URL: https://github.com/apache/kafka/pull/17569#issuecomment-2447655971
Yes, removed the WIP. Thanks for the reminder.
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822938162
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
oh @TaiJuWu just asked too, I will let you two sort it out as you prefer, keep me in the loop if I can help with reviews. Thanks!
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread (WIP) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822808489
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
This bit got me thinking if there is maybe room for improvement here.... I would leave it as it is in this PR, just to keep the behaviour of the classic (poll loop iteration evaluates the regex if metadata changes and does not carry on until it finishes the eval), and to not "hide" such a change in this PR, but I will file a jira to think if we could improve performance here, and make this non-blocking for the new consumer, given that the eval already takes place in a diff thread
-- update
https://issues.apache.org/jira/browse/KAFKA-17900
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822943740
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
I just saw his comment. It's good for him to handle the issue. Thanks 👍
--
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-17726: New consumer subscribe/subscribeFromPattern in background thread [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17569:
URL: https://github.com/apache/kafka/pull/17569#discussion_r1822960666
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1679,7 +1658,13 @@ private void sendPrefetches(Timer timer) {
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
maybeThrowFencedInstanceException();
offsetCommitCallbackInvoker.executeCallbacks();
- maybeUpdateSubscriptionMetadata();
+ try {
+ applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
Review Comment:
Thanks a lot!
--
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