Kafka 源码解析：消费者运行机制

Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP);
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(properties);
// 订阅主题
consumer.subscribe(Collections.singleton(DEFAULT_TOPIC));

while (!Thread.currentThread().isInterrupted()) {
    ConsumerRecords<Integer, String> records = consumer.poll(TimeUnit.SECONDS.toMillis(1));
    try {
        for (final ConsumerRecord<Integer, String> record : records) {
            // ... 这里是消费逻辑
            this.printResult(record);
        }
        // 异步提交
        consumer.commitAsync();
    } catch (Throwable e) {
        // ... 异常
    } finally {
        // 同步提交
        consumer.commitSync();
    }
}

public class KafkaConsumer<K, V> implements Consumer<K, V> {

    /** 客户端 ID 生成器 */
    private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
    /** 客户端 ID */
    private final String clientId;
    /** 控制消费者与 GroupCoordinator 之间交互 */
    private final ConsumerCoordinator coordinator;
    /** key 反序列化器 */
    private final Deserializer<K> keyDeserializer;
    /** value 反序列化器 */
    private final Deserializer<V> valueDeserializer;
    /** 负责从服务端拉取消息 */
    private final Fetcher<K, V> fetcher;
    /** 拦截器集合， 在方法返回给用户之前进行拦截修改 */
    private final ConsumerInterceptors<K, V> interceptors;
    /** 时间戳工具 */
    private final Time time;
    /** 集群网络通信客户端，对 NetworkClient 的封装 */
    private final ConsumerNetworkClient client;
    /** 维护消费者的消费状态 */
    private final SubscriptionState subscriptions;
    /** 集群元数据 */
    private final Metadata metadata;
    /** 重试间隔 */
    private final long retryBackoffMs;
    /** 请求超时时间 */
    private final long requestTimeoutMs;
    /** 标识当前消费者是否关闭 */
    private volatile boolean closed = false;
    /** 记录当前正在使用 KafkaConsumer 的线程 ID，防止多个线程同时使用同一个 KafkaConsumer 对象 */
    private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD);
    /** 记录线程重入次数 */
    private final AtomicInteger refcount = new AtomicInteger(0);

    // ... 省略方法定义

}

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
    // 防止一个 KafkaConsumer 对象被多个线程同时使用，以保证线程安全
    this.acquire();
    try {
        if (topics == null) {
            throw new IllegalArgumentException("Topic collection to subscribe to cannot be null");
        } else if (topics.isEmpty()) {
            // 如果传递空的 topic 订阅列表，则视为解除订阅
            this.unsubscribe();
        } else {
            // ... 校验输入的 topic 不为 null 或空，如果是则抛出 IllegalArgumentException 异常，省略

            // 订阅当前 topic 列表
            subscriptions.subscribe(new HashSet<>(topics), listener);
            metadata.setTopics(subscriptions.groupSubscription());
        }
    } finally {
        // 线程重入计数 refcount 减 1，如果 refcount = 0，则标记当前 KafkaConsumer 对象没有线程占用
        this.release();
    }
}

this.acquire();
try {
    // do somthing here
} finally {
    this.release();
}

private void acquire() {
    // 检测当前 consumer 是否关闭，如果关闭则抛出异常
    this.ensureNotClosed();
    long threadId = Thread.currentThread().getId();
    // 如果存在多个线程使用同一个 consumer 对象，则抛出异常
    if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId)) {
        throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
    }
    // 线程重入次数加 1
    refcount.incrementAndGet();
}

private void release() {
    // 线程重入次数减 1
    if (refcount.decrementAndGet() == 0) {
        // 如果当前线程重入次数为 0，则表示当前 KafkaConsumer 对象没有线程占用
        currentThread.set(NO_CURRENT_THREAD);
    }
}

public void subscribe(Set<String> topics, ConsumerRebalanceListener listener) {
    if (listener == null) {
        throw new IllegalArgumentException("RebalanceListener cannot be null");
    }
    // 设置 topic 订阅模式为 AUTO_TOPICS
    this.setSubscriptionType(SubscriptionType.AUTO_TOPICS);
    this.listener = listener;
    // 更新本地缓存的订阅的信息
    this.changeSubscription(topics);
}

private enum SubscriptionType {
    NONE,
    /** 按照指定的 topic 的名字进行订阅，自动分配分区 */
    AUTO_TOPICS,
    /** 按照正则匹配 topic 名称进行订阅，自动分配分区 */
    AUTO_PATTERN,
    /** 用户手动指定消费的 topic 以及分区 */
    USER_ASSIGNED
}

private void setSubscriptionType(SubscriptionType type) {
    if (this.subscriptionType == SubscriptionType.NONE) {
        // NONE 表示没有设置过，设置为目标模式
        this.subscriptionType = type;
    } else if (this.subscriptionType != type) {
        // 如果之前设置过，且目标模式不是之前的模式，则抛出异常
        throw new IllegalStateException(SUBSCRIPTION_EXCEPTION_MESSAGE);
    }
}

