graph TD
    subgraph "上游调用方"
        App[用户应用程序]
        Streams[Kafka Streams]
        Connect[Kafka Connect]
    end
    
    subgraph "Clients 模块"
        Producer[KafkaProducer]
        Consumer[KafkaConsumer]
        Admin[AdminClient]
        Network[NetworkClient]
        Metadata[Metadata]
    end
    
    subgraph "下游依赖"
        Broker[Kafka Broker]
        Serializer[Serializer/Deserializer]
        Interceptor[Interceptor]
    end
    
    App --> Producer
    App --> Consumer
    App --> Admin
    Streams --> Producer
    Streams --> Consumer
    Connect --> Producer
    Connect --> Consumer
    
    Producer --> Network
    Consumer --> Network
    Admin --> Network
    Producer --> Serializer
    Consumer --> Serializer
    Producer --> Interceptor
    Consumer --> Interceptor
    Producer --> Metadata
    Consumer --> Metadata
    
    Network --> Broker
    Metadata --> Broker

stateDiagram-v2
    [*] --> 初始化: new KafkaProducer(config)
    初始化 --> 运行中: 构造完成
    运行中 --> 运行中: send() 发送消息
    运行中 --> 运行中: flush() 等待发送完成
    运行中 --> 事务中: beginTransaction()
    事务中 --> 事务中: send() 事务性消息
    事务中 --> 运行中: commitTransaction() / abortTransaction()
    运行中 --> 关闭中: close()
    关闭中 --> [*]: 资源释放完成
    
    note right of 初始化

        - 初始化网络客户端
        - 创建 RecordAccumulator
        - 启动 Sender 线程
    end note
    
    note right of 事务中
        - 只有配置了 transactional.id
        - 才能进入事务状态
    end note

stateDiagram-v2
    [*] --> 初始化: new KafkaConsumer(config)
    初始化 --> 订阅中: subscribe(topics)
    订阅中 --> 运行中: poll() 首次调用
    运行中 --> 运行中: poll() 拉取消息
    运行中 --> 运行中: commitSync() / commitAsync()
    运行中 --> 再平衡中: 触发再平衡
    再平衡中 --> 运行中: 再平衡完成
    运行中 --> 关闭中: close()
    关闭中 --> [*]: 资源释放完成
    
    note right of 初始化

        - 初始化网络客户端
        - 创建 Fetcher
        - 初始化 Coordinator
    end note
    
    note right of 再平衡中
        - 触发 onPartitionsRevoked()
        - 参与组协调器的 JoinGroup/SyncGroup
        - 触发 onPartitionsAssigned()
    end note

flowchart TB
    subgraph "Producer 架构"
        direction TB
        PA[KafkaProducer<br/>主线程入口]
        PA --> RA[RecordAccumulator<br/>消息批量累加器]
        PA --> PM[ProducerMetadata<br/>元数据管理]
        PA --> TM[TransactionManager<br/>事务管理]
        RA --> BP[BufferPool<br/>内存池]
        RA --> PB[ProducerBatch<br/>消息批次]
        
        ST[Sender Thread<br/>后台I/O线程] --> RA
        ST --> NC1[NetworkClient<br/>网络客户端]
        ST --> TM
        
        NC1 --> SocketChannel1[SocketChannel<br/>NIO通道]
        SocketChannel1 --> Broker1[Kafka Broker]
    end
    
    subgraph "Consumer 架构"
        direction TB
        CA[KafkaConsumer<br/>主线程入口]
        CA --> FE[Fetcher<br/>消息拉取器]
        CA --> CC[ConsumerCoordinator<br/>协调器客户端]
        CA --> SS[SubscriptionState<br/>订阅状态]
        
        FE --> FB[FetchBuffer<br/>拉取缓冲区]
        FE --> CF[CompletedFetch<br/>已完成拉取]
        CC --> HB[Heartbeat<br/>心跳管理]
        
        CA --> NC2[NetworkClient<br/>网络客户端]
        NC2 --> SocketChannel2[SocketChannel<br/>NIO通道]
        SocketChannel2 --> Broker2[Kafka Broker]
    end
    
    subgraph "AdminClient 架构"
        direction TB
        AC[KafkaAdminClient<br/>管理客户端]
        AC --> RQ[RequestQueue<br/>请求队列]
        AC --> NC3[NetworkClient<br/>网络客户端]
        
        RQ --> CR[CallRequest<br/>调用请求]
        CR --> NC3
        NC3 --> SocketChannel3[SocketChannel<br/>NIO通道]
        SocketChannel3 --> Broker3[Kafka Broker]
    end

