lucasbru commented on code in PR #15275:
URL: https://github.com/apache/kafka/pull/15275#discussion_r1474503361
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -1392,4 +1402,16 @@ public void registerStateListener(MemberStateListener
listener) {
}
this.stateUpdatesListeners.add(listener);
}
+
+ /**
+ * If either a new target assignment or new metadata is available that we
have not yet attempted
+ * to reconcile, and we are currently in state RECONCILING, trigger
reconciliation.
+ */
+ @Override
+ public PollResult poll(final long currentTimeMs) {
+ if (state == MemberState.RECONCILING && attemptReconciliation) {
+ reconcile();
Review Comment:
I think doing this in `poll` isn't strictly necessary. We could call
`reconcile` in 3 places (on metadata update, on new assignment, after an
ongoing reconciliation finished), and wouldn't have to rely on `poll` to
attempt the reconciliation. That would be mostly the same change, but simpler.
But since the ticket was written that way, for now I implemented it inside poll.
--
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]