public ConsumerRecords<K, V> poll(long timeout) {
    this.acquire();
    try {
        // 超时时间不允许设置为负数，但是允许设置为 0
        if (timeout < 0) {
            throw new IllegalArgumentException("Timeout must not be negative");
        }
        // 当前消费者未订阅任何 topic
        if (subscriptions.hasNoSubscriptionOrUserAssignment()) {
            throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
        }

        long start = time.milliseconds();
        long remaining = timeout;
        do {
            // 拉取消息，优先从本地缓存中获取，如果没有则会请求服务端，期间会尝试执行分区再分配策略，以及异步提交 offset
            Map<TopicPartition, List<ConsumerRecord<K, V>>> records = this.pollOnce(remaining);
            if (!records.isEmpty()) {
                /*
                 * 为了提升效率，在对响应的消息处理之前，先发送下一次 fetch 请求，
                 * 从而让处理消息的过程与拉取消息的过程并行，以减少等待网络 IO 的时间
                 */
                if (fetcher.sendFetches() > 0 || client.pendingRequestCount() > 0) {
                    // 如果有待发送的请求，执行一次不可中断的 poll 请求
                    client.pollNoWakeup();
                }

                if (this.interceptors == null) {
                    return new ConsumerRecords<>(records);
                } else {
                    // 如果注册了拦截器，则在返回之前先应用拦截器
                    return this.interceptors.onConsume(new ConsumerRecords<>(records));
                }
            }

            long elapsed = time.milliseconds() - start;
            remaining = timeout - elapsed;
        } while (remaining > 0);
        return ConsumerRecords.empty();
    } finally {
        this.release();
    }
}

public class Fetcher<K, V> implements SubscriptionState.Listener {

    /** 网络客户端 */
    private final ConsumerNetworkClient client;
    /** 时间戳工具 */
    private final Time time;
    /** 服务端返回的消息并不是立即响应，而是累积到 minBytes 再响应 */
    private final int minBytes;
    /** 请求时指定的服务端最大响应字节数 */
    private final int maxBytes;
    /** 累积等待的最大时长，达到该时间时，即使消息数据量不够，也会执行响应 */
    private final int maxWaitMs;
    /** 每次 fetch 操作的最大字节数 */
    private final int fetchSize;
    /** 重试间隔时间戳 */
    private final long retryBackoffMs;
    /** 每次获取 record 的最大数量 */
    private final int maxPollRecords;
    /** 是否对结果执行 CRC 校验 */
    private final boolean checkCrcs;
    /** 集群元数据 */
    private final Metadata metadata;
    /** 记录每个 topic 分区的消息消费情况 */
    private final SubscriptionState subscriptions;
    /** 每个响应在解析之前都会先转换成 CompletedFetch 对象记录到该队列中 */
    private final ConcurrentLinkedQueue<CompletedFetch> completedFetches;
    /** key 反序列化器 */
    private final Deserializer<K> keyDeserializer;
    /** value 反序列化器 */
    private final Deserializer<V> valueDeserializer;
    /** 缓存，用于对响应结果进行解析 */
    private final BufferSupplier decompressionBufferSupplier = BufferSupplier.create();
    /** 保存响应的分区、消息 起始位移等 */
    private PartitionRecords<K, V> nextInLineRecords = null;
    /** 封装在解析指定 offset 时的异常信息 */
    private ExceptionMetadata nextInLineExceptionMetadata = null;

    // ... 省略方法定义

}

public int sendFetches() {
    // 获取可以 fetch 的 topic 分区，并创建到分区 leader 副本所在节点的 FetchRequest 请求
    Map<Node, FetchRequest.Builder> fetchRequestMap = this.createFetchRequests();
    // 遍历并往各个目标节点发送 FetchRequest 请求
    for (Map.Entry<Node, FetchRequest.Builder> fetchEntry : fetchRequestMap.entrySet()) {
        final FetchRequest.Builder request = fetchEntry.getValue();
        final Node fetchTarget = fetchEntry.getKey();

        // 往目标节点发送 FetchRequest 请求
        log.debug("Sending fetch for partitions {} to broker {}", request.fetchData().keySet(), fetchTarget);
        client.send(fetchTarget, request)
                // 添加监听器用于处理 FetchResponse 响应
                .addListener(new RequestFutureListener<ClientResponse>() {
                    @Override
                    public void onSuccess(ClientResponse resp) {
                        FetchResponse response = (FetchResponse) resp.responseBody();
                        // 响应中的 topic 分区集合与请求的 topic 分区集合不匹配，忽略本次响应
                        if (!matchesRequestedPartitions(request, response)) {
                            log.warn("Ignoring fetch response containing partitions {} since it does not match the requested partitions {}", response.responseData().keySet(), request.fetchData().keySet());
                            return;
                        }

                        // 响应中的 topic 分区集合
                        Set<TopicPartition> partitions = new HashSet<>(response.responseData().keySet());
                        FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
                        // 遍历处理响应中的数据
                        for (Map.Entry<TopicPartition, FetchResponse.PartitionData> entry : response.responseData().entrySet()) {
                            TopicPartition partition = entry.getKey();
                            // 当前 topic 分区对应请求的 offset
                            long fetchOffset = request.fetchData().get(partition).offset;
                            FetchResponse.PartitionData fetchData = entry.getValue();
                            // 将结果包装成 CompletedFetch 缓存到 completedFetches 队列中
                            completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator, request.version()));
                        }

                        sensors.fetchLatency.record(resp.requestLatencyMs());
                        sensors.fetchThrottleTimeSensor.record(response.getThrottleTime());
                    }

                    @Override
                    public void onFailure(RuntimeException e) {
                        log.debug("Fetch request to {} for partitions {} failed", fetchTarget, request.fetchData().keySet(), e);
                    }
                });
    }
    // 返回本次发送的请求数目
    return fetchRequestMap.size();
}