public interface Serializer<T> extends Closeable {
    void configure(Map<String, ?> configs, boolean isKey);
    byte[] serialize(String topic, T data);
    byte[] serialize(String topic, Headers headers, T data);
    void close();
}

public interface Partitioner extends Configurable, Closeable {
    int partition(String topic, Object key, byte[] keyBytes, Object value,
                  byte[] valueBytes, Cluster cluster);
    void close();
    default void onNewBatch(String topic, Cluster cluster, int prevPartition) {}
}

public interface ProducerInterceptor<K, V> extends Configurable, Closeable {
    ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
    void onAcknowledgement(RecordMetadata metadata, Exception exception);
    void close();
}

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

flowchart TD
    Start[用户调用 send(record)] --> Serialize[序列化 key 和 value]
    Serialize --> Partition[选择分区<br/>通过 Partitioner]
    Partition --> Append[追加到 RecordAccumulator]
    Append --> TryAppend{尝试追加到现有批次}
    TryAppend -->|成功| Return1[返回 Future]
    TryAppend -->|失败| AllocBuffer[从 BufferPool 分配内存]
    AllocBuffer --> NewBatch[创建新批次 ProducerBatch]
    NewBatch --> AddBatch[加入队列]
    AddBatch --> Return2[返回 Future]
    
    Return1 --> WakeupSender[唤醒 Sender 线程]
    Return2 --> WakeupSender
    
    WakeupSender --> SenderRun[Sender 线程运行]
    SenderRun --> Ready{批次准备好?}
    Ready -->|是| DrainBatch[提取批次]
    Ready -->|否| Wait[等待 linger.ms]
    Wait --> SenderRun
    
    DrainBatch --> GroupByNode[按 Broker 分组]
    GroupByNode --> SendRequest[发送 ProduceRequest]
    SendRequest --> WaitResponse[等待响应]
    WaitResponse --> HandleResponse[处理响应]
    HandleResponse --> Callback[触发用户回调]
    Callback --> ReleaseBuf[释放 BufferPool 内存]
    ReleaseBuf --> End[完成]

flowchart TD
    Start[用户调用 poll&lpar;timeout&rpar;] --> UpdateMeta[更新元数据<br/>如果需要]
    UpdateMeta --> Heartbeat[发送心跳<br/>如果需要]
    Heartbeat --> CheckRebalance{需要再平衡?}
    CheckRebalance -->|是| Rebalance[执行再平衡]
    CheckRebalance -->|否| FetchCheck{FetchBuffer 有数据?}
    
    Rebalance --> FetchCheck
    
    FetchCheck -->|是| ReturnData[反序列化并返回消息]
    FetchCheck -->|否| SendFetch[发送 FetchRequest]
    
    SendFetch --> WaitResponse[等待响应或超时]
    WaitResponse --> HandleResponse[处理 FetchResponse]
    HandleResponse --> ParseRecords[解析消息记录]
    ParseRecords --> AddBuffer[加入 FetchBuffer]
    AddBuffer --> UpdatePos[更新消费位置]
    UpdatePos --> FetchCheck
    
    ReturnData --> CommitCheck{需要提交偏移量?}
    CommitCheck -->|自动提交| AutoCommit[异步提交偏移量]
    CommitCheck -->|手动提交| End1[完成]
    AutoCommit --> End2[完成]

batch.size=32768
linger.ms=10
buffer.memory=67108864
compression.type=lz4
max.in.flight.requests.per.connection=5

