lianetm commented on code in PR #14357: URL: https://github.com/apache/kafka/pull/14357#discussion_r1344659132
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/AssignmentReconciler.java: ########## @@ -85,122 +111,162 @@ enum ReconciliationResult { private final BlockingQueue<BackgroundEvent> backgroundEventQueue; private Optional<RebalanceCallbackEvent> inflightCallback; - public MemberAssignmentReconciler(LogContext logContext, - SubscriptionState subscriptions, - ConsumerMetadata metadata, - BlockingQueue<BackgroundEvent> backgroundEventQueue) { + AssignmentReconciler(LogContext logContext, + SubscriptionState subscriptions, + ConsumerMetadata metadata, + BlockingQueue<BackgroundEvent> backgroundEventQueue) { this.log = logContext.logger(getClass()); this.subscriptions = subscriptions; this.metadata = metadata; this.backgroundEventQueue = backgroundEventQueue; } /** - * Perform the revocation process, if necessary, depending on the given {@link Assignment target assignment}. If the - * {@link SubscriptionState#assignedPartitions() current set of assigned partitions} includes entries that are - * <em>not</em> in the target assignment, these will be considered for revocation. If there is already a - * reconciliation in progress (revocation or assignment), this method will return without performing any - * revocation. + * Perform the reconciliation process, as necessary to meet the given {@link Assignment target assignment}. Note + * that the reconciliation is a multi-step process, and this method should be invoked on each heartbeat if + * the coordinator provides a {@link Assignment target assignment}. * * @param assignment Target {@link Assignment} * @return {@link ReconciliationResult} */ - ReconciliationResult revoke(Optional<Assignment> assignment) { + ReconciliationResult maybeReconcile(Optional<Assignment> assignment) { // Check for any outstanding operations first. If a conclusive result has already been reached, return that // before processing any further. - Optional<ReconciliationResult> inflightStatus = checkInflightStatus(); + if (inflightCallback.isPresent()) { + // We don't actually need the _result_ of the event, just to know that it's complete. + if (inflightCallback.get().future().isDone()) { + // This is the happy path--we completed the callback. Clear out our inflight callback first, though. + inflightCallback = Optional.empty(); Review Comment: What about the case where the callback execution does not complete? We need to somehow clear this I expect. (If the callback does not complete, the broker will eventually kick the member out of the group, the member will rejoin so there should be a new `inflightCallback`) -- 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