// 获取并移除队首元素
CompletedFetch completedFetch = completedFetches.poll();
if (completedFetch == null) break; // completedFetches 已空
try {
    // 解析 CompletedFetch 成 PartitionRecords 对象
    nextInLineRecords = this.parseCompletedFetch(completedFetch);
} catch (KafkaException e) {
    if (drained.isEmpty()) {
        throw e;
    }
    // 封装当前分区 offset 对应的异常信息，在下次获取该分区 offset 消息时抛出
    nextInLineExceptionMetadata = new ExceptionMetadata(completedFetch.partition, completedFetch.fetchedOffset, e);
}

private PartitionRecords<K, V> parseCompletedFetch(CompletedFetch completedFetch) {
    TopicPartition tp = completedFetch.partition;
    FetchResponse.PartitionData partition = completedFetch.partitionData;
    long fetchOffset = completedFetch.fetchedOffset;
    int bytes = 0;
    int recordsCount = 0;
    PartitionRecords<K, V> parsedRecords = null;
    // 解析获取响应错误码
    Errors error = Errors.forCode(partition.errorCode);
    try {
        // 当前 topic 分区不允许 fetch 消息，一般是因为当前正在执行分区再分配，或者消费者被暂停
        if (!subscriptions.isFetchable(tp)) {
            log.debug("Ignoring fetched records for partition {} since it is no longer fetchable", tp);
        }
        // 正常响应
        else if (error == Errors.NONE) {
            // 获取 topic 分区对应的下次获取消息的 offset
            Long position = subscriptions.position(tp);
            if (position == null || position != fetchOffset) {
                // 请求的 offset 与响应的不匹配
                log.debug("Discarding stale fetch response for partition {} since its offset {} does not match the expected offset {}", tp, fetchOffset, position);
                return null;
            }

            List<ConsumerRecord<K, V>> parsed = new ArrayList<>();
            boolean skippedRecords = false;
            for (LogEntry logEntry : partition.records.deepEntries(decompressionBufferSupplier)) {
                // 跳过请求 offset 位置之前的消息
                if (logEntry.offset() >= position) {
                    // 封装成 ConsumerRecord 对象
                    parsed.add(this.parseRecord(tp, logEntry));
                    bytes += logEntry.sizeInBytes();
                } else {
                    skippedRecords = true;
                }
            }
            recordsCount = parsed.size();

            log.trace("Adding fetched record for partition {} with offset {} to buffered record list", tp, position);
            // 封装结果为 PartitionRecords 对象
            parsedRecords = new PartitionRecords<>(fetchOffset, tp, parsed);

            // ... 省略一些异常情况的处理

            // 更新本地记录的对应 topic 分区最新的 HW 值
            if (partition.highWatermark >= 0) {
                log.trace("Received {} records in fetch response for partition {} with offset {}", parsed.size(), tp, position);
                subscriptions.updateHighWatermark(tp, partition.highWatermark);
            }
        }
        // ... 省略对于错误响应的处理
    } finally {
        completedFetch.metricAggregator.record(tp, bytes, recordsCount);
    }

    /*
     * we move the partition to the end if we received some bytes or if there was an error.
     * This way, it's more likely that partitions for the same topic can remain together (allowing for more efficient serialization).
     */
    if (bytes > 0 || error != Errors.NONE) {
        subscriptions.movePartitionToEnd(tp);
    }

    return parsedRecords;
}

public Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords() {
    // 如果之前解析当前分区 offset 存在异常，处理该异常
    if (nextInLineExceptionMetadata != null) {
        ExceptionMetadata exceptionMetadata = nextInLineExceptionMetadata;
        nextInLineExceptionMetadata = null;
        TopicPartition tp = exceptionMetadata.partition;
        // 如果当前消费者处于运行状态，但是期望的 offset 在解析响应时存在异常，则直接抛出
        if (subscriptions.isFetchable(tp) && subscriptions.position(tp) == exceptionMetadata.fetchedOffset) {
            throw exceptionMetadata.exception;
        }
    }

    Map<TopicPartition, List<ConsumerRecord<K, V>>> drained = new HashMap<>();
    int recordsRemaining = maxPollRecords; // 剩余获取 record 的数量
    while (recordsRemaining > 0) {
        // 如果当前 topic 分区没有可以处理的记录
        if (nextInLineRecords == null || nextInLineRecords.isDrained()) {
            // ... 解析 CompletedFetch 成 PartitionRecords 对象，上面已经分析过
        } else {
            TopicPartition partition = nextInLineRecords.partition;
            // 从之前解析得到的 PartitionRecords 对象中拉取指定数量的消息
            List<ConsumerRecord<K, V>> records = this.drainRecords(nextInLineRecords, recordsRemaining);
            if (!records.isEmpty()) {
                List<ConsumerRecord<K, V>> currentRecords = drained.get(partition);
                if (currentRecords == null) {
                    drained.put(partition, records);
                } else {
                    /*
                     * 合并同一个分区的记录（发生的概率很小）
                     *
                     * this case shouldn't usually happen because we only send one fetch at a time per partition,
                     * but it might conceivably happen in some rare cases (such as partition leader changes).
                     * we have to copy to a new list because the old one may be immutable
                     */
                    List<ConsumerRecord<K, V>> newRecords = new ArrayList<>(records.size() + currentRecords.size());
                    newRecords.addAll(currentRecords);
                    newRecords.addAll(records);
                    drained.put(partition, newRecords);
                }
                recordsRemaining -= records.size();
            }
        }
    }

    // 返回每个 topic 分区拉取到的消息
    return drained;
}