fetch.min.bytes=10240
fetch.max.wait.ms=500
max.poll.records=1000
enable.auto.commit=false

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        if (exception instanceof RetriableException) {
            // 客户端会自动重试，无需额外处理
        } else {
            // 不可重试异常，需要应用层处理
            log.error("发送失败: {}", exception.getMessage());
        }
    }
});

try {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
} catch (OffsetOutOfRangeException e) {
    // 偏移量越界，重置到最早
    consumer.seekToBeginning(e.partitions());
} catch (SerializationException e) {
    // 反序列化失败，跳过该消息
    consumer.seek(e.partition(), e.offset() + 1);
}

// 使用 Properties 配置
public KafkaProducer(Properties properties)

// 使用 Map 配置
public KafkaProducer(Map<String, Object> configs)

// 指定 Serializer
public KafkaProducer(Map<String, Object> configs,
                     Serializer<K> keySerializer,
                     Serializer<V> valueSerializer)

// 指定 Properties 和 Serializer
public KafkaProducer(Properties properties,
                     Serializer<K> keySerializer,
                     Serializer<V> valueSerializer)

// 示例 1：使用 Properties
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("enable.idempotence", true);

Producer<String, String> producer = new KafkaProducer<>(props);

// 示例 2：使用 Map 和指定 Serializer
Map<String, Object> configs = new HashMap<>();
configs.put("bootstrap.servers", "localhost:9092");
configs.put("acks", "all");

Producer<String, String> producer = new KafkaProducer<>(
    configs,
    new StringSerializer(),
    new StringSerializer()
);

// 示例 3：事务性 Producer
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id"); // 启用事务
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // 初始化事务

// 异步发送，无回调
public Future<RecordMetadata> send(ProducerRecord<K, V> record)

// 异步发送，带回调
public Future<RecordMetadata> send(ProducerRecord<K, V> record,
                                   Callback callback)

public class ProducerRecord<K, V> {
    private final String topic;        // Topic 名称
    private final Integer partition;   // 分区号（可选）
    private final Headers headers;     // 消息头（可选）
    private final K key;               // 消息 key（可选）
    private final V value;             // 消息 value
    private final Long timestamp;      // 时间戳（可选）
}

// 最简单：只指定 topic 和 value
ProducerRecord<String, String> record1 = new ProducerRecord<>("my-topic", "message");

// 指定 key 和 value
ProducerRecord<String, String> record2 = new ProducerRecord<>("my-topic", "key1", "value1");

// 指定 partition, key, value
ProducerRecord<String, String> record3 = new ProducerRecord<>("my-topic", 0, "key1", "value1");

// 完整参数：topic, partition, timestamp, key, value, headers
Headers headers = new RecordHeaders();
headers.add("header1", "value1".getBytes());
ProducerRecord<String, String> record4 = new ProducerRecord<>(
    "my-topic",
    0,                          // partition
    System.currentTimeMillis(), // timestamp
    "key1",                     // key
    "value1",                   // value
    headers                     // headers
);

public final class RecordMetadata {
    private final long offset;              // 消息偏移量
    private final long timestamp;           // 消息时间戳
    private final int serializedKeySize;    // 序列化后 key 大小
    private final int serializedValueSize;  // 序列化后 value 大小
    private final TopicPartition topicPartition; // Topic 和 Partition
}

// 方式 1：阻塞等待结果（不推荐）
RecordMetadata metadata = producer.send(record).get();
System.out.println("offset: " + metadata.offset());
System.out.println("partition: " + metadata.partition());

// 方式 2：异步回调（推荐）
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        System.err.println("发送失败: " + exception.getMessage());
    } else {
        System.out.println("offset: " + metadata.offset());
        System.out.println("partition: " + metadata.partition());
        System.out.println("timestamp: " + metadata.timestamp());
    }
});

// KafkaProducer.send() 核心代码
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // 1. 拦截器前置处理
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    
    return doSend(interceptedRecord, callback);
}

