adikou commented on code in PR #22271:
URL: https://github.com/apache/kafka/pull/22271#discussion_r3724686225
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -2241,8 +2253,9 @@ private void release() {
currentThread.set(NO_CURRENT_THREAD);
}
- private void subscribeInternal(Pattern pattern,
Optional<ConsumerRebalanceListener> listener) {
+ private void subscribeInternal(Pattern pattern, ConsumerRebalanceListener
listener) {
acquireAndEnsureOpen();
+ subscriptions.setRebalanceListener(listener, this);
Review Comment:
Yes, this was the behaviour previously because 1-arg subscribe was
implicitly removing the listener - the KIP changes this contract and subscribe
inherently does not change the listener state; only the subscription state.
We've documented the change in the KafkaConsumer
[JavaDoc](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L734)
and I've moved the null check to
[SubscriptionState](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java#L379).
So, we'll remove the listener that was set before only from
`Consumer#setRebalanceListener(null)` ->
`SubscriptionState#setRebalanceListener(null)`.
--
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]