private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollOnce(long timeout) {
    // 执行分区再分配策略，以及异步提交 offset
    coordinator.poll(time.milliseconds());

    if (!subscriptions.hasAllFetchPositions()) {
        /*
         * 如果存在没有分配 offset 的 topic 分区，则执行更新：
         * 1. 如果需要重置，则按照指定策略重置 offset
         * 2. 否则，尝试获取上次提交的 offset，如果结果为空则按照默认重置策略进行重置
         * 3. 否则，使用上次提交的 offset 更新本地记录的 offset 值
         */
        this.updateFetchPositions(subscriptions.missingFetchPositions());
    }

    // 尝试从本地获取缓存的消息
    Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();
    if (!records.isEmpty()) {
        return records;
    }

    // 如果本地没有直接可用的消息，则创建 FetchRequest 请求，从集群拉取消息数据
    fetcher.sendFetches();
    long now = time.milliseconds();
    long pollTimeout = Math.min(coordinator.timeToNextPoll(now), timeout);
    // 发送 FetchRequest 请求
    client.poll(pollTimeout, now, new PollCondition() {
        @Override
        public boolean shouldBlock() {
            return !fetcher.hasCompletedFetches();
        }
    });

    // 检查是否需要执行分区再分配，如果是则返回空的结果，以保证尽快对分区执行再平衡操作
    if (coordinator.needRejoin()) {
        return Collections.emptyMap();
    }

    // 获取 FetchRequest 请求返回的消息
    return fetcher.fetchedRecords();
}

private void updateFetchPositions(Set<TopicPartition> partitions) {
    // 对于需要重置 offset 的分区，请求分区 leader 副本所在节点获取对应的 offset 值
    fetcher.resetOffsetsIfNeeded(partitions);

    // 如果仍然存在没有分配 offset 的分区
    if (!subscriptions.hasAllFetchPositions(partitions)) {
        // 如果需要从 GroupCoordinator 获取上次提交的 offset，则发送 OffsetFetchRequest 请求更新
        coordinator.refreshCommittedOffsetsIfNeeded();
        /*
         * 再次尝试对未分配 offset 的分区进行更新：
         * 1. 如果需要重置，则按照指定策略重置 offset
         * 2. 如果获取到的上次提交的 offset 为空，则按照默认重置策略进行重置
         * 3. 使用上次提交的 offset 更新本地记录的 offset 值
         */
        fetcher.updateFetchPositions(partitions);
    }
}

public void poll(long now) {
    // 触发执行注册的监听 offset 提交完成的方法
    this.invokeCompletedOffsetCommitCallbacks();

    // 确保当前是 AUTO_TOPICS 或 AUTO_PATTERN（USER_ASSIGNED 不需要再平衡）订阅模式，
    // 且目标 GroupCoordinator 节点可达，如果不可达，则会尝试寻找一个可用的节点
    if (subscriptions.partitionsAutoAssigned() && this.coordinatorUnknown()) {
        this.ensureCoordinatorReady();
        now = time.milliseconds();
    }

    // 需要执行再平衡
    if (this.needRejoin()) {
        /*
         * due to a race condition between the initial metadata fetch and the initial rebalance,
         * we need to ensure that the metadata is fresh before joining initially.
         * This ensures that we have matched the pattern against the cluster's topics at least once before joining.
         *
         * 如果是 AUTO_PATTERN 订阅模式，则检查是否需要更新集群元数据
         */
        if (subscriptions.hasPatternSubscription()) {
            client.ensureFreshMetadata();
        }

        /*
         * 1. 检查目标 GroupCoordinator 节点是否准备好接收请求
         * 2. 启动心跳线程
         * 3. 执行分区再分配操作
         */
        this.ensureActiveGroup();
        now = time.milliseconds();
    }

    // 发送心跳
    this.pollHeartbeat(now);
    // 异步提交 offset
    this.maybeAutoCommitOffsetsAsync(now);
}

public abstract class AbstractCoordinator implements Closeable {

    /** 分区再分配操作超时时间 */
    protected final int rebalanceTimeoutMs;
    /** 消费者与服务端会话超时时间，超过该时间则认为与服务端断开连接 */
    private final int sessionTimeoutMs;
    /** 指定消费者被关闭时是否离开所属 group，如果为 true 的话会触发分区再分配操作 */
    private final boolean leaveGroupOnClose;
    /** 心跳机制 */
    private final Heartbeat heartbeat;
    /** 执行心跳机制的线程 */
    private HeartbeatThread heartbeatThread = null;
    /** 当前消费者所属的 group */
    protected final String groupId;
    /** 网络通信客户端 */
    protected final ConsumerNetworkClient client;
    /** 时间戳工具 */
    protected final Time time;
    /** 重试时间间隔 */
    protected final long retryBackoffMs;
    /** 标记是否需要重新发送 {@link JoinGroupRequest} 的请求条件之一 */
    private boolean rejoinNeeded = true;
    /** 标记是否需要执行发送 {@link JoinGroupRequest} 请求前的准备工作 */
    private boolean needsJoinPrepare = true;
    /** 记录当前消费者的运行状态 */
    private MemberState state = MemberState.UNJOINED;
    /** 分区再分配操作请求对应的 future 对象，避免多个请求同时执行 */
    private RequestFuture<ByteBuffer> joinFuture = null;
    /** 服务端 GroupCoordinator 所在节点 */
    private Node coordinator = null;
    /** 服务端 GroupCoordinator 返回的年代信息，用于区分两次分区再分配操作 */
    private Generation generation = Generation.NO_GENERATION;
    /** 获取可用 GroupCoordinator 节点请求对应的 future，避免多个请求同时执行 */
    private RequestFuture<Void> findCoordinatorFuture = null;