private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
    // 2. 创建回调包装器（包含拦截器和用户回调）
    AppendCallbacks appendCallbacks = new AppendCallbacks(callback, this.interceptors, record);

    try {
        // 3. 检查 Producer 状态
        throwIfProducerClosed();
        throwIfInPreparedState();

        // 4. 等待元数据可用（可能阻塞 max.block.ms）
        long nowMs = time.milliseconds();
        ClusterAndWaitTime clusterAndWaitTime = waitOnMetadata(
            record.topic(),
            record.partition(),
            nowMs,
            maxBlockTimeMs
        );
        Cluster cluster = clusterAndWaitTime.cluster;

        // 5. 序列化 key 和 value
        byte[] serializedKey = keySerializer.serialize(record.topic(), record.headers(), record.key());
        byte[] serializedValue = valueSerializer.serialize(record.topic(), record.headers(), record.value());

        // 6. 计算分区（如果用户未指定）
        int partition = partition(record, serializedKey, serializedValue, cluster);

        // 7. 校验消息大小
        int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(
            RecordBatch.CURRENT_MAGIC_VALUE, compression.type(),
            serializedKey, serializedValue, record.headers().toArray()
        );
        ensureValidRecordSize(serializedSize);

        // 8. 追加到 RecordAccumulator（批量累加器）
        RecordAccumulator.RecordAppendResult result = accumulator.append(
            record.topic(),
            partition,
            timestamp,
            serializedKey,
            serializedValue,
            record.headers(),
            appendCallbacks,
            remainingWaitMs,
            true
        );

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

        // 10. 返回 Future
        return result.future;
        
    } catch (Exception e) {
        // 11. 异常处理：触发回调，释放资源
        appendCallbacks.onCompletion(null, e);
        throw e;
    }
}

用户代码
  ↓
KafkaProducer.send(record, callback)
  ↓
KafkaProducer.doSend(record, callback)
  ↓
RecordAccumulator.append(...)
  ↓
ProducerBatch.tryAppend(...) // 尝试追加到现有批次
  或
RecordAccumulator.appendNewBatch(...) // 创建新批次
  ↓
BufferPool.allocate(...) // 分配内存（如果需要）
  ↓
Sender.wakeup() // 唤醒 Sender 线程
  ↓
Sender.run() // 后台线程发送
  ↓
NetworkClient.send(produceRequest)
  ↓
Broker 响应 ProduceResponse
  ↓
Sender 处理响应，触发 Callback

// Sender.run() 核心逻辑
void run() {
    while (running) {
        try {
            // 1. 提取准备好的批次
            long currentTimeMs = time.milliseconds();
            long pollTimeout = sendProducerData(currentTimeMs);
            
            // 2. 执行网络 I/O
            client.poll(pollTimeout, currentTimeMs);
            
        } catch (Exception e) {
            log.error("Uncaught error in kafka producer I/O thread", e);
        }
    }
}

private long sendProducerData(long now) {
    // 1. 获取准备好的批次（按 Broker 分组）
    ReadyCheckResult result = accumulator.ready(metadata.fetch(), now);
    
    // 2. 如果有不可用的分区 Leader，请求元数据更新
    if (!result.unknownLeaderTopics.isEmpty()) {
        for (String topic : result.unknownLeaderTopics) {
            metadata.add(topic, now);
        }
    }
    
    // 3. 移除未准备好的节点（网络不可用等）
    Iterator<Node> iter = result.readyNodes.iterator();
    while (iter.hasNext()) {
        Node node = iter.next();
        if (!client.ready(node, now)) {
            iter.remove();
        }
    }
    
    // 4. 从 RecordAccumulator 提取批次
    Map<Integer, List<ProducerBatch>> batches = accumulator.drain(
        metadata.fetch(),
        result.readyNodes,
        maxRequestSize,
        now
    );
    
    // 5. 处理过期批次
    List<ProducerBatch> expiredBatches = accumulator.expiredBatches(now);
    for (ProducerBatch expiredBatch : expiredBatches) {
        failBatch(expiredBatch, new TimeoutException(...));
    }
    
    // 6. 构建 ProduceRequest 并发送
    sendProduceRequests(batches, now);
    
    return pollTimeout;
}

