adikou commented on code in PR #22271:
URL: https://github.com/apache/kafka/pull/22271#discussion_r3520719405
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:
##########
@@ -1004,8 +1062,42 @@ synchronized void movePartitionToEnd(TopicPartition tp) {
assignment.moveToEnd(tp);
}
- public synchronized Optional<ConsumerRebalanceListener>
rebalanceListener() {
- return rebalanceListener;
+ private synchronized Optional<RebalanceListener> rebalanceListener() {
+ return Optional.ofNullable(rebalanceListener.get());
+ }
+
+ private synchronized Consumer<?, ?> rebalanceConsumerOrThrow() {
+ return Optional.ofNullable(rebalanceConsumer.get())
+ .orElseThrow(() -> new IllegalStateException("Rebalance
consumer has not been initialized"));
+ }
+
+ /** @return true if a {@link RebalanceListener} has been registered with
this subscription. */
+ public synchronized boolean hasRebalanceListener() {
+ return rebalanceListener.get() != null;
+ }
+
+ /** Invoke {@link RebalanceListener#onPartitionsAssigned(Collection,
RebalanceConsumer)} on the given partitions.*/
+ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+ rebalanceListener().ifPresent(rl ->
invokeRebalanceListener(partitions, rl::onPartitionsAssigned));
Review Comment:
It's subtle - my initial impl had invokeRebalanceListeners as synchronized
so they were all in the same critical section. but this broke because we were
running user code under the lock and it ran into a deadlock.
After I removed it became not atomic like you've caught.
It would've have helped if we used the `@GuardedBy` anno - my IDE would've
caught it if we had declared the guards at the beginning.
Any reason why we don't want to introduce jsr305? Even nullity annotations
are all free form strings in javadocs. If there's interest we can start adding
it in with a KIP if it includes public packages or at least internal packages.
Actually, I see #21337 shipped. Maybe we can roll our own annotations
somewhere there. Would that be better?
--
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]