    // ... 省略方法定义

}

public final class ConsumerCoordinator extends AbstractCoordinator {

    /**
     * 消费者在发送 JoinGroupRequest 请求时会传递自己支持的分区分配策略，服务端会从所有消费者都支持的策略中选择一种，
     * 并通知 group leader 使用此分配策略进行分配
     */
    private final List<PartitionAssignor> assignors;
    /** 集群元数据信息 */
    private final Metadata metadata;
    /** 记录 topic 分区和 offset 的对应关系 */
    private final SubscriptionState subscriptions;
    /** 默认的 offset 提交完成时的 callback */
    private final OffsetCommitCallback defaultOffsetCommitCallback;
    /** 是否启用 offset 自动提交策略 */
    private final boolean autoCommitEnabled;
    /** offset 自动提交时间间隔 */
    private final int autoCommitIntervalMs;
    /** 注册的拦截器集合 */
    private final ConsumerInterceptors<?, ?> interceptors;
    /** 是否排除内部 topic，即 offset topic */
    private final boolean excludeInternalTopics;
    /** 记录正在等待异步提交 offset 的请求数目 */
    private final AtomicInteger pendingAsyncCommits;
    /** 记录每个 offset 提交对应的响应 callback */
    private final ConcurrentLinkedQueue<OffsetCommitCompletion> completedOffsetCommits;
    /** 标记当前消费者是不是 group leader */
    private boolean isLeader = false;
    /** 当前消费者成功订阅的 topic 集合 */
    private Set<String> joinedSubscription;
    /** 元数据快照，用于检测 topic 分区数量是否发生变化 */
    private MetadataSnapshot metadataSnapshot;
    /** 元数据快照，用于检测分区分配过程中分区数量是否发生变化 */
    private MetadataSnapshot assignmentSnapshot;
    /** 下一次自动提交 offset 的截止时间 */
    private long nextAutoCommitDeadline;

    // ... 省略方法定义

}

public boolean needRejoin() {
    // USER_ASSIGNED 订阅模式不需要执行分区再分配
    if (!subscriptions.partitionsAutoAssigned()) {
        return false;
    }
    // 再平衡过程中分区数量发生变化
    if (assignmentSnapshot != null && !assignmentSnapshot.equals(metadataSnapshot)) {
        return true;
    }
    // 消费者 topic 订阅信息发生变化
    if (joinedSubscription != null && !joinedSubscription.equals(subscriptions.subscription())) {
        return true;
    }
    // 其它标识需要再平衡的操作，例如分区再分配执行失败、重置年代信息等
    return super.needRejoin();
}

void joinGroupIfNeeded() {
    // 如果需要执行分区再分配，且目前正在进行中
    while (this.needRejoin() || this.rejoinIncomplete()) {
        // 再次检查目标 GroupCoordinator 节点是否准备好接收请求
        this.ensureCoordinatorReady();

        // 执行前期准备工作
        if (needsJoinPrepare) {
            /*
             * 1. 如果开启了 offset 自动提交，则同步提交 offset
             * 2. 调用注册的 ConsumerRebalanceListener 监听器的 onPartitionsRevoked 方法
             * 3. 取消当前消费者的 leader 身份（如果是的话），恢复成为一个普通的消费者
             */
            this.onJoinPrepare(generation.generationId, generation.memberId);
            needsJoinPrepare = false;
        }

        // 创建并发送 JoinGroupRequest 请求，申请加入目标 group
        RequestFuture<ByteBuffer> future = this.initiateJoinGroup();
        client.poll(future);
        // 申请加入 group 完成，将 joinFuture 置为 null，表示允许发送下一次 JoinGroupRequest 请求
        this.resetJoinGroupFuture();

        // 执行分区分配成功
        if (future.succeeded()) {
            needsJoinPrepare = true;
            this.onJoinComplete(generation.generationId, generation.memberId, generation.protocol, future.value());
        }
        // 执行分区分配失败，依据失败类型考虑是否重试
        else {
            RuntimeException exception = future.exception();
            if (exception instanceof UnknownMemberIdException ||
                    exception instanceof RebalanceInProgressException ||
                    exception instanceof IllegalGenerationException) {
                continue;
            } else if (!future.isRetriable()) {
                throw exception;
            }
            time.sleep(retryBackoffMs);
        }
    }
}

public interface ConsumerRebalanceListener {
    void onPartitionsRevoked(Collection<TopicPartition> partitions);
    void onPartitionsAssigned(Collection<TopicPartition> partitions);
}

private RequestFuture<ByteBuffer> sendJoinGroupRequest() {
    if (this.coordinatorUnknown()) {
        // 如果目标 GroupCoordinator 节点不可达，则返回异常
        return RequestFuture.coordinatorNotAvailable();
    }

    log.info("(Re-)joining group {}", groupId);
    // 构建 JoinGroupRequest 请求
    JoinGroupRequest.Builder requestBuilder = new JoinGroupRequest.Builder(
            groupId,
            sessionTimeoutMs,
            generation.memberId,
            protocolType(),
            metadata()).setRebalanceTimeout(rebalanceTimeoutMs);

    log.debug("Sending JoinGroup ({}) to coordinator {}", requestBuilder, this.coordinator);
    // 发送 JoinGroupRequest 请求，并注册结果处理器 JoinGroupResponseHandler
    return client.send(coordinator, requestBuilder).compose(new JoinGroupResponseHandler());
}