sequenceDiagram
    autonumber
    participant U as 用户线程
    participant P as KafkaProducer
    participant RA as RecordAccumulator
    participant BP as BufferPool
    participant S as Sender线程
    participant NC as NetworkClient
    participant B as Broker

    U->>P: send(record, callback)
    P->>P: 序列化 key 和 value
    P->>P: 选择分区（Partitioner）
    P->>RA: append(topic, partition, ...)
    RA->>RA: 尝试追加到现有批次
    
    alt 批次已满或不存在
        RA->>BP: allocate(batchSize)
        BP-->>RA: ByteBuffer
        RA->>RA: 创建新 ProducerBatch
    end
    
    RA-->>P: RecordAppendResult(future)
    P->>S: wakeup() 唤醒 Sender 线程
    P-->>U: Future<RecordMetadata>
    
    Note over S: Sender 线程异步发送
    
    S->>RA: ready() 检查准备好的批次
    RA-->>S: ReadyCheckResult
    S->>RA: drain() 提取批次
    RA-->>S: Map<Node, List<ProducerBatch>>
    
    S->>S: 构建 ProduceRequest
    S->>NC: send(ProduceRequest)
    NC->>B: ProduceRequest（TCP）
    B->>B: 写入日志
    B-->>NC: ProduceResponse
    NC-->>S: ProduceResponse
    
    S->>S: 处理响应
    S->>RA: completeBatch() 完成批次
    S->>BP: deallocate() 释放内存
    S->>U: callback(metadata, null) 触发回调

producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        if (exception instanceof RetriableException) {
            // 可重试异常：Producer 会自动重试（最多 retries 次或 delivery.timeout.ms 超时）
            log.warn("发送失败，正在重试: {}", exception.getMessage());
        } else {
            // 不可重试异常：需要应用层处理
            log.error("发送失败，无法重试: {}", exception.getMessage());
            // 应用层决策：丢弃、记录到死信队列、人工介入等
        }
    }
});

# 高吞吐量配置
batch.size=32768
linger.ms=10
buffer.memory=67108864
compression.type=lz4
max.in.flight.requests.per.connection=5
acks=all
enable.idempotence=true

public void initTransactions()
public void initTransactions(boolean keepPreparedTxn)

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // 必须在第一次事务前调用

public void initTransactions() {
    // 1. 检查是否配置了 transactional.id
    if (transactionManager == null) {
        throw new IllegalStateException("无法初始化事务：未配置 transactional.id");
    }
    
    // 2. 确保 Sender 线程已启动
    throwIfProducerClosed();
    
    // 3. 调用 TransactionManager 初始化
    TransactionalRequestResult result = transactionManager.initializeTransactions();
    sender.wakeup();
    
    // 4. 阻塞等待初始化完成（发送 InitProducerIdRequest）
    result.await();
}

sequenceDiagram
    autonumber
    participant P as Producer
    participant TM as TransactionManager
    participant S as Sender
    participant TC as TransactionCoordinator

    P->>TM: initializeTransactions()
    TM->>TM: 状态转换: UNINITIALIZED -> INITIALIZING
    TM-->>P: TransactionalRequestResult
    P->>S: wakeup()
    
    S->>TM: maybeAddInitPidRequest()
    TM->>S: InitProducerIdRequest
    S->>TC: InitProducerIdRequest(transactional.id)
    TC->>TC: 分配 ProducerId & Epoch
    TC->>TC: 写入 __transaction_state topic
    TC-->>S: InitProducerIdResponse(producerId, epoch)
    S->>TM: handleResponse()
    TM->>TM: 状态转换: INITIALIZING -> READY
    TM-->>P: 初始化完成

