lianetm commented on code in PR #14690:
URL: https://github.com/apache/kafka/pull/14690#discussion_r1394570775
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -17,25 +17,86 @@
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import
org.apache.kafka.clients.consumer.internals.Utils.TopicPartitionComparator;
+import org.apache.kafka.common.ClusterResource;
+import org.apache.kafka.common.ClusterResourceListener;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
/**
- * Membership manager that maintains group membership for a single member,
following the new
- * consumer group protocol.
+ * Group manager for a single consumer that has a group id defined in the
config
+ * {@link ConsumerConfig#GROUP_ID_CONFIG}, to use the Kafka-based offset
management capability,
+ * and the consumer group protocol to get automatically assigned partitions
when calling the
+ * subscribe API.
+ *
+ * <p/>
+ *
+ * While the subscribe API hasn't been called (or if the consumer called
unsubscribe), this manager
+ * will only be responsible for keeping the member in the {@link
MemberState#UNSUBSCRIBED} state,
+ * where it can commit offsets to the group identified by the {@link
#groupId()}, without joining
+ * the group.
+ *
+ * <p/>
+ *
+ * If the consumer subscribe API is called, this manager will use the {@link
#groupId()} to join the
+ * consumer group, and based on the consumer group protocol heartbeats, will
handle the full
+ * lifecycle of the member as it joins the group, reconciles assignments,
handles fencing and
+ * fatal errors, and leaves the group.
+ *
* <p/>
- * This is responsible for:
- * <li>Keeping member info (ex. member id, member epoch, assignment, etc.)</li>
- * <li>Keeping member state as defined in {@link MemberState}.</li>
+ *
+ * Reconciliation process:<p/>
+ * The member accepts all assignments received from the broker, resolves topic
names from
+ * metadata, reconciles the resolved assignments, and keeps the unresolved to
be reconciled when
+ * discovered with a metadata update. Reconciliations of resolved assignments
are executed
+ * sequentially and acknowledged to the server as they complete. The
reconciliation process
+ * involves multiple async operations, so the member will continue to
heartbeat while these
+ * operations complete, to make sure that the member stays in the group while
reconciling.
+ *
* <p/>
- * Member info and state are updated based on the heartbeat responses the
member receives.
+ *
+ * Reconciliation steps:
+ * <ol>
+ * <li>Resolve topic names for all topic IDs received in the target
assignment. Topic names
+ * found in metadata are then ready to be reconciled. Topic IDs not found
are kept as
+ * unresolved, and the member request metadata updates until it resolves
them (or the broker
+ * removes it from the target assignment.</li>
+ * <li>Commit offsets if auto-commit is enabled.</li>
+ * <li>Invoke the user-defined onPartitionsRevoked listener.</li>
+ * <li>Invoke the user-defined onPartitionsAssigned listener.</li>
+ * <li>When the above steps complete, the member acknowledges the target
assignment by
+ * sending a heartbeat request back to the broker, including the full
target assignment
+ * that was just reconciled.</li>
Review Comment:
Yes, you're right, I will rephrase this. It acknowledges the reconciled
assignment, which is the subset of the target that was resolved from metadata
and actually reconciled.
--
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
