philipnee commented on code in PR #14638:
URL: https://github.com/apache/kafka/pull/14638#discussion_r1372112998
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -1081,4 +1026,74 @@ boolean updateAssignmentMetadataIfNeeded(Timer timer) {
// logic
return updateFetchPositions(timer);
}
+
+ @Override
+ public void subscribe(Collection<String> topics) {
+ subscribeInternal(topics, Optional.empty());
+ }
+
+ @Override
+ public void subscribe(Collection<String> topics, ConsumerRebalanceListener
listener) {
+ if (listener == null)
+ throw new IllegalArgumentException("RebalanceListener cannot be
null");
+
+ subscribeInternal(topics, Optional.of(listener));
+ }
+
+ @Override
+ public void subscribe(Pattern pattern) {
+ subscribeInternal(pattern, Optional.empty());
+ }
+
+ @Override
+ public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)
{
+ if (listener == null)
+ throw new IllegalArgumentException("RebalanceListener cannot be
null");
+
+ subscribeInternal(pattern, Optional.of(listener));
+ }
+
+ private void subscribeInternal(Pattern pattern,
Optional<ConsumerRebalanceListener> listener) {
+ maybeThrowInvalidGroupIdException();
+ if (pattern == null || pattern.toString().isEmpty())
+ throw new IllegalArgumentException("Topic pattern to subscribe to
cannot be " + (pattern == null ?
+ "null" : "empty"));
+
+ throwIfNoAssignorsConfigured();
+ log.info("Subscribed to pattern: '{}'", pattern);
+ subscriptions.subscribe(pattern, listener);
+ updatePatternSubscription(metadata.fetch());
+ metadata.requestUpdateForNewTopics();
+ }
+
+ private void subscribeInternal(Collection<String> topics,
Optional<ConsumerRebalanceListener> listener) {
+ maybeThrowInvalidGroupIdException();
+ if (topics == null)
+ throw new IllegalArgumentException("Topic collection to subscribe
to cannot be null");
+ if (topics.isEmpty()) {
+ // treat subscribing to empty topic list as the same as
unsubscribing
+ unsubscribe();
+ } else {
+ for (String topic : topics) {
+ if (isBlank(topic))
+ throw new IllegalArgumentException("Topic collection to
subscribe to cannot contain null or empty topic");
+ }
+
+ throwIfNoAssignorsConfigured();
+
+ // Clear the buffered data which are not a part of newly assigned
topics
+ final Set<TopicPartition> currentTopicPartitions = new HashSet<>();
+
+ for (TopicPartition tp : subscriptions.assignedPartitions()) {
+ if (topics.contains(tp.topic()))
Review Comment:
ah, seems like this is done for my comment above
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