public void beginTransaction() throws ProducerFencedException

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    // Producer 被 fence，需要关闭并创建新实例
    producer.close();
} catch (KafkaException e) {
    // 其他异常，回滚事务
    producer.abortTransaction();
}

public void beginTransaction() throws ProducerFencedException {
    throwIfNoTransactionManager();
    
    // 状态检查：必须在 READY 状态
    transactionManager.beginTransaction();
    // 状态转换: READY -> IN_TRANSACTION
}

public void commitTransaction() throws ProducerFencedException

sequenceDiagram
    autonumber
    participant P as Producer
    participant TM as TransactionManager
    participant S as Sender
    participant TC as TransactionCoordinator
    participant B as Broker

    P->>TM: commitTransaction()
    TM->>TM: 状态转换: IN_TRANSACTION -> COMMITTING_TRANSACTION
    
    S->>TC: EndTxnRequest(commit=true)
    TC->>TC: 状态转换: Ongoing -> PrepareCommit
    TC->>TC: 写入 __transaction_state
    TC->>B: WriteTxnMarkersRequest（所有参与分区）
    B->>B: 写入 COMMIT Marker
    B-->>TC: WriteTxnMarkersResponse
    TC->>TC: 状态转换: PrepareCommit -> CompleteCommit
    TC->>TC: 清理事务状态
    TC-->>S: EndTxnResponse
    S->>TM: 状态转换: COMMITTING_TRANSACTION -> READY
    TM-->>P: 提交完成

public void abortTransaction() throws ProducerFencedException

try {
    producer.beginTransaction();
    producer.send(record1);
    // 业务逻辑抛出异常
    throw new BusinessException();
    producer.send(record2);
    producer.commitTransaction();
} catch (BusinessException e) {
    // 回滚事务
    producer.abortTransaction();
}

public void flush()

for (int i = 0; i < 1000; i++) {
    producer.send(new ProducerRecord<>("my-topic", "key-" + i, "value-" + i));
}
producer.flush(); // 阻塞直到所有 1000 条消息发送完成

public List<PartitionInfo> partitionsFor(String topic)

public class PartitionInfo {
    private final String topic;
    private final int partition;
    private final Node leader;      // Leader 副本所在节点
    private final Node[] replicas;  // 所有副本节点
    private final Node[] inSyncReplicas; // ISR 副本节点
}

public Map<MetricName, ? extends Metric> metrics()

public void close()
public void close(Duration timeout)

try (Producer<String, String> producer = new KafkaProducer<>(props)) {
    producer.send(record);
} // 自动调用 close()

// 使用 Properties 配置
public KafkaConsumer(Properties properties)

// 使用 Map 配置
public KafkaConsumer(Map<String, Object> configs)

// 指定 Deserializer
public KafkaConsumer(Map<String, Object> configs,
                     Deserializer<K> keyDeserializer,
                     Deserializer<V> valueDeserializer)

// 指定 Properties 和 Deserializer
public KafkaConsumer(Properties properties,
                     Deserializer<K> keyDeserializer,
                     Deserializer<V> valueDeserializer)

// 示例 1：基本配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");

Consumer<String, String> consumer = new KafkaConsumer<>(props);

// 示例 2：手动提交配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("enable.auto.commit", "false"); // 手动提交
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Consumer<String, String> consumer = new KafkaConsumer<>(props);

// 示例 3：事务性消费（只读已提交事务）
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("isolation.level", "read_committed"); // 只读已提交事务
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Consumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅主题列表
public void subscribe(Collection<String> topics)

// 订阅主题列表，带再平衡监听器
public void subscribe(Collection<String> topics,
                     ConsumerRebalanceListener listener)

// 订阅匹配模式的主题
public void subscribe(Pattern pattern)

// 订阅匹配模式的主题，带再平衡监听器
public void subscribe(Pattern pattern,
                     ConsumerRebalanceListener listener)

// 订阅单个主题
consumer.subscribe(Collections.singletonList("my-topic"));

// 订阅多个主题
consumer.subscribe(Arrays.asList("topic1", "topic2", "topic3"));

