lianetm commented on code in PR #14690:
URL: https://github.com/apache/kafka/pull/14690#discussion_r1395138981
##########
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);
+
+ FENCED.previousValidStates = Arrays.asList(JOINING, STABLE,
RECONCILING, ACKNOWLEDGING);
Review Comment:
Not with the current usage of fatal as I see it (message above)
--
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
