Kafka客户端实现解析：Producer与Consumer核心机制深度剖析

graph TB
    subgraph "Kafka客户端架构对比"
        subgraph "Producer架构 Message Publishing"
            P[KafkaProducer 主接口]
            S[Serializer 序列化器]
            PART[Partitioner 分区器]
            ACC[RecordAccumulator 记录累加器]
            SEND[Sender 发送器]
            NET[NetworkClient 网络客户端]
            TM[TransactionManager 事务管理器]
        end
        
        subgraph "Consumer架构 Message Consumption"
            C[KafkaConsumer 主接口]
            DS[Deserializer 反序列化器]
            SUB[SubscriptionState 订阅状态]
            FETCH[Fetcher 拉取器]
            CC[ConsumerCoordinator 消费者协调器]
            NETC[NetworkClient 网络客户端]
            OF[OffsetFetcher 偏移量拉取器]
        end
        
        subgraph "共享组件 Shared Components"
            META[Metadata 元数据管理]
            METRICS[Metrics 监控指标]
            CONFIG[Config 配置管理]
        end
        
        %% Producer流程
        P --> S
        P --> PART
        S --> ACC
        PART --> ACC
        ACC --> SEND
        SEND --> NET
        P --> TM
        
        %% Consumer流程
        C --> DS
        C --> SUB
        SUB --> FETCH
        FETCH --> NETC
        C --> CC
        CC --> OF
        
        %% 共享组件连接
        NET --> META
        NETC --> META
        NET --> METRICS
        NETC --> METRICS
        P --> CONFIG
        C --> CONFIG
    end
    
    style P fill:#e1f5fe
    style C fill:#e8f5e8
    style ACC fill:#f3e5f5
    style FETCH fill:#fff3e0

sequenceDiagram
    participant App as 应用程序
    participant P as KafkaProducer
    participant S as Serializer
    participant PART as Partitioner
    participant ACC as RecordAccumulator
    participant SEND as Sender
    participant NET as NetworkClient
    participant B as Broker
    
    Note over App,B: Kafka Producer 完整发送流程
    
    App->>P: send(ProducerRecord)
    P->>S: serialize(key, value)
    S-->>P: 返回序列化数据
    P->>PART: partition(topic, key, value)
    PART-->>P: 返回分区号
    
    P->>ACC: append(topic, partition, records)
    ACC->>ACC: 查找或创建ProducerBatch
    
    alt 批次未满
        ACC->>ACC: 添加记录到现有批次
        ACC-->>P: 返回Future
    else 批次已满或超时
        ACC->>SEND: 唤醒Sender线程
        SEND->>ACC: drain() 获取就绪批次
        SEND->>NET: send(ProduceRequest)
        NET->>B: 发送网络请求
        B-->>NET: ProduceResponse
        NET->>SEND: 处理响应
        SEND->>ACC: 完成批次处理
        ACC->>App: 触发callback
    end
    
    Note over App: 异步非阻塞返回

/**
 * KafkaProducer - Kafka生产者的主要实现类
 * 提供线程安全的异步消息发送功能
 */
public class KafkaProducer<K, V> implements Producer<K, V> {
    
    // 核心组件
    private final ProducerMetadata metadata;           // 元数据管理
    private final RecordAccumulator accumulator;       // 记录累加器
    private final Sender sender;                       // 发送器
    private final Sender.SenderThread ioThread;        // I/O线程
    private final TransactionManager transactionManager; // 事务管理器
    
    // 序列化和分区
    private final Plugin<Serializer<K>> keySerializerPlugin;     // 键序列化器
    private final Plugin<Serializer<V>> valueSerializerPlugin;   // 值序列化器
    private final Plugin<Partitioner> partitionerPlugin;        // 分区器
    
    /**
     * 异步发送消息记录的核心实现
     * 
     * @param record 要发送的消息记录
     * @param callback 发送完成后的回调函数
     * @return 包含消息元数据的Future对象
     */
    @Override
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        // 1. 拦截器预处理
        ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
        