synchronized (AbstractCoordinator.this) {
    if (state != MemberState.REBALANCING) {
        // 在接收到响应之前，消费者的状态发生变更（可能已经从所属 group 离开），抛出异常
        future.raise(new UnjoinedGroupException());
    } else {
        // 基于响应，更新 group 的年代信息
        generation = new Generation(
                joinResponse.generationId(), joinResponse.memberId(), joinResponse.groupProtocol());
        rejoinNeeded = false;
        // 如果当前消费者是 group 中的 leader 角色
        if (joinResponse.isLeader()) {
            /*
             * 基于分区分配策略执行分区分配，leader 需要关注当前 group 中所有消费者订阅的 topic，
             * 并发送 SyncGroupRequest 请求反馈分区分配结果给 GroupCoordinator 节点
             */
            onJoinLeader(joinResponse)
                    // 这里调用 chain 方法，是希望当 SyncGroupResponse 处理完成之后，能够将结果传递给 future
                    .chain(future);
        } else {
            // 如果是 follower 消费者，则只关注自己订阅的 topic，这一步仅发送 SyncGroupRequest 请求
            onJoinFollower().chain(future);
        }
    }
}

private RequestFuture<ByteBuffer> onJoinLeader(JoinGroupResponse joinResponse) {
    try {
        // 基于分区分配策略分配分区
        Map<String, ByteBuffer> groupAssignment = this.performAssignment(
                joinResponse.leaderId(), joinResponse.groupProtocol(), joinResponse.members());

        // 创建 SyncGroupRequest 请求，反馈分区分配结果给 GroupCoordinator 节点
        SyncGroupRequest.Builder requestBuilder =
                new SyncGroupRequest.Builder(groupId, generation.generationId, generation.memberId, groupAssignment);
        log.debug("Sending leader SyncGroup for group {} to coordinator {}: {}", groupId, this.coordinator, requestBuilder);
        // 发送 SyncGroupRequest 请求
        return this.sendSyncGroupRequest(requestBuilder);
    } catch (RuntimeException e) {
        return RequestFuture.failure(e);
    }
}

protected Map<String, ByteBuffer> performAssignment(String leaderId, // leader 消费者 ID
                                                    String assignmentStrategy, // 服务端最终确定的分区分配策略
                                                    Map<String, ByteBuffer> allSubscriptions // group 名下所有消费者的 topic 订阅信息
) {
    // 从消费者支持的分区分配策略集合中选择指定策略对应的分区分配器
    PartitionAssignor assignor = this.lookupAssignor(assignmentStrategy);
    if (assignor == null) {
        throw new IllegalStateException("Coordinator selected invalid assignment protocol: " + assignmentStrategy);
    }

    // 解析封装 topic 订阅信息
    Set<String> allSubscribedTopics = new HashSet<>(); // 记录 group 名下所有消费者订阅的 topic 集合
    Map<String, Subscription> subscriptions = new HashMap<>(); // Map<String, ByteBuffer> -> Map<String, Subscription>
    for (Map.Entry<String, ByteBuffer> subscriptionEntry : allSubscriptions.entrySet()) {
        // ByteBuffer -> Subscription
        Subscription subscription = ConsumerProtocol.deserializeSubscription(subscriptionEntry.getValue());
        subscriptions.put(subscriptionEntry.getKey(), subscription);
        allSubscribedTopics.addAll(subscription.topics());
    }

    /*
     * 对于 leader 消费者来说，需要关注 group 名下所有消费者订阅的 topic，
     * 以保证当相应 topic 对应的元数据发生变化，能够感知
     */
    this.subscriptions.groupSubscribe(allSubscribedTopics);
    metadata.setTopics(this.subscriptions.groupSubscription());

    // 分区再分配之后，检测是否需要更新集群元数据信息，如果需要则立即更新
    client.ensureFreshMetadata();
    // 标记当前消费者为 leader 角色
    isLeader = true;

    log.debug("Performing assignment for group {} using strategy {} with subscriptions {}", groupId, assignor.name(), subscriptions);

    /*
     * 基于分区分配器（range/round-robin）执行分区分配，
     * 返回结果：key 是消费者 ID，value 是对应的分区分配结果
     */
    Map<String, Assignment> assignment = assignor.assign(metadata.fetch(), subscriptions);

    // 记录所有完成分配的 topic 集合
    Set<String> assignedTopics = new HashSet<>();
    for (Assignment assigned : assignment.values()) {
        for (TopicPartition tp : assigned.partitions())
            assignedTopics.add(tp.topic());
    }
    // 如果 group 中存在一些已经订阅的 topic 并未分配，则日志记录
    if (!assignedTopics.containsAll(allSubscribedTopics)) {
        Set<String> notAssignedTopics = new HashSet<>(allSubscribedTopics);
        notAssignedTopics.removeAll(assignedTopics);
        log.warn("The following subscribed topics are not assigned to any members in the group {} : {} ", groupId, notAssignedTopics);
    }

    // 如果分配的 topic 集合包含一些未订阅的 topic 集合
    if (!allSubscribedTopics.containsAll(assignedTopics)) {
        // 日志记录这些未订阅的 topic
        Set<String> newlyAddedTopics = new HashSet<>(assignedTopics);
        newlyAddedTopics.removeAll(allSubscribedTopics);
        log.info("The following not-subscribed topics are assigned to group {}, and their metadata will be fetched from the brokers : {}", groupId, newlyAddedTopics);

        // 将这些已分配但是未订阅的 topic 添加到 group 集合中
        allSubscribedTopics.addAll(assignedTopics);
        this.subscriptions.groupSubscribe(allSubscribedTopics);
        metadata.setTopics(this.subscriptions.groupSubscription());
        client.ensureFreshMetadata(); // 更新元数据信息
    }

    // 更新本地缓存的元数据信息快照
    assignmentSnapshot = metadataSnapshot;

    log.debug("Finished assignment for group {}: {}", groupId, assignment);

    // 对分区分配结果进行序列化，后续需要反馈给集群
    Map<String, ByteBuffer> groupAssignment = new HashMap<>();
    for (Map.Entry<String, Assignment> assignmentEntry : assignment.entrySet()) {
        ByteBuffer buffer = ConsumerProtocol.serializeAssignment(assignmentEntry.getValue());
        groupAssignment.put(assignmentEntry.getKey(), buffer);
    }

    return groupAssignment;
}

