消息丢失与可靠性保证#

Producer ──发送──→ Broker ──存储──→ Consumer
   │                │                 │
   ├── 网络丢失     ├── 宕机丢失      ├── 自动提交后处理失败
   ├── 超时未重试   ├── 副本不足      └── offset 跳过
   └── 序列化失败   └── 刷盘前崩溃

如果 ISR 中只有 Leader 一个副本（其他副本被踢出 ISR）
→ acks=all 效果等于 acks=1
→ Leader 宕机仍会丢消息

解决：配合 min.insync.replicas=2
→ ISR 副本数 < 2 时，acks=all 的写入直接报错（NotEnoughReplicas），acks=0/1 不受此参数约束
→ 用可用性换可靠性

# Producer 端（Kafka 3.0 起 acks=all、enable.idempotence=true 已是默认值）
acks = all
enable.idempotence = true            # 幂等生产者：重试不重复，且 in.flight ≤ 5 时仍保证单分区有序
retries = 2147483647                 # 默认值即 Integer.MAX_VALUE，实际重试时长由 delivery.timeout.ms（默认 120s）限制
retry.backoff.ms = 100               # 重试间隔
max.in.flight.requests.per.connection = 5   # 开幂等后不必再设 1；没开幂等时要设 1 才能防重试乱序

# Broker 端
min.insync.replicas = 2              # ISR 至少 2 个副本
replication.factor = 3               # 每个 Partition 3 个副本

Producer → Leader Broker (写入 Page Cache) → 操作系统异步刷盘
                         ↓
                 Follower 从 Leader 拉取复制

Consumer 拉取消息 → 处理失败但异常被吞掉 / 交给异步线程处理
→ 下次 poll() 时自动提交了这批 offset → 消息不会再投递 → 消息"丢失"（实际是跳过了）
（同步处理完再 poll，自动提交也是 at-least-once；丢消息出在「提交早于处理完成」）

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    // 1. 先处理业务逻辑
    processMessage(record);
    // 2. 处理成功后再提交这一条的 offset（提交的是「下一条要读的位置」= offset + 1）
    consumer.commitSync(Map.of(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)));
}
// 注意：无参 commitSync() 提交的是本次 poll 返回的全部位点，
// 放在循环里处理完第 1 条就会把整批都标成已消费，后面几条处理失败就丢了。
// 按批提交时应在整个 for 循环结束后再调用无参 commitSync()。

Producer:
  ✅ acks = all
  ✅ retries 保持默认（Integer.MAX_VALUE，受 delivery.timeout.ms 限制）
  ✅ enable.idempotence = true（防止重试导致的重复消息；3.0 起默认开启）

Broker:
  ✅ replication.factor = 3
  ✅ min.insync.replicas = 2
  ✅ unclean.leader.election.enable = false（不允许非 ISR 副本成为 Leader）

Consumer:
  ✅ enable.auto.commit = false
  ✅ 手动 commitSync / commitAsync + callback
  ✅ 消费逻辑做幂等