lianetm commented on code in PR #14357: URL: https://github.com/apache/kafka/pull/14357#discussion_r1362685795
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java: ########## @@ -248,4 +261,28 @@ public void updateAssignment(ConsumerGroupHeartbeatResponseData.Assignment assig } maybeTransitionToStable(); } + + @Override + public void completeReconcile(Set<TopicPartition> revokedPartitions, + Set<TopicPartition> assignedPartitions, + Optional<KafkaException> callbackError) { + if (callbackError.isPresent()) { + // TODO: how to react to callback errors? + } + + assignmentReconciler.completeReconcile(revokedPartitions, assignedPartitions); + transitionTo(MemberState.STABLE); + // TODO: update state to signal the HeartbeatRequestManager to send an ACK heartbeat + } + + @Override + public void completeLost(Set<TopicPartition> lostPartitions, Optional<KafkaException> callbackError) { + if (callbackError.isPresent()) { + // TODO: how to react to callback errors? + } + + assignmentReconciler.completeLost(lostPartitions); + transitionTo(MemberState.UNJOINED); + // TODO: update state to signal the HeartbeatRequestManager to send an ACK heartbeat Review Comment: Thinking more about this, wondering if this couple of new states would do: - existing RECONCILING : starts when member gets new assignment and triggers callbacks. No HB. - new RECONCILIATION_COMPLETE : starts when callbacks complete. Send HB on this state (and end the state when HB is sent out, transitioning to STABLE). - existing FENCED (renamed to RELEASING_ASSIGNMENT): starts when member receives a fencing error, resets epoch to 0 and triggers callbacks. No HB. - new RELEASING_ASSIGNMENT_COMPLETE: starts when callback completes. Send HB on this state (and end state when HB is sent out, transitioning to UNJOINED) Based on these, the HB manager could send HB based on the interval or if `shouldHeartBeatNow` true (state == RECONCILIATION_COMPLETE || RELEASING_ASSIGNMENT_COMPLETE) -- 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