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