        // 2. 实际发送处理
        return doSend(interceptedRecord, callback);
    }

    /**
     * 执行实际的发送逻辑
     */
    private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        // 1. 验证生产者状态
        throwIfProducerClosed();
        
        // 2. 回调包装器，用于拦截器后处理
        Callback interceptCallback = new InterceptorCallback<>(callback, this.interceptors, record);

        try {
            // 3. 等待元数据更新（如果需要）
            long nowMs = time.milliseconds();
            ClusterAndWaitTime clusterAndWaitTime;
            try {
                clusterAndWaitTime = waitOnMetadata(record.topic(), record.partition(), 
                                                   nowMs, maxBlockTimeMs);
            } catch (KafkaException e) {
                if (metadata.isClosed())
                    throw new KafkaException("Producer closed while send in progress", e);
                throw e;
            }
            
            nowMs += clusterAndWaitTime.waitedOnMetadataMs;
            long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);
            Cluster cluster = clusterAndWaitTime.cluster;
            
            // 4. 序列化键值对
            byte[] serializedKey;
            try {
                serializedKey = keySerializer.serialize(record.topic(), record.key());
            } catch (ClassCastException cce) {
                throw new SerializationException("无法序列化键", cce);
            }
            
            byte[] serializedValue;
            try {
                serializedValue = valueSerializer.serialize(record.topic(), record.value());
            } catch (ClassCastException cce) {
                throw new SerializationException("无法序列化值", cce);
            }
            
            // 5. 计算分区
            int partition = partition(record, serializedKey, serializedValue, cluster);
            setReadOnly(record.headers());
            Header[] headers = record.headers().toArray();

            int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(
                RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), 
                serializedKey, serializedValue, headers);
            
            // 6. 确保记录大小不超过限制
            ensureValidRecordSize(serializedSize);

            // 7. 构建时间戳
            long timestamp = record.timestamp() == null ? nowMs : record.timestamp();

            // 8. 添加到累加器进行批处理
            RecordAccumulator.RecordAppendResult result = accumulator.append(
                record.topic(), 
                partition, 
                timestamp, 
                serializedKey, 
                serializedValue, 
                headers,
                interceptCallback, 
                remainingWaitMs, 
                true, 
                nowMs, 
                cluster
            );

            // 9. 如果批次已满或新创建批次，唤醒sender线程
            if (result.batchIsFull || result.newBatchCreated) {
                log.trace("唤醒sender，因为批次已满或新创建批次");
                this.sender.wakeup();
            }

            return result.future;
        } catch (ApiException e) {
            log.debug("发送记录时发生API异常", e);
            if (callback != null) {
                callback.onCompletion(null, e);
            }
            this.errors.record();
            if (this.interceptors != null) {
                this.interceptors.onSendError(record, null, e);
            }
            return new FutureFailure(e);
        } catch (InterruptedException e) {
            this.errors.record();
            if (this.interceptors != null) {
                this.interceptors.onSendError(record, null, e);
            }
            throw new InterruptException(e);
        } catch (KafkaException e) {
            this.errors.record();
            if (this.interceptors != null) {
                this.interceptors.onSendError(record, null, e);
            }
            throw e;
        }
    }

    /**
     * 计算消息应该发送到的分区
     */
    private int partition(ProducerRecord<K, V> record, byte[] serializedKey, 
                         byte[] serializedValue, Cluster cluster) {
        if (record.partition() != null) {
            return record.partition();
        }
        return partitioner.partition(
            record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
    }
}

/**
 * RecordAccumulator - 记录累加器
 * 实现消息的批处理和内存管理，提高发送效率
 */
public class RecordAccumulator {
    
