lianetm commented on code in PR #14357:
URL: https://github.com/apache/kafka/pull/14357#discussion_r1333212830
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -237,4 +253,64 @@ public void
updateAssignment(ConsumerGroupHeartbeatResponseData.Assignment assig
}
maybeTransitionToStable();
}
+
+ /**
+ * Performs the reconciliation between the current topic assignment and
the {@link Assignment} provided in the
+ * heartbeat response. Per the KIP-848 protocol, we perform the following:
+ *
+ * <ol>
+ * <li>Revoke partitions, if any</li>
+ * <li>Heartbeat to acknowledge revoked partitions</li>
+ * <li>Assign partitions, if any</li>
+ * <li>Heartbeat to acknowledge assigned partitions</li>
+ * </ol>
+ *
+ * TODO: What are the the state changes here?
+ * TODO: Is SubscriptionState sufficient to build the next heartbeat
request?
+ * TODO: Where do we need to call
ConsumerRebalanceListener.onPartitionsLost()?
Review Comment:
I don't think invoking the `onPartitionsLost` should be driven by the state
machine but just for the record, we need to call `onPartitionsLost` whenever
the member is left out of the group and it has an assignment. The basic case is
when we receive a HB response with a fencing error.
This needs more thought, mainly around what other cases would lead to a
member out of a group while still holding an assignment.
--
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]
