lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1776831516
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -458,14 +466,24 @@ String memberIdInfoForLog() {
}
/**
- * Join the group with the updated subscription, if the member is not part
of it yet. If the
- * member is already part of the group, this will only ensure that the
updated subscription
+ * Set {@link #subscriptionUpdated} to true to indicate that the
subscription has been updated.
+ * The next {@link #maybeJoinGroup()} will join the group with the updated
subscription, if the member is not part of it yet.
+ * If the member is already part of the group, this will only ensure that
the updated subscription
* is included in the next heartbeat request.
* <p/>
* Note that list of topics of the subscription is taken from the shared
subscription state.
*/
public void onSubscriptionUpdated() {
- if (state == MemberState.UNSUBSCRIBED) {
+ subscriptionUpdated.compareAndSet(false, true);
+ }
+
+ /**
+ * Join the group if the member is not part of it yet. This function
separates {@link #transitionToJoining}
+ * from the {@link #onSubscriptionUpdated} to fulfill the requirement of
the "rebalances will only occur during an
+ * active call to {@link
org.apache.kafka.clients.consumer.KafkaConsumer#poll(Duration)}"
+ */
+ public void maybeJoinGroup() {
Review Comment:
nit: what do you think about calling this something like `onConsumerPoll`?
Just to make the func self explanatory (on poll -> transition to joining), no
need to find usages to understand what this is all about. No strong feeling
though, up to you.
--
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