    // 批处理配置
    private final int batchSize;                           // 批次大小
    private final Compression compression;                 // 压缩算法
    private final int lingerMs;                           // 延迟发送时间
    private final ExponentialBackoff retryBackoff;        // 重试退避策略
    
    // 内存管理
    private final BufferPool free;                        // 内存池
    
    // 批次管理
    private final ConcurrentMap<String, TopicInfo> topicInfoMap = new CopyOnWriteMap<>();
    private final IncompleteBatches incomplete;           // 未完成批次追踪
    
    /**
     * 向累加器添加消息记录
     * 实现批处理逻辑的核心方法
     */
    public RecordAppendResult append(String topic,
                                   int partition,
                                   long timestamp,
                                   byte[] key,
                                   byte[] value,
                                   Header[] headers,
                                   AppendCallbacks callbacks,
                                   long maxTimeToBlock,
                                   long nowMs,
                                   Cluster cluster) throws InterruptedException {
        
        // 1. 获取或创建主题信息
        TopicInfo topicInfo = topicInfoMap.computeIfAbsent(topic, 
            k -> new TopicInfo(createBuiltInPartitioner(logContext, k, batchSize)));

        // 2. 跟踪正在进行的append操作
        appendsInProgress.incrementAndGet();
        ByteBuffer buffer = null;
        if (headers == null) headers = Record.EMPTY_HEADERS;
        
        try {
            // 3. 分区选择逻辑
            while (true) {
                // 选择有效分区
                int effectivePartition = partition == RecordMetadata.UNKNOWN_PARTITION 
                    ? topicInfo.builtInPartitioner.partition(topic, key, value, cluster)
                    : partition;

                // 4. 获取分区对应的批次队列
                Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(
                    effectivePartition, k -> new ArrayDeque<>());
                
                synchronized (dq) {
                    // 5. 尝试添加到现有批次
                    RecordAppendResult appendResult = tryAppend(timestamp, key, value, 
                                                              headers, callbacks, dq, nowMs);
                    if (appendResult != null) {
                        // 成功添加到现有批次
                        boolean enableSwitch = allBatchesFull(dq);
                        topicInfo.builtInPartitioner.updatePartitionInfo(
                            partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
                        return appendResult;
                    }
                }

                // 6. 需要创建新批次，先分配内存
                if (buffer == null) {
                    int size = Math.max(this.batchSize, 
                        AbstractRecords.estimateSizeInBytesUpperBound(
                            RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), 
                            key, value, headers));
                    
                    log.trace("为主题 {} 分区 {} 分配新的 {} 字节消息缓冲区，剩余超时 {}ms", 
                        topic, effectivePartition, size, maxTimeToBlock);
                    
                    // 这个调用可能会阻塞，如果缓冲池空间不足
                    buffer = free.allocate(size, maxTimeToBlock);
                    nowMs = time.milliseconds();
                }

                synchronized (dq) {
                    // 7. 创建新批次
                    RecordAppendResult appendResult = appendNewBatch(topic, effectivePartition, 
                        dq, timestamp, key, value, headers, callbacks, buffer, nowMs);
                    
                    // 如果成功创建新批次，buffer将被使用，设为null避免释放
                    if (appendResult.newBatchCreated) {
                        buffer = null;
                    }
                    
                    boolean enableSwitch = allBatchesFull(dq);
                    topicInfo.builtInPartitioner.updatePartitionInfo(
                        partitionInfo, appendResult.appendedBytes, cluster, enableSwitch);
                    return appendResult;
                }
            }
        } finally {
            // 8. 清理资源
            free.deallocate(buffer);
            appendsInProgress.decrementAndGet();
        }
    }

