dajac commented on code in PR #14690: URL: https://github.com/apache/kafka/pull/14690#discussion_r1395665723
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/MemberState.java: ########## @@ -17,57 +17,113 @@ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.common.protocol.Errors; import java.util.ArrayList; import java.util.Arrays; import java.util.List; public enum MemberState { /** - * Member has not joined a consumer group yet, or has been fenced and needs to re-join. + * Member has a group id, but it is not subscribed to any topic to receive automatic + * assignments. This will be the state when the member has never subscribed, or when it has + * unsubscribed from all topics. While in this state the member can commit offsets but won't + * be an active member of the consumer group (no heartbeats sent). */ - UNJOINED, + UNSUBSCRIBED, + + /** + * Member is attempting to join a consumer group. While in this state, the member will send + * heartbeat requests on the interval, with epoch 0, until it gets a response with an epoch > 0 + * or a fatal failure. A member transitions to this state when it tries to join the group for + * the first time with a call to subscribe, or when it has been fenced and tries to re-join. + */ + JOINING, /** * Member has received a new target assignment (partitions could have been assigned or - * revoked), and it is processing it. While in this state, the member will - * invoke the user callbacks for onPartitionsAssigned or onPartitionsRevoked, and then make - * the new assignment effective. + * revoked), and it is processing it. While in this state, the member will continue to send + * heartbeat on the interval, and reconcile the assignment (it will commit offsets if + * needed, invoke the user callbacks for onPartitionsAssigned or onPartitionsRevoked, and make + * the new assignment effective). Note that while in this state the member may be trying to + * resolve metadata for the target assignment, or triggering commits/callbacks if topic names + * already resolved. */ - // TODO: determine if separate state will be needed for assign/revoke (not for now) RECONCILING, /** - * Member is active in a group (heartbeating) and has processed all assignments received. + * Member has completed reconciling an assignment received, and stays in this state only until + * the next heartbeat request is sent out to acknowledge the assignment to the server. This + * state indicates that the next heartbeat request must be sent without waiting for the + * heartbeat interval to expire. Note that once the ack is sent, the member could go back to + * {@link #RECONCILING} if it still has assignment waiting to be reconciled (assignments + * waiting for metadata, assignments for which metadata was resolved, or new assignments + * received from the broker) + */ + ACKNOWLEDGING, + + /** + * Member is active in a group and has processed all assignments received. While in this + * state, the member will send heartbeats on the interval. */ STABLE, /** - * Member transitions to this state when it receives a - * {@link org.apache.kafka.common.protocol.Errors#UNKNOWN_MEMBER_ID} or - * {@link org.apache.kafka.common.protocol.Errors#FENCED_MEMBER_EPOCH} error from the - * broker. This is a recoverable state, where the member - * gives up its partitions by invoking the user callbacks for onPartitionsLost, and then - * transitions to {@link #UNJOINED} to rejoin the group as a new member. + * Member transitions to this state when it receives a {@link Errors#UNKNOWN_MEMBER_ID} or + * {@link Errors#FENCED_MEMBER_EPOCH} error from the broker, indicating that it has been + * left out of the group. While in this state, the member will stop sending heartbeats, it + * will give up its partitions by invoking the user callbacks for onPartitionsLost, and then + * transition to {@link #JOINING} to re-join the group as a new member. */ FENCED, /** - * The member failed with an unrecoverable error + * The member transitions to this state after a call to unsubscribe. While in this state, the + * member will stop sending heartbeats, will commit offsets if needed and release its + * assignment (calling user's callback for partitions revoked or lost). When all these + * actions complete, the member will transition out of this state into {@link #LEAVING} to + * effectively leave the group. */ - FAILED; + PREPARE_LEAVING, + /** + * Member has committed offsets and releases its assignment, so it stays in this state until + * the next heartbeat request is sent out with epoch -1 to effectively leave the group. This + * state indicates that the next heartbeat request must be sent without waiting for the + * heartbeat interval to expire. + */ + LEAVING, + + /** + * The member failed with an unrecoverable error received in a heartbeat response. This in an + * unrecoverable state where the member won't send any requests to the broker and cannot + * perform any other transition. + */ + FATAL; + + /** + * Valid state transitions + */ static { - // Valid state transitions - STABLE.previousValidStates = Arrays.asList(UNJOINED, RECONCILING); - RECONCILING.previousValidStates = Arrays.asList(STABLE, UNJOINED); + STABLE.previousValidStates = Arrays.asList(JOINING, ACKNOWLEDGING); + + RECONCILING.previousValidStates = Arrays.asList(STABLE, JOINING, ACKNOWLEDGING); + + ACKNOWLEDGING.previousValidStates = Arrays.asList(RECONCILING); + + FATAL.previousValidStates = Arrays.asList(JOINING, STABLE, RECONCILING, ACKNOWLEDGING); Review Comment: > There is no transition from prepare leaving to fatal with the current usage of fatal (only when receiving fatal errors in HB response), because we don't send HB while in prepare leaving. We don't send an explicit HB while in prepare leaving. However, we continue to heartbeat so I assume that we could receive an error or be fenced while in this state. Regarding the ACK case, I think that we could receive a similar response while in ACK so the same applies. Do you agree? -- 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