chia7712 commented on code in PR #22271:
URL: https://github.com/apache/kafka/pull/22271#discussion_r3505066698
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -2165,32 +2166,39 @@ public void subscribe(Collection<String> topics,
StreamsRebalanceListener stream
.orElseThrow(() -> new IllegalStateException("Consumer was not
created to be used with Streams rebalance protocol events"))
.setRebalanceListener(streamsRebalanceListener);
- subscribeInternal(topics, Optional.empty());
+ subscribeInternal(topics, null);
}
@Override
public void subscribe(Pattern pattern) {
- subscribeInternal(pattern, Optional.empty());
+ subscribeInternal(pattern, null);
}
@Override
public void subscribe(SubscriptionPattern pattern,
ConsumerRebalanceListener listener) {
if (listener == null)
throw new IllegalArgumentException("RebalanceListener cannot be
null");
- subscribeToRegex(pattern, Optional.of(listener));
+ subscribeToRegex(pattern, listener);
}
@Override
public void subscribe(SubscriptionPattern pattern) {
- subscribeToRegex(pattern, Optional.empty());
+ subscribeToRegex(pattern, null);
}
@Override
public void subscribe(Pattern pattern, ConsumerRebalanceListener listener)
{
if (listener == null)
throw new IllegalArgumentException("RebalanceListener cannot be
null");
- subscribeInternal(pattern, Optional.of(listener));
+ subscribeInternal(pattern, listener);
+ }
+
+
+ @Override
+ public void setRebalanceListener(RebalanceListener callback) {
+ acquireAndEnsureOpen();
Review Comment:
we need to release the lock, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:
##########
@@ -607,6 +612,12 @@ public void unsubscribe() {
}
}
+ @Override
+ public void setRebalanceListener(RebalanceListener callback) {
+ acquireAndEnsureOpen();
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