    /**
     * 尝试将记录添加到现有批次中
     */
    private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, 
                                       Header[] headers, AppendCallbacks callbacks, 
                                       Deque<ProducerBatch> deque, long nowMs) {
        if (closed) {
            throw new KafkaException("Producer已关闭，发送过程中被中断");
        }
        
        ProducerBatch last = deque.peekLast();
        if (last != null) {
            int initialBytes = last.estimatedSizeInBytes();
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, 
                                                        callbacks, nowMs);
            if (future == null) {
                // 批次已满，关闭以供发送
                last.closeForRecordAppends();
            } else {
                int appendedBytes = last.estimatedSizeInBytes() - initialBytes;
                return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), 
                                            false, appendedBytes);
            }
        }
        return null;
    }

    /**
     * 创建新的生产者批次
     */
    private RecordAppendResult appendNewBatch(String topic, int partition,
                                            Deque<ProducerBatch> dq,
                                            long timestamp, byte[] key, byte[] value,
                                            Header[] headers, AppendCallbacks callbacks,
                                            ByteBuffer buffer, long nowMs) {
        assert partition != RecordMetadata.UNKNOWN_PARTITION;

        // 再次尝试现有批次（可能在等待锁期间有其他线程创建了批次）
        RecordAppendResult appendResult = tryAppend(timestamp, key, value, 
                                                  headers, callbacks, dq, nowMs);
        if (appendResult != null) {
            return appendResult;
        }

        // 创建新批次
        MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer);
        ProducerBatch batch = new ProducerBatch(new TopicPartition(topic, partition), 
                                               recordsBuilder, nowMs);
        FutureRecordMetadata future = Objects.requireNonNull(
            batch.tryAppend(timestamp, key, value, headers, callbacks, nowMs));

        // 添加到队列和未完成批次追踪
        dq.addLast(batch);
        incomplete.add(batch);

        return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), 
                                    true, batch.estimatedSizeInBytes());
    }

    /**
     * 获取准备发送的批次
     * 由Sender线程调用以获取就绪的批次进行网络发送
     */
    public ReadyCheckResult ready(Cluster cluster, long nowMs) {
        Set<Node> readyNodes = new HashSet<>();
        long nextReadyCheckDelayMs = Long.MAX_VALUE;
        Set<String> unknownLeaderTopics = new HashSet<>();

        boolean exhausted = this.free.queued() > 0;
        
        for (Map.Entry<String, TopicInfo> topicInfoEntry : this.topicInfoMap.entrySet()) {
            String topic = topicInfoEntry.getKey();
            TopicInfo topicInfo = topicInfoEntry.getValue();

            for (Map.Entry<Integer, Deque<ProducerBatch>> entry : topicInfo.batches.entrySet()) {
                int partition = entry.getKey();
                Deque<ProducerBatch> deque = entry.getValue();

                Node leader = cluster.leaderFor(new TopicPartition(topic, partition));
                synchronized (deque) {
                    if (leader == null) {
                        unknownLeaderTopics.add(topic);
                    } else if (!readyNodes.contains(leader) && !isMuted(leader)) {
                        ProducerBatch batch = deque.peekFirst();
                        if (batch != null) {
                            long timeSinceCreationMs = max(0, nowMs - batch.createdMs);
                            boolean full = deque.size() > 1 || batch.isFull();
                            boolean expired = timeSinceCreationMs >= lingerMs;
                            boolean transactionCompleting = transactionManager != null 
                                && transactionManager.isCompleting();

                            boolean sendable = full 
                                || expired 
                                || exhausted 
                                || closed 
                                || flushInProgress() 
                                || transactionCompleting;

                            if (sendable && !batch.inRetry()) {
                                readyNodes.add(leader);
                            } else {
                                long timeLeftMs = max(lingerMs - timeSinceCreationMs, 0);
                                nextReadyCheckDelayMs = min(nextReadyCheckDelayMs, timeLeftMs);
                            }
                        }
                    }
                }
            }
        }

        return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);
    }
}

/**
 * Sender - 后台线程，负责将批次发送到Kafka集群
 * 实现异步网络I/O和重试机制
 */
public class Sender implements Runnable {
    