protected void onJoinComplete(int generation, String memberId, String assignmentStrategy, ByteBuffer assignmentBuffer) {
    // only the leader is responsible for monitoring for metadata changes (i.e. partition changes)
    if (!isLeader) {
        assignmentSnapshot = null;
    }

    // 获取最终确定的分区分配策略对应的分区分配器
    PartitionAssignor assignor = this.lookupAssignor(assignmentStrategy);
    if (assignor == null) {
        throw new IllegalStateException("Coordinator selected invalid assignment protocol: " + assignmentStrategy);
    }

    // 反序列化获取分区分配信息
    Assignment assignment = ConsumerProtocol.deserializeAssignment(assignmentBuffer);
    // 标记需要从 GroupCoordinator 节点获取最近提交的 offset 值
    subscriptions.needRefreshCommits();
    // 设置每个 topic 分区对应的消费状态
    subscriptions.assignFromSubscribed(assignment.partitions());

    // 遍历获取新分配的 topic，并更新本地记录的订阅信息
    Set<String> addedTopics = new HashSet<>();
    for (TopicPartition tp : subscriptions.assignedPartitions()) {
        if (!joinedSubscription.contains(tp.topic())) {
            addedTopics.add(tp.topic()); // 新分配的分区
        }
    }
    if (!addedTopics.isEmpty()) {
        Set<String> newSubscription = new HashSet<>(subscriptions.subscription());
        Set<String> newJoinedSubscription = new HashSet<>(joinedSubscription);
        newSubscription.addAll(addedTopics);
        newJoinedSubscription.addAll(addedTopics);

        // 使用 AUTO_PATTERN 模式进行订阅
        subscriptions.subscribeFromPattern(newSubscription);
        joinedSubscription = newJoinedSubscription;
    }

    // 更新本地缓存的集群元数据信息
    metadata.setTopics(subscriptions.groupSubscription());
    client.ensureFreshMetadata();

    // give the assignor a chance to update internal state based on the received assignment
    assignor.onAssignment(assignment);

    // 重置下次自动提交 offset 的截止时间
    nextAutoCommitDeadline = time.milliseconds() + autoCommitIntervalMs;

    // 应用监听分区再分配操作完成的监听器
    ConsumerRebalanceListener listener = subscriptions.listener();
    log.info("Setting newly assigned partitions {} for group {}", subscriptions.assignedPartitions(), groupId);
    try {
        Set<TopicPartition> assigned = new HashSet<>(subscriptions.assignedPartitions());
        listener.onPartitionsAssigned(assigned);
    } catch (WakeupException | InterruptException e) {
        throw e;
    } catch (Exception e) {
        log.error("User provided listener {} for group {} failed on partition assignment", listener.getClass().getName(), groupId, e);
    }
}

+ maybeAutoCommitOffsetsSync
| ---- + commitOffsetsSync

+ maybeAutoCommitOffsetsNow / maybeAutoCommitOffsetsAsync
| ---- + doAutoCommitOffsetsAsync
| ---- | ---- + commitOffsetsAsync
| ---- | ---- | ---- + doCommitOffsetsAsync

private void maybeAutoCommitOffsetsSync(long timeoutMs) {
    if (autoCommitEnabled) {
        // 获取当前消费者订阅的所有 topic 分区，以及分区对应的消费状态信息
        Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
        try {
            log.debug("Sending synchronous auto-commit of offsets {} for group {}", allConsumedOffsets, groupId);
            // 执行同步 offset 提交
            if (!this.commitOffsetsSync(allConsumedOffsets, timeoutMs)) {
                log.debug("Auto-commit of offsets {} for group {} timed out before completion", allConsumedOffsets, groupId);
            }
        }
        // ... 省略异常处理
    }
}

public boolean commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets, long timeoutMs) {
    // 触发注册的监听 offset 提交完成的方法
    this.invokeCompletedOffsetCommitCallbacks();
    // 如果提交的 offset 数据为空，则直接返回
    if (offsets.isEmpty()) {
        return true;
    }

    long now = time.milliseconds();
    long startMs = now;
    long remainingMs = timeoutMs;
    do {
        // 如果目标 GroupCoordinator 节点不可用
        if (this.coordinatorUnknown()) {
            // 尝试寻找负载最小且可用的 GroupCoordinator 节点
            if (!this.ensureCoordinatorReady(now, remainingMs)) {
                // 如果目标 GroupCoordinator 节点未准备好接收请求
                return false;
            }
            remainingMs = timeoutMs - (time.milliseconds() - startMs);
        }

        // 创建并发送 OffsetCommitRequest 请求，提交 offset 值
        RequestFuture<Void> future = this.sendOffsetCommitRequest(offsets);
        client.poll(future, remainingMs);

        // 提交成功
        if (future.succeeded()) {
            if (interceptors != null) {
                interceptors.onCommit(offsets);
            }
            return true;
        }
        // 提交失败，且不可重试，则抛出异常
        if (!future.isRetriable()) {
            throw future.exception();
        }

        time.sleep(retryBackoffMs);

        now = time.milliseconds();
        remainingMs = timeoutMs - (now - startMs);
    } while (remainingMs > 0);

    return false;
}