// 订阅匹配模式的主题（如所有以 "log-" 开头的主题）
consumer.subscribe(Pattern.compile("log-.*"));

// 订阅主题并注册再平衡监听器
consumer.subscribe(Collections.singletonList("my-topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 分区被撤销时调用（再平衡前）
        System.out.println("撤销分区: " + partitions);
        // 通常在这里提交偏移量或保存状态
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 分区被分配时调用（再平衡后）
        System.out.println("分配分区: " + partitions);
        // 通常在这里重置偏移量或恢复状态
    }
});

public ConsumerRecords<K, V> poll(Duration timeout)

public class ConsumerRecords<K, V> implements Iterable<ConsumerRecord<K, V>> {
    // 按 Topic-Partition 组织的消息记录
    private final Map<TopicPartition, List<ConsumerRecord<K, V>>> records;
}

public class ConsumerRecord<K, V> {
    private final String topic;           // Topic 名称
    private final int partition;          // 分区号
    private final long offset;            // 偏移量
    private final long timestamp;         // 时间戳
    private final TimestampType timestampType; // 时间戳类型
    private final int serializedKeySize;  // 序列化后 key 大小
    private final int serializedValueSize; // 序列化后 value 大小
    private final Headers headers;        // 消息头
    private final K key;                  // 消息 key
    private final V value;                // 消息 value
    private final Optional<Integer> leaderEpoch; // Leader epoch
}

// KafkaConsumer.poll() 核心代码
public ConsumerRecords<K, V> poll(final Duration timeout) {
    return poll(time.timer(timeout));
}

private ConsumerRecords<K, V> poll(final Timer timer) {
    acquireAndEnsureOpen(); // 获取锁，确保线程安全
    try {
        // 1. 检查订阅状态
        if (subscriptions.hasNoSubscriptionOrUserAssignment()) {
            throw new IllegalStateException("Consumer未订阅任何主题或分配任何分区");
        }

        do {
            // 2. 触发可能的唤醒
            client.maybeTriggerWakeup();

            // 3. 更新分配（可能触发再平衡）
            updateAssignmentMetadataIfNeeded(timer, false);

            // 4. 拉取消息
            final Fetch<K, V> fetch = pollForFetches(timer);
            
            if (!fetch.isEmpty()) {
                // 5. 在返回前发送下一轮拉取请求（流水线优化）
                if (sendFetches() > 0 || client.hasPendingRequests()) {
                    client.transmitSends();
                }

                // 6. 返回拉取到的记录
                return this.interceptors.onConsume(
                    new ConsumerRecords<>(fetch.records(), fetch.nextOffsets())
                );
            }
        } while (timer.notExpired());

        return ConsumerRecords.empty();
    } finally {
        release(); // 释放锁
    }
}

private Fetch<K, V> pollForFetches(Timer timer) {
    long pollTimeout = Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()),
                                timer.remainingMs());

    // 1. 从 FetchBuffer 获取已拉取的消息
    final Fetch<K, V> fetch = fetcher.collectFetch();
    if (!fetch.isEmpty()) {
        return fetch;
    }

    // 2. 如果 FetchBuffer 为空，发送 FetchRequest
    if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {
        client.poll(pollTimeout, timer.currentTimeMs());
    }

    // 3. 再次尝试从 FetchBuffer 获取
    return fetcher.collectFetch();
}

// 基本用法
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic=%s, partition=%d, offset=%d, key=%s, value=%s%n",
            record.topic(), record.partition(), record.offset(),
            record.key(), record.value());
        
        // 处理消息
        processRecord(record);
    }
    
    // 手动提交偏移量（如果禁用了自动提交）
    consumer.commitSync();
}

// 按分区处理
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (TopicPartition partition : records.partitions()) {
    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    for (ConsumerRecord<String, String> record : partitionRecords) {
        processRecord(record);
    }
    // 按分区提交偏移量
    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
    consumer.commitSync(Collections.singletonMap(partition,
        new OffsetAndMetadata(lastOffset + 1)));
}