    private final KafkaClient client;              // 网络客户端
    private final RecordAccumulator accumulator;   // 记录累加器
    private final ProducerMetadata metadata;       // 元数据管理
    private final TransactionManager transactionManager; // 事务管理器
    
    // 配置参数
    private final int maxRequestSize;              // 最大请求大小
    private final short acks;                      // ACK配置
    private final int retries;                     // 重试次数
    private final long retryBackoffMs;             // 重试间隔

    /**
     * Sender线程的主循环
     * 持续检查并发送就绪的批次
     */
    @Override
    public void run() {
        log.debug("启动Kafka生产者I/O线程");

        while (running) {
            try {
                // 执行一轮发送循环
                runOnce();
            } catch (Exception e) {
                log.error("发送过程中发生未捕获的异常", e);
            }
        }

        log.debug("开始关闭Kafka生产者I/O线程，首先处理待处理的事务状态");
        
        // 关闭处理
        while (!forceClose && transactionManager != null && transactionManager.hasOngoingTransaction()) {
            try {
                runOnce();
            } catch (Exception e) {
                log.error("关闭过程中发生未捕获的异常", e);
            }
        }
        
        if (forceClose) {
            log.debug("强制关闭，放弃所有待处理请求");
        }
        
        // 最后清理
        this.accumulator.abortIncompleteBatches();
        this.client.close();
        log.debug("Kafka生产者I/O线程已完全关闭");
    }

    /**
     * 执行一轮发送循环
     */
    void runOnce() {
        // 1. 处理事务状态（如果启用事务）
        if (transactionManager != null) {
            if (!transactionManager.isTransactional()) {
                maybeWaitForProducerId();
            } else if (transactionManager.hasInFlightRequest() || hasPendingTransactionalRequests()) {
                client.poll(retryBackoffMs, time.milliseconds());
                return;
            } else if (transactionManager.hasAbortableError()) {
                accumulator.abortUndrainedBatches(transactionManager.lastError());
                transactionManager.close();
                return;
            } else if (transactionManager.isAborting()) {
                handleAbortableError(transactionManager.lastError());
                return;
            }
        }

        long currentTimeMs = time.milliseconds();
        long pollTimeout = sendProducerData(currentTimeMs);
        client.poll(pollTimeout, currentTimeMs);
    }

    /**
     * 发送生产者数据的核心方法
     */
    private long sendProducerData(long currentTimeMs) {
        Cluster cluster = metadata.fetch();
        
        // 1. 获取就绪的批次
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, currentTimeMs);
        
        // 2. 如果有未知Leader的主题，请求元数据更新
        if (!result.unknownLeaderTopics.isEmpty()) {
            for (String topic : result.unknownLeaderTopics)
                this.metadata.add(topic, currentTimeMs);

            log.debug("请求更新以下主题的元数据：{}", result.unknownLeaderTopics);
            this.metadata.requestUpdate();
        }