private RequestFuture<Void> sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
    if (offsets.isEmpty()) {
        // 如果没有请求的数据，则直接返回
        return RequestFuture.voidSuccess();
    }

    // 获取 GroupCoordinator 节点，并检查其可达性
    Node coordinator = this.coordinator();
    if (coordinator == null) {
        return RequestFuture.coordinatorNotAvailable();
    }

    // 封装每个分区对应提交的 offset 数据
    Map<TopicPartition, OffsetCommitRequest.PartitionData> offsetData = new HashMap<>(offsets.size());
    for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
        OffsetAndMetadata offsetAndMetadata = entry.getValue();
        if (offsetAndMetadata.offset() < 0) {
            // 非法的 offset 值
            return RequestFuture.failure(new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset()));
        }
        // key 是分区，value 是分区对应的请求数据
        offsetData.put(entry.getKey(),
                new OffsetCommitRequest.PartitionData(offsetAndMetadata.offset(), offsetAndMetadata.metadata()));
    }

    // 获取当前消费者所属 group 的年代信息
    final Generation generation;
    if (subscriptions.partitionsAutoAssigned()) {
        // 如果是 AUTO_TOPICS 或 AUTO_PATTERN 订阅模式，则获取年代信息
        generation = this.generation();
    } else {
        // 对于 USER_ASSIGNED 模式，因为不涉及到分区再分配操作，所以没有年代信息
        generation = Generation.NO_GENERATION;
    }
    if (generation == null) {
        // 如果获取 group 年代信息失败，则说明当前消费者并不属于该 group，抛出异常，需要执行分区再分配
        return RequestFuture.failure(new CommitFailedException());
    }

    // 创建 OffsetCommitRequest 请求
    OffsetCommitRequest.Builder builder =
            new OffsetCommitRequest.Builder(groupId, offsetData)
                    .setGenerationId(generation.generationId)
                    .setMemberId(generation.memberId)
                    .setRetentionTime(OffsetCommitRequest.DEFAULT_RETENTION_TIME);

    log.trace("Sending OffsetCommit request with {} to coordinator {} for group {}", offsets, coordinator, groupId);

    // 发送 OffsetCommitRequest 请求，并注册响应处理器
    return client.send(coordinator, builder).compose(new OffsetCommitResponseHandler(offsets));
}

public void handle(OffsetCommitResponse commitResponse, RequestFuture<Void> future) {
    sensors.commitLatency.record(response.requestLatencyMs());
    Set<String> unauthorizedTopics = new HashSet<>();

    // 遍历对所有 topic 分区的响应
    for (Map.Entry<TopicPartition, Short> entry : commitResponse.responseData().entrySet()) {
        TopicPartition tp = entry.getKey();
        OffsetAndMetadata offsetAndMetadata = offsets.get(tp);
        long offset = offsetAndMetadata.offset();

        // 获取当前分区对应的响应错误码
        Errors error = Errors.forCode(entry.getValue());
        // 正常响应
        if (error == Errors.NONE) {
            log.debug("Group {} committed offset {} for partition {}", groupId, offset, tp);
            if (subscriptions.isAssigned(tp)) {
                // 更新分区对应的消费状态
                subscriptions.committed(tp, offsetAndMetadata);
            }
        }
        //  ... 省略异常响应处理的过程
    }

    if (!unauthorizedTopics.isEmpty()) {
        log.error("Not authorized to commit to topics {} for group {}", unauthorizedTopics, groupId);
        future.raise(new TopicAuthorizationException(unauthorizedTopics));
    } else {
        future.complete(null);
    }
}

private void maybeAutoCommitOffsetsAsync(long now) {
    if (autoCommitEnabled) {
        if (this.coordinatorUnknown()) {
            // 目标 GroupCoordinator 节点不可达，稍后再试
            this.nextAutoCommitDeadline = now + retryBackoffMs;
        } else if (now >= nextAutoCommitDeadline) {
            // 时间已到，执行异步自动提交
            this.nextAutoCommitDeadline = now + autoCommitIntervalMs;
            this.doAutoCommitOffsetsAsync();
        }
    }
}

private void doCommitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, final OffsetCommitCallback callback) {
    // 标记需要从 GroupCoordinator 节点获取最近提交的 offset 值
    subscriptions.needRefreshCommits();
    // 创建并发送 OffsetCommitRequest 请求
    RequestFuture<Void> future = this.sendOffsetCommitRequest(offsets);
    // 封装 callback，用于监听 offset 提交结果
    final OffsetCommitCallback cb = callback == null ? defaultOffsetCommitCallback : callback;
    future.addListener(new RequestFutureListener<Void>() {
        @Override
        public void onSuccess(Void value) {
            if (interceptors != null) {
                interceptors.onCommit(offsets);
            }
            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, null));
        }

        @Override
        public void onFailure(RuntimeException e) {
            Exception commitException = e;
            if (e instanceof RetriableException) {
                commitException = new RetriableCommitFailedException(e);
            }
            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
        }
    });
}