        // 3. 过滤掉未准备好的节点
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            if (!this.client.ready(node, currentTimeMs)) {
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, currentTimeMs));
            }
        }

        // 4. 为每个准备好的节点创建生产请求
        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster,
                                                                           result.readyNodes,
                                                                           this.maxRequestSize,
                                                                           currentTimeMs);
        
        // 5. 处理超时的批次
        List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(currentTimeMs);
        List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(currentTimeMs);
        expiredBatches.addAll(expiredInflightBatches);

        if (!expiredBatches.isEmpty()) {
            log.trace("过期 {} 个批次", expiredBatches.size());
        }
        for (ProducerBatch expiredBatch : expiredBatches) {
            String errorMessage = "由于delivery.timeout.ms已过期，批次在发送前已过期";
            failBatch(expiredBatch, new TimeoutException(errorMessage), false);
        }

        // 6. 发送请求
        sendProduceRequests(batches, currentTimeMs);
        
        return Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
    }

    /**
     * 发送生产请求到各个节点
     */
    private void sendProduceRequests(Map<Integer, List<ProducerBatch>> collated, long now) {
        for (Map.Entry<Integer, List<ProducerBatch>> entry : collated.entrySet()) {
            sendProduceRequest(now, entry.getKey(), entry.getValue());
        }
    }

    /**
     * 向指定节点发送生产请求
     */
    private void sendProduceRequest(long now, int destination, List<ProducerBatch> batches) {
        if (batches.isEmpty()) return;

        final Map<TopicPartition, MemoryRecords> produceRecordsByPartition = new HashMap<>(batches.size());
        final Map<TopicPartition, ProducerBatch> recordsByPartition = new HashMap<>(batches.size());

        // 准备请求数据
        for (ProducerBatch batch : batches) {
            TopicPartition tp = batch.topicPartition;
            MemoryRecords records = batch.records();

            produceRecordsByPartition.put(tp, records);
            recordsByPartition.put(tp, batch);
        }

        String transactionalId = null;
        if (transactionManager != null && transactionManager.isTransactional()) {
            transactionalId = transactionManager.transactionalId();
        }

        // 构建生产请求
        ProduceRequest.Builder requestBuilder = ProduceRequest.forMagic(RecordBatch.CURRENT_MAGIC_VALUE,
                (short) acks, requestTimeoutMs, produceRecordsByPartition, transactionalId);
        RequestCompletionHandler callback = new RequestCompletionHandler() {
            @Override
            public void onComplete(ClientResponse response) {
                handleProduceResponse(response, recordsByPartition, time.milliseconds());
            }
        };

        String nodeId = Integer.toString(destination);
        ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0, 
                                                             requestTimeoutMs, callback);
        client.send(clientRequest, now);
        log.trace("发送生产请求到 {}: {}", nodeId, requestBuilder);
    }

    /**
     * 处理生产响应
     */
    private void handleProduceResponse(ClientResponse response, Map<TopicPartition, ProducerBatch> batches, 
                                     long now) {
        RequestHeader requestHeader = response.requestHeader();
        int correlationId = requestHeader.correlationId();
        
        if (response.wasDisconnected()) {
            log.trace("取消发送到节点 {} 的请求 {}，因为连接断开", response.destination(), correlationId);
            for (ProducerBatch batch : batches.values())
                completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.NETWORK_EXCEPTION), correlationId, now);
        } else if (response.versionMismatch() != null) {
            log.warn("收到不支持的版本错误：{}", response.versionMismatch());
            for (ProducerBatch batch : batches.values())
                completeBatch(batch, new ProduceResponse.PartitionResponse(Errors.UNSUPPORTED_VERSION), correlationId, now);
        } else {
            log.trace("收到来自节点 {} 相关ID {} 的生产响应", response.destination(), correlationId);
            
            // 解析响应
            ProduceResponse produceResponse = (ProduceResponse) response.responseBody();
            for (Map.Entry<TopicPartition, ProduceResponse.PartitionResponse> entry : produceResponse.responses().entrySet()) {
                TopicPartition tp = entry.getKey();
                ProduceResponse.PartitionResponse partResp = entry.getValue();
                ProducerBatch batch = batches.get(tp);
                completeBatch(batch, partResp, correlationId, now);
            }
            this.sensors.recordLatency(response.destination(), response.requestLatencyMs());
        }
    }

    /**
     * 完成批次处理，触发回调
     */
    private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, 
                             long correlationId, long now) {
        Errors error = response.error;
        
        if (error == Errors.NONE) {
            if (batch.done(response.baseOffset, response.logAppendTime, null)) {
                log.trace("生产者批次已成功发送：correlationId={}, topic={}, partition={}, offset={}", 
                         correlationId, batch.topicPartition.topic(), batch.topicPartition.partition(), response.baseOffset);
            }
        } else if (canRetry(batch, response, now)) {
            log.trace("生产者批次可重试：correlationId={}, topic={}, partition={}, error={}", 
                     correlationId, batch.topicPartition.topic(), batch.topicPartition.partition(), error);
            reenqueueBatch(batch, now);
        } else if (error == Errors.DUPLICATE_SEQUENCE_NUMBER) {
            // 幂等性producer的重复序列号，视为成功
            if (batch.done(response.baseOffset, response.logAppendTime, null)) {
                log.debug("生产者批次重复序列号，视为成功：correlationId={}, topic={}, partition={}", 
                         correlationId, batch.topicPartition.topic(), batch.topicPartition.partition());
            }
        } else {
            final RuntimeException exception;
            if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
                exception = new TopicAuthorizationException(Collections.singleton(batch.topicPartition.topic()));
            } else if (error == Errors.CLUSTER_AUTHORIZATION_FAILED) {
                exception = new ClusterAuthorizationException("集群未授权");
            } else {
                exception = error.exception();
            }
            
            batch.done(RecordMetadata.UNKNOWN_OFFSET, RecordMetadata.NO_TIMESTAMP, exception);
        }
    }
}

sequenceDiagram
    participant App as 应用程序
    participant C as KafkaConsumer
    participant CC as ConsumerCoordinator
    participant F as Fetcher
    participant NET as NetworkClient
    participant B as Broker
    participant GC as GroupCoordinator
    
    Note over App,GC: Kafka Consumer 完整拉取流程
    
    App->>C: poll(timeout)
    C->>CC: 检查协调器状态
    
    alt 需要加入组
        CC->>GC: JoinGroupRequest
        GC->>CC: JoinGroupResponse
        CC->>GC: SyncGroupRequest
        GC->>CC: SyncGroupResponse
        CC->>C: 分区分配完成
    end
    
    C->>F: 发起fetch请求
    F->>NET: FetchRequest
    NET->>B: 网络请求
    B->>B: 读取日志数据
    B-->>NET: FetchResponse
    NET->>F: 响应处理
    F->>F: 反序列化消息
    F-->>C: 返回ConsumerRecords
    
    C->>CC: 提交偏移量（如果自动提交）
    CC->>B: OffsetCommitRequest
    
    C-->>App: 返回消息记录
    
    Note over App: 业务逻辑处理消息

/**
 * KafkaConsumer - Kafka消费者的主要实现类
 * 提供消息拉取、偏移量管理和组协调功能
 */
public class KafkaConsumer<K, V> implements Consumer<K, V> {
    
    // 核心组件
    private final ConsumerMetadata metadata;               // 元数据管理
    private final SubscriptionState subscriptions;        // 订阅状态
    private final Fetcher<K, V> fetcher;                 // 消息拉取器
    private final ConsumerCoordinator coordinator;        // 消费者协调器
    private final OffsetFetcher offsetFetcher;            // 偏移量拉取器
    
    // 反序列化器
    private final Plugin<Deserializer<K>> keyDeserializerPlugin;
    private final Plugin<Deserializer<V>> valueDeserializerPlugin;
    
    // 配置和网络
    private final ConsumerNetworkClient client;           // 网络客户端
    private final ConsumerConfig config;                  // 配置管理

    /**
     * 拉取消息记录的核心实现
     * 
     * @param timeout 最大等待时间
     * @return 拉取到的消息记录集合
     */
    @Override
    public ConsumerRecords<K, V> poll(final Duration timeout) {
        return poll(time.timer(timeout), true);
    }

    /**
     * poll方法的内部实现
     */
    private ConsumerRecords<K, V> poll(final Timer timer, final boolean includeMetadataInTimeout) {
        acquireAndEnsureOpen();
        try {
            kafkaConsumerMetrics.recordPollStart(timer.currentTimeMs());

            if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) {
                throw new IllegalStateException("Consumer未订阅任何主题或分配任何分区");
            }

            do {
                client.maybeTriggerWakeup();

                // 1. 更新分配元数据（如果需要）
                if (includeMetadataInTimeout) {
                    updateAssignmentMetadataIfNeeded(timer, false);
                } else {
                    while (!updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE), true)) {
                        log.warn("仍在等待元数据");
                    }
                }

                // 2. 执行实际的拉取操作
                final Fetch<K, V> fetch = pollForFetches(timer);
                if (!fetch.isEmpty()) {
                    // 在返回获取的记录之前，可以发送下一轮拉取
                    // 避免在用户处理获取记录时阻塞等待响应，实现流水线处理
                    if (sendFetches() > 0 || client.hasPendingRequests()) {
                        client.transmitSends();
                    }

                    if (fetch.records().isEmpty()) {
                        log.trace("从poll()返回空记录，因为consumer的位置至少为一个分区前进了");
                    }

                    return this.interceptors.onConsume(new ConsumerRecords<>(fetch.records(), fetch.nextOffsets()));
                }
            } while (timer.notExpired());

            return ConsumerRecords.empty();
        } finally {
            release();
            kafkaConsumerMetrics.recordPollEnd(timer.currentTimeMs());
        }
    }

    /**
     * 更新分配元数据
     */
    private boolean updateAssignmentMetadataIfNeeded(Timer timer, boolean waitForJoinGroup) {
        if (coordinator != null && !coordinator.poll(timer, waitForJoinGroup)) {
            return false;
        }
        return updateFetchPositions(timer);
    }

    /**
     * 更新拉取位置
     */
    private boolean updateFetchPositions(Timer timer) {
        cachedSubscriptionHasAllFetchPositions = subscriptions.hasAllFetchPositions();
        if (cachedSubscriptionHasAllFetchPositions) {
            return true;
        }

        // 如果我们还没有获取偏移量，需要获取它们
        if (!coordinator.refreshCommittedOffsetsIfNeeded(timer)) {
            return false;
        }

        // 如果仍有分区需要偏移量，获取它们
        subscriptions.resetInitializingPositions();
        if (!subscriptions.hasAllFetchPositions()) {
            offsetFetcher.resetPositionsIfNeeded();
        }

        cachedSubscriptionHasAllFetchPositions = subscriptions.hasAllFetchPositions();
        return cachedSubscriptionHasAllFetchPositions;
    }

    /**
     * 执行拉取操作
     */
    private Fetch<K, V> pollForFetches(Timer timer) {
        long pollTimeout = Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs());

        // 如果数据已经可用，立即返回
        final Fetch<K, V> fetch = collectFetch();
        if (!fetch.isEmpty()) {
            return fetch;
        }

        // 发送新的拉取请求（不会重发待处理的拉取请求）
        sendFetches();

        // 我们不希望因为缺少某些位置而被阻塞在poll中
        // 因为偏移量查找在失败后可能正在退避
        if (!cachedSubscriptionHasAllFetchPositions && pollTimeout > retryBackoffMs) {
            pollTimeout = retryBackoffMs;
        }

        log.trace("以超时时间 {} 轮询拉取", pollTimeout);

        Timer pollTimer = time.timer(pollTimeout);

        // 等待一段时间让拉取的数据到达，因为可能没有立即可用的数据
        // 使用专用pollTimer
        // 这样调用方法(poll)将正确处理整体超时
        try {
            client.poll(pollTimer);
        } catch (InterruptException e) {
            log.trace("拉取过程中发生中断", e);
            throw e;
        } finally {
            timer.update(pollTimer.currentTimeMs());
        }

        return collectFetch();
    }

    /**
     * 收集拉取的数据
     */
    private Fetch<K, V> collectFetch() {
        final Fetch<K, V> fetch = fetcher.collectFetch();
        return fetch;
    }

    /**
     * 发送拉取请求
     */
    private int sendFetches() {
        return fetcher.sendFetches();
    }
}