Kafka 可靠性与事务：从副本、ISR 到 Exactly Once 的 Failure Model#

深入 acks、High Watermark、ELR、幂等 Producer、Transaction Coordinator 与端到端一致性边界

阅读目标#

前四篇已经完成 Kafka 核心数据链的搭建：

1
第一篇
2
Log → Partition → Offset → Consumer Position
3

4
第二篇
5
RecordBatch → Segment → Index → Page Cache → Broker Data Path
6

7
第三篇
8
KafkaProducer.send()
9
→ RecordAccumulator
10
→ Sender
11
→ NetworkClient
12
→ Partition Leader
13

14
第四篇
15
poll()
16
→ Fetch
17
→ Consumer Group
18
→ Rebalance
19
→ Offset Commit

第五篇要面对一个更现实的问题：

如果链路中的任何节点都可能失败，Kafka 如何定义一条消息究竟是成功、失败、可见、可恢复，还是需要重试？

真实链路中可能发生：

1
Producer 重复调用 send
2
Request 没有到达 Broker
3
Leader 已追加但 ACK 丢失
4
Follower 落后或掉出 ISR
5
Leader 在返回成功后永久故障
6
旧 Leader 在网络分区后恢复
7
Consumer 读到未完成事务的数据
8
业务数据库成功，但 Offset Commit 失败

完成本篇后，你应该能够：

从 Failure Boundary 分析消息丢失、重复与乱序；
准确解释 acks=0、acks=1、acks=all；
说明 Replication Factor、Leader、Follower 和 Replica Fetch 的关系；
区分 Replica Set、ISR、ELR、HW、LEO 和 LSO；
解释 min.insync.replicas 与 acks=all 如何共同工作；
解释 Kafka 4.x ELR 与 Unclean Leader Election 的差异；
说明 PID、Producer Epoch 与 Sequence Number 如何实现 Broker 端去重；
区分 Idempotent Producer 与 Kafka Transaction；
画出 Transaction Coordinator 与 __transaction_state；
解释 Commit/Abort Marker、LSO 与 read_committed；
实现 Kafka → Process → Kafka 的事务处理；
明确 Exactly Once 在 Kafka、数据库和 HTTP 边界上的差异；
根据副本、Producer 和 Transaction 指标排查故障；
沿 ReplicaManager、Partition、ProducerStateManager 与 Transaction Coordinator 阅读源码。

本文以 Apache Kafka 4.3.x 为基线。Kafka 4.3 Producer 默认 acks=all，在不存在冲突配置时默认启用 Idempotence；启用 Idempotence 要求 acks=all、retries>0 且 max.in.flight.requests.per.connection<=5。Kafka 4.x 还引入并继续完善 Eligible Leader Replicas（ELR），用于在严格 min.insync.replicas 语义下改进安全选举。

0. Broker 已经写入，但 ACK 丢了：成功了吗#

1
Producer
2
    ↓ ProduceRequest(A)
3
Partition Leader
4
    ↓ A 已追加到 Log
5
ProduceResponse
6
    ↓ ACK 在网络中丢失
7
Producer Timeout

ACK 丢失后的不确定结果

Producer 无法区分：

1
Request 根本没有到达 Broker
2
Broker 收到但尚未追加
3
Broker 已追加，ACK 在回程丢失
4
Broker 已追加并复制，但连接中断

如果不重试，第一种情况会丢消息；如果重试，第三种情况可能重复。

这说明：

分布式发送最难的不是服务器明确返回失败，而是客户端无法判断服务器执行到了哪一步。

Idempotent Producer 没有让网络变可靠。它解决的是：

当 Producer 因不确定结果重试同一个 Batch 时，Broker 能否识别它已经追加过。

1. 一条消息的 Failure Boundary#

消息链路中的 Failure Boundary

“Kafka 会不会丢消息”不是一个足够精确的问题。至少要拆成七段。

1.1 Application Boundary#

应用可能因为 HTTP 重试、任务重放或代码错误重复调用：

1
producer.send(event);

Kafka Idempotence 无法理解两个不同 send() 在业务上是不是同一事件。业务层仍需要：

1
Event ID
2
Request ID
3
订单号 + 事件类型
4
业务状态机
5
幂等记录

1.2 Producer Boundary#

Record 可能仍处于：

1
序列化阶段
2
RecordAccumulator
3
等待 Buffer
4
In-flight Request
5
Retry Backoff
6
Delivery Timeout

send() 返回 Future 不等于消息已经进入 Broker。

1.3 Network Boundary#

可能发生：

1
Request 丢失
2
Response 丢失
3
连接中断
4
Broker 成功但客户端未知

1.4 Leader Boundary#

Leader 负责：

1
校验 RecordBatch
2
分配最终 Offset
3
追加本地 Log
4
校验 Producer Sequence
5
等待复制条件
6
返回 ProduceResponse

必须区分：

1
Leader 已追加
2
形成安全复制条件
3
Producer 已收到成功响应

1.5 Replica Boundary#

Follower 可能：

1
在线且同步
2
在线但落后
3
网络不可达
4
磁盘过慢
5
掉出 ISR

三副本不等于当前消息一定已经在三台机器上安全存在。

1.6 Consumer Boundary#

Consumer 是否可见受到：

1
High Watermark
2
Isolation Level
3
Last Stable Offset
4
Position

限制。

1.7 External Side Effect Boundary#

消费后可能：

1
更新 MySQL
2
调用支付 API
3
修改库存
4
执行 Agent Tool

这些副作用不属于 Kafka Log，Kafka Transaction 不能自动覆盖它们。

因此排障时固定问：

1
失败发生在哪一段？
2
这一段的 Source of Truth 是什么？
3
调用方能否确定执行结果？
4
Retry 是否会产生重复？
5
恢复状态由谁保存？

2. `acks`：Producer 何时认为写入完成#

acks 配置的确认边界

acks 定义 Producer 等待的 Broker 确认条件。它不是可靠性的唯一开关，可靠性还取决于：

1
Replication Factor
2
ISR
3
min.insync.replicas
4
Leader Election
5
Retry
6
Idempotence
7
故障域分布
8
应用是否检查 Future

2.1 `acks=0`#

1
acks=0

Producer 不等待 Broker Response。消息进入 Socket Buffer 后就认为已发送。

结果：

1
Broker 是否收到：未知
2
是否被拒绝：未知
3
Retry 无法依赖 Broker 错误
4
RecordMetadata.offset=-1

只适合可容忍丢失的低价值遥测。

2.2 `acks=1`#

1
acks=1

Leader 本地追加后立即返回，不等待 Follower。

1
Leader 追加 A
2
→ 返回成功
3
→ Follower 尚未复制
4
→ Leader 永久故障
5
→ 新 Leader 没有 A

Producer 已经看到成功，但 Record 可能丢失。

2.3 `acks=all`#

1
acks=all

等价于 -1。Leader 等待当前 ISR 全部确认，并检查 min.insync.replicas。

Kafka 4.3 Producer 默认使用 acks=all。

2.4 `acks=all` 不是固定等待两个副本#

假设：

1
RF=3
2
ISR={1,2,3}
3
min ISR=2

acks=all 等待 1、2、3，而不是只等两个。

如果：

1
ISR={1,2}

等待 1、2。

如果：

1
ISR={1}

因为不满足 min ISR=2，写入被拒绝。

2.5 成功响应不等于每块磁盘都同步 fsync#

需要区分：

1
Producer Response Success
2
Consumer Visibility
3
Physical Device Flush

Kafka 的核心可靠性依赖副本协议、确认条件和选举约束，不是每条消息都同步刷所有物理磁盘。

3. Kafka 副本模型：Leader 与 Follower#

一个 Partition 的：

1
Replication Factor=N

表示它配置了 N 个 Replica，其中一个 Leader，其余 Follower。

Follower 主动 Fetch Leader

3.1 Producer 只写 Leader#

所有写入经 Leader 排序并分配 Offset，保证 Partition 只有一个合法追加顺序。

3.2 Follower 主动 Fetch#

1
Follower LEO
2
    ↓ FetchRequest(offset=LEO)
3
Leader
4
    ↓ 返回 RecordBatch
5
Follower 追加本地 Log
6
    ↓ 再次 Fetch

使用 Pull 的好处：

每个 Follower 根据自身进度拉取；
支持批量复制；
慢 Follower 不直接占用 Producer 线程；
Leader 可以依据 Fetch 进度判断同步状态。

3.3 Follower 不是独立可写副本#

Kafka 不是多主写。Follower 必须收敛到当前 Leader Epoch 对应的合法历史。

Leader 切换后，旧副本可能需要：

1
比较 Leader Epoch
2
定位 Diverging Offset
3
截断不合法尾部
4
重新 Fetch

副本系统的目标不是保留每台 Broker 曾经写过的所有尾部，而是形成唯一合法 Partition History。

4. ISR：当前安全同步集合#

ISR 是：

1
In-Sync Replicas

包括 Leader 和当前满足同步条件的 Followers。

ISR、HW 与 LEO

4.1 ISR 是动态集合#

1
Replicas=[1,2,3]
2
ISR=[1,2,3]

如果 Broker 3 长时间不 Fetch 或无法追上：

1
Replicas=[1,2,3]
2
ISR=[1,2]

恢复并追上后可重新进入 ISR。

4.2 同步不是零延迟#

Follower 不必每个瞬间与 Leader Offset 完全相同。Kafka 关注它是否在限定时间内持续 Fetch 和追赶。

Kafka 4.3 默认：

1
replica.lag.time.max.ms=30000

长时间不 Fetch 或不能追上会被移出 ISR。

4.3 ISR 参与哪些机制#

1
acks=all 的确认范围
2
min ISR 写入门槛
3
High Watermark 推进
4
Leader Election
5
故障告警

4.4 Replica Set、ISR 与 ELR#

1
Replica Set
2
→ 配置的所有副本
3

4
ISR
5
→ 当前同步集合
6

7
ELR
8
→ 已不在 ISR，但仍包含全部安全提交历史、
9
   可以安全参与选举的 Eligible Replicas

5. High Watermark、LEO 与可见性#

5.1 Log End Offset#

每个 Replica 都有自己的 LEO：

1
Leader LEO=12
2
Follower A LEO=10
3
Follower B LEO=8

表示本地下一写入位置。

5.2 High Watermark#

HW 是 Consumer 可安全读取的复制提交边界。

Leader 本地 Offset 8～11 即使存在，也可能还不能返回给 Consumer。否则 Consumer 读到了 11，而故障后新 Leader 只有 8，就出现已经被观察过的数据从合法历史消失。

5.3 HW 不能只背成 `min(ISR LEO)`#

可用近似模型理解 HW 与最慢同步副本有关，但实现还涉及：

1
Follower Fetch 确认轮次
2
Leader Epoch
3
ISR 变化
4
严格 min ISR 语义
5
ELR

不要把简化公式当成完整实现。

5.4 为什么需要下一轮 Fetch#

Follower 请求 offset=5，Leader 返回到 10。Leader不会立即收到独立的磁盘 ACK。

Follower 下一次发：

1
Fetch(offset=10)

Leader 才知道它已经复制到 10。因此 HW 推进天然包含 Replica Fetch 协议轮次。

5.5 事务下的读取边界#

1
isolation.level=read_uncommitted

通常读到 HW。

1
isolation.level=read_committed

还受 LSO 限制，并跳过 Abort Transaction 数据。

6. `min.insync.replicas`：持久性与可用性的选择#

min ISR 的可用性与持久性权衡

假设：

1
RF=3
2
min ISR=2

6.1 ISR={1,2,3}#

满足写入条件，acks=all 等待三个当前 ISR。

6.2 ISR={1,2}#

仍满足 min ISR=2，继续写入，等待两个 ISR。

6.3 ISR={1}#

写入被拒绝，Producer 可能收到：

1
NotEnoughReplicasException
2
NotEnoughReplicasAfterAppendException

6.4 为什么拒绝写入#

只剩 Leader 仍继续确认：

1
唯一 Leader 写入
2
→ Producer 成功
3
→ Leader 永久丢失
4
→ 无副本可恢复

拒绝写入是在明确表达：

当前集群无法继续满足声明的持久性级别。

6.5 权衡#

提高 min ISR：

1
+ 已确认数据更难丢失
2
- 故障时更早停止写入

降低：

1
+ 故障期间更容易保持可写
2
- 可安全恢复副本更少

常见关键业务组合：

1
RF=3
2
min ISR=2
3
acks=all
4
unclean election disabled
5
Rack Awareness

7. Leader Election：ISR、ELR 与 Unclean#

Kafka 4.x Leader Election 候选层级

7.1 优先 ISR#

ISR 中的 Replica 包含安全提交历史，因此首先从 ISR 选择 Leader。

7.2 ELR 为什么出现#

严格 min ISR 下，当 ISR 数低于门槛：

1
HW 停止推进
2
新写入无法形成成功确认

此时某些 Replica 即使被移出 ISR，仍可能包含全部已经提交的数据。Kafka 4.0 引入 ELR，Kafka 4.1 起新集群默认启用，用于记录这些仍可安全选举的 Replica。

Kafka 4.3 的候选顺序可以概括为：

1
ISR 非空 → 从 ISR 选择
2
否则 ELR 非空 → 从未 Fenced ELR 选择
3
否则考虑未 Fenced 的 Last Known Leader

7.3 ELR 与 Unclean 的根本差异#

1
ELR
2
→ 虽不在 ISR，但仍被证明包含安全提交历史
3

4
Unclean
5
→ 可能缺失已确认数据

ELR 目标是在安全前提下改善可用性；Unclean 用数据风险换可用性。

7.4 Unclean Leader Election#

1
unclean.leader.election.enable=true

假设：

1
旧 Leader: 0..100
2
Follower: 0..80

Follower 成为新 Leader 后，81～100 会从合法历史消失。

关键交易通常不应开启 Unclean Election。

7.5 Leader Epoch 与旧 Leader Fencing#

网络分区后，旧 Leader 可能仍认为自己是 Leader。KRaft Controller 通过 Leader Epoch 和 Metadata 隔离旧 Leader。

旧 Broker 恢复后必须：

1
确认新 Epoch
2
截断不合法尾部
3
作为 Follower 重新追赶

8. Retry、重复与乱序#

Retry 引发的重复与乱序

8.1 ACK 丢失导致重复#

1
A 已追加
2
ACK 丢失
3
Producer Retry A

没有去重状态时，Broker 无法判断它是新 Batch 还是旧 Batch 重试。

8.2 多 In-flight 导致乱序#

同一 Partition：

1
A 失败等待重试
2
B 后发先成功
3
A 重试成功

可能写成：

1
B → A

Kafka 4.3 官方配置明确说明：禁用 Idempotence、允许 Retry 且 max.in.flight>1 时可能重排。

8.3 把 In-flight 设为 1#

能避免上述重排，但降低连接 Pipeline 吞吐。

Idempotence 允许 max.in.flight<=5 仍保持顺序，因为 Broker 能校验 Sequence Number。

8.4 应用级重发仍可能重复#

即使 Idempotence 开启：

1
catch (Exception e) {
2
    producer.send(rebuildEvent());
3
}

第二次调用可能成为新的业务发送，不再是原 Batch Retry。最终失败后是否重发，需要业务 Event ID 和状态查询判断。

9. Idempotent Producer：PID、Epoch 与 Sequence#

Idempotent Producer 的去重状态

Kafka 4.3 在没有冲突配置时默认：

1
enable.idempotence=true

依赖：

1
acks=all
2
retries>0
3
max.in.flight.requests.per.connection<=5

9.1 Producer ID#

Broker 为 Producer 分配 PID，RecordBatch Header 携带 PID。

9.2 Producer Epoch#

PID 对应 Epoch。新会话取代旧会话时 Epoch 增加，旧 Epoch 会被识别为 Stale。

Transactional Producer 使用同一 transactional.id 重启时，新实例可以 Fence 旧实例。

9.3 Sequence Number#

Producer 为每个：

1
PID + TopicPartition

维护独立 Sequence。

1
P0: 0,1,2...
2
P1: 0,1...

9.4 Broker 的判断#

期望下一批 Seq=3：

1
收到 3 → 正常追加
2
收到 2 → 可能是已成功 Batch 的 Retry
3
收到 5 → 中间存在 Gap，状态异常
4
旧 Epoch → Stale / Fenced

9.5 去重作用域#

Idempotence 能解决：

1
同一 Producer Session
2
同一 TopicPartition
3
同一 Batch 的 Broker Retry

不能解决：

1
两个 Producer 发送相同 Event
2
应用重启后重新构造事件
3
Consumer 重复更新数据库
4
人工 Replay
5
HTTP 上游重复请求

9.6 为什么要求 `acks=all`#

如果使用 acks=1，Leader 确认 Sequence 后永久故障，而 Follower 未复制 Batch 和 Producer State，新 Leader 无法可靠延续去重状态。

9.7 Broker 状态恢复#

源码重点：

1
ProducerStateManager
2
ProducerAppendInfo
3
BatchMetadata
4
VerificationStateEntry

负责 Sequence 校验、Epoch、Duplicate Detection、Transaction 状态和 Snapshot 恢复。

10. Idempotence 为什么不等于 Transaction#

幂等和事务的边界

Idempotence 解决单 Partition Retry 去重。

业务可能需要：

1
P0: OrderCreated
2
P1: InventoryReserved

如果 P0 成功、P1 失败，即使两个写入各自幂等，业务仍是部分成功。

Kafka Transaction 提供：

1
多个 Topic
2
多个 Partition
3
Consumer Offset

的原子可见性。

10.1 事务不是磁盘回滚#

事务 Record 会先追加进普通 Log。Abort 时不物理删除，而是使用：

1
Transaction Metadata
2
Abort Marker
3
Consumer Isolation

让 read_committed Consumer 跳过。

Kafka 事务控制的是原子可见性，不是 Undo 写入。

11. Transactional Producer API#

配置：

1
transactional.id=fulfillment-worker-01

设置后自动启用 Idempotence。

11.1 `initTransactions()`#

1
producer.initTransactions();

它会：

查找 Transaction Coordinator；
完成或 Abort 同 ID 的旧事务；
获取 PID 与 Epoch；
Fence 旧 Producer 实例。

11.2 `beginTransaction()`#

1
producer.beginTransaction();

一个 Producer 同时只能有一个 Open Transaction。

11.3 `send()`#

beginTransaction() 和 Commit/Abort 之间的所有发送都属于当前事务。

11.4 `sendOffsetsToTransaction()`#

1
producer.sendOffsetsToTransaction(
2
    records.nextOffsets(),
3
    consumer.groupMetadata()
4
);

这些 Offset 只有在事务最终 Commit 后才算已提交。

11.5 `commitTransaction()`#

会先 Flush 事务内 Record，再提交。事务中任一不可恢复 Send 失败都会导致 Commit 失败。

11.6 `abortTransaction()`#

将当前事务标记为 Abort，read_committed 不返回其中的业务 Record。

11.7 Transactional ID 设计#

应该稳定且唯一映射一个有状态实例或 Shard：

1
application + task/shard identity

多个活跃实例共享 ID 会互相 Fence；每次完全随机又无法安全接管旧事务。

12. Transaction Coordinator 与状态机#

Transaction Coordinator 状态机

12.1 Coordinator 定位#

1
hash(transactional.id)
2
→ __transaction_state Partition
3
→ 该 Partition Leader
4
→ Transaction Coordinator

Kafka 4.3 默认：

1
transaction.state.log.num.partitions=50
2
transaction.state.log.replication.factor=3
3
transaction.state.log.min.isr=2

12.2 Transaction Metadata#

概念上包括：

1
transactional.id
2
PID
3
Producer Epoch
4
Transaction Timeout
5
State
6
参与 TopicPartitions
7
Start Timestamp

12.3 状态机#

1
Empty
2
→ Ongoing
3
  ├→ PrepareCommit → CompleteCommit
4
  └→ PrepareAbort  → CompleteAbort

完成过程：

持久化即将 Commit/Abort 的决定；
向参与 Partition 写 Control Marker；
所有 Marker 完成后写最终状态。

12.4 AddPartitionsToTxn#

Producer 首次向某 Partition 写事务数据时，需要将其加入当前事务参与集合，因为 Coordinator 最终必须知道向哪些 Partition 写 Marker。

12.5 Transaction Timeout#

Broker 默认上限：

1
transaction.max.timeout.ms=900000

长事务会：

1
阻挡 LSO
2
延迟 read_committed
3
增加恢复成本
4
占用 Transaction State

Kafka Transaction 不适合包裹分钟级人工审批或长时间 Agent Workflow。

13. Commit/Abort Marker、LSO 与隔离读取#

事务标记、LSO 与 Consumer 隔离

13.1 事务数据先追加#

事务 Record 已占用正常 Offset。

13.2 Control Marker#

Commit 时写：

1
COMMIT Marker

Abort 时写：

1
ABORT Marker

Control Batch 在 Log 中存在，但普通 Consumer 不把它作为业务 Record 返回。

13.3 Last Stable Offset#

LSO 可以理解为最早未完成事务之前的稳定读取边界。

即使 HW 已到 20，最早 Open Transaction 从 4 开始，read_committed 仍可能停在 4 之前。

13.4 Abort Transaction Index#

Broker 使用 Transaction Index 等结构记录中止事务范围。read_committed Fetch 会跳过这些 Record。

13.5 Open Transaction 造成的 Lag#

如果：

1
HW/LEO 持续前进
2
read_committed Position 停滞

可能不是 Consumer 慢，而是 LSO 被长事务阻塞。

14. Kafka → Process → Kafka 的 Exactly Once#

Kafka Read-Process-Write 的事务边界

不用事务时：

1
输出成功
2
Offset Commit 失败
3
→ 重启后输出重复

或者：

1
Offset Commit 成功
2
输出失败
3
→ 输入被跳过

14.1 输出和 Offset 进入同一事务#

1
producer.beginTransaction();
2

3
for (ConsumerRecord<String, String> record : records) {
4
    producer.send(transform(record));
5
}
6

7
producer.sendOffsetsToTransaction(
8
    records.nextOffsets(),
9
    consumer.groupMetadata()
10
);
11

12
producer.commitTransaction();

失败：

1
producer.abortTransaction();

输出不可见，输入 Offset 不推进。

14.2 下游 Consumer#

1
isolation.level=read_committed

否则可能看到未 Commit 数据。

14.3 Rebalance 与 Fencing#

如果处理期间失去 Partition，sendOffsetsToTransaction() 可能因 Group Metadata 失效而失败。应用应 Abort，重新加入组并从 Committed Offset 恢复。

15. Exactly Once 的边界#

Exactly Once 在不同系统边界上的机制

“Kafka 支持 Exactly Once”必须继续问：

1
在哪个边界？
2
对哪个副作用？
3
由谁观察？

15.1 Producer → Kafka#

Idempotent Producer 保证 Broker Retry 不会在单 Partition Log 中追加重复 Batch。

15.2 Kafka 多 Partition#

Kafka Transaction 提供多个 TopicPartition 原子可见性。

15.3 Kafka Input → Kafka Output#

1
Transaction
2
+ sendOffsetsToTransaction
3
+ read_committed

实现 Kafka 内部 Read-Process-Write EOS。

15.4 Kafka → MySQL#

Kafka Transaction 无法原子提交 MySQL Local Transaction。

1
MySQL Commit 成功
2
进程崩溃
3
Kafka Offset 未 Commit
4
→ 重复消费

需要：

1
Event ID 唯一约束
2
条件更新
3
Inbox Pattern
4
业务幂等

业务 DB 更新后发布 Kafka Event，通常使用：

1
Transactional Outbox
2
+ CDC / Reliable Publisher

15.5 Kafka → HTTP / Payment#

需要下游支持：

1
Idempotency-Key
2
Request ID
3
状态查询
4
业务状态机
5
补偿与对账

15.6 Effectively Once#

跨系统无法建立全局事务时，常见目标：

1
At-Least-Once Delivery
2
+ Idempotent Side Effect
3
= Effectively Once

16. 丢失、重复与乱序的来源#

16.1 Producer 端“丢失”#

acks=0；
不检查 Future/Callback；
进程退出前未 Close；
Delivery Timeout；
Buffer 异常被吞；
Serializer 失败；
把发送失败误当成功。

16.2 Broker 端“丢失”#

RF=1；
acks=1 后 Leader 永久故障；
min ISR 过低；
Unclean Election；
副本位于同一故障域；
错误 Reassignment 或人工操作。

16.3 Consumer 端“丢失”#

消息仍在 Kafka，但业务没执行：

Auto Commit 过早；
处理前 Commit；
Worker 未完成就推进水位；
auto.offset.reset=latest；
人工 Reset；
Retention 早于恢复窗口。

16.4 重复#

ACK 丢失后的非幂等 Retry；
应用重复 Send；
业务成功后 Commit 失败；
Rebalance；
Replay；
Outbox Publisher 重试；
外部 API 无幂等。

16.5 乱序#

Kafka 只保证 Partition 内 Log 顺序。乱序可能来自：

1
相同 Key 进入不同 Partition
2
Topic 扩容改变 Key 映射
3
禁用 Idempotence + 多 In-flight Retry
4
同 Partition 并发业务 Worker
5
Retry Topic / DLT
6
多 Topic 合流
7
事件时间与到达时间不同

17. 可靠性配置矩阵#

17.1 Producer#

1
acks=all
2
enable.idempotence=true
3
retries=2147483647
4
max.in.flight.requests.per.connection=5
5
delivery.timeout.ms=120000
6
request.timeout.ms=30000

多数场景保留 Kafka 4.3 默认值，主要通过 SLO 调整 delivery.timeout.ms，不要盲目设置巨大 Retry 次数和应用级重发。

17.2 Topic#

关键事件常见：

1
RF=3
2
min.insync.replicas=2
3
unclean.leader.election.enable=false
4
Rack Awareness

17.3 Transaction Internal Topic#

生产默认：

1
transaction.state.log.replication.factor=3
2
transaction.state.log.min.isr=2

单节点实验临时降低后不要复制到生产。

17.4 Consumer#

事务下游：

1
isolation.level=read_committed
2
enable.auto.commit=false

具体 Offset 是否手动提交取决于 Kafka Transaction、Streams 或框架 Container Transaction。

18. Metrics 与告警#

Kafka 可靠性故障诊断树

18.1 Broker 副本#

1
UnderReplicatedPartitions
2
UnderMinIsrPartitionCount
3
OfflineReplicaCount
4
IsrShrinksPerSec
5
IsrExpandsPerSec
6
FailedIsrUpdatesPerSec
7
ReplicaFetcher MaxLag

正常情况下 Offline Replica 为 0，非故障恢复期 ISR Shrink/Expansion 应接近 0。

18.2 Producer#

1
record-error-rate
2
record-retry-rate
3
request-latency
4
requests-in-flight
5
record-queue-time
6
bufferpool-wait-time
7
produce-throttle-time

18.3 Transaction#

1
transaction-start-rate
2
transaction-commit-rate
3
transaction-abort-rate
4
transaction-duration
5
Open Transaction
6
LSO Lag
7
Transaction Coordinator Load

实际指标名应以具体 Kafka 4.3.x 客户端和 Reporter 输出为准。

18.4 告警#

至少包括：

1
UnderMinISR > 0
2
OfflineReplicaCount > 0
3
ISR Shrink 突增
4
Unclean Election
5
Producer Error/Retry 激增
6
Transaction Abort 激增
7
Transaction Duration 超限
8
read_committed LSO Lag

19. 源码阅读地图#

固定 Kafka 4.3.x Tag 阅读，避免 trunk 方法变化。

19.1 Produce 与 Replica#

1
KafkaApis
2
ReplicaManager
3
Partition
4
UnifiedLog

关注：

ProduceRequest 如何进入 Append；
acks 如何影响 Delayed Produce；
min ISR 在 Append 前后如何校验；
ProduceResponse 何时完成；
Leader Epoch 如何传递。

19.2 Replica Fetch#

1
ReplicaFetcherManager
2
ReplicaFetcherThread
3
AbstractFetcherThread

关注：

Follower Fetch Offset；
FetchResponse 追加；
Diverging Epoch；
Log Truncation；
ISR 恢复。

19.3 ISR、HW 与 ELR#

1
Partition
2
ReplicaManager
3
AlterPartitionManager
4
KRaft Partition Registration

关注 ISR Expand/Shrink、HW 更新、Controller 确认和 ELR 候选状态。

19.4 Idempotent Producer#

Client：

1
TransactionManager
2
ProducerIdAndEpoch
3
Sender
4
ProducerBatch

Broker：

1
ProducerStateManager
2
ProducerAppendInfo
3
BatchMetadata
4
VerificationStateEntry

19.5 Transaction Coordinator#

1
TransactionCoordinatorService
2
TransactionCoordinatorShard
3
TransactionMetadata
4
TransactionState
5
TransactionLog
6
TransactionMarkerChannelManager

关注：

1
transactional.id 分区
2
InitProducerId
3
AddPartitions
4
Prepare Commit/Abort
5
Marker
6
Timeout Abort
7
Coordinator Failover

19.6 Isolation Fetch#

1
UnifiedLog.read
2
FetchIsolation
3
AbortedTxn
4
TransactionIndex
5
CompletedTxn

推荐顺序：

1
acks/min ISR 文档
2
→ Partition Append
3
→ Replica Fetcher
4
→ ISR/HW/ELR
5
→ ProducerStateManager
6
→ Client TransactionManager
7
→ Transaction Coordinator
8
→ Marker/LSO/Isolation Fetch

20. 可运行实验：三 Broker KRaft#

以下用于本地学习。故障注入和 Broker Stop 不要直接在生产环境执行。

20.1 Docker Compose#

1
x-kafka-common: &kafka-common
2
  image: apache/kafka:4.3.0
3
  environment: &kafka-env
4
    KAFKA_PROCESS_ROLES: broker,controller
5
    KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
6
    KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
7
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
8
    KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
9
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
10
    KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
11
    KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
12
    KAFKA_DEFAULT_REPLICATION_FACTOR: 3
13
    KAFKA_MIN_INSYNC_REPLICAS: 2
14

15
services:
16
  kafka-1:
17
    <<: *kafka-common
18
    container_name: kafka-r1
19
    hostname: kafka-1
20
    ports:
21
      - "19092:19092"
22
    environment:
23
      <<: *kafka-env
24
      KAFKA_NODE_ID: 1
25
      KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:19092,CONTROLLER://:9093
26
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-1:9092,EXTERNAL://localhost:19092
27
    volumes:
28
      - kafka_r1:/var/lib/kafka/data
29

30
  kafka-2:
31
    <<: *kafka-common
32
    container_name: kafka-r2
33
    hostname: kafka-2
34
    ports:
35
      - "29092:29092"
36
    environment:
37
      <<: *kafka-env
38
      KAFKA_NODE_ID: 2
39
      KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:29092,CONTROLLER://:9093
40
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-2:9092,EXTERNAL://localhost:29092
41
    volumes:
42
      - kafka_r2:/var/lib/kafka/data
43

44
  kafka-3:
45
    <<: *kafka-common
46
    container_name: kafka-r3
47
    hostname: kafka-3
48
    ports:
49
      - "39092:39092"
50
    environment:
51
      <<: *kafka-env
52
      KAFKA_NODE_ID: 3
53
      KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:39092,CONTROLLER://:9093
54
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-3:9092,EXTERNAL://localhost:39092
55
    volumes:
56
      - kafka_r3:/var/lib/kafka/data
57

58
volumes:
59
  kafka_r1:
60
  kafka_r2:
61
  kafka_r3:

启动：

1
docker compose up -d

20.2 创建 Topic#

1
docker exec kafka-r1 \
2
  /opt/kafka/bin/kafka-topics.sh \
3
  --bootstrap-server kafka-1:9092 \
4
  --create \
5
  --topic reliable-events \
6
  --partitions 3 \
7
  --replication-factor 3 \
8
  --config min.insync.replicas=2

查看：

1
docker exec kafka-r1 \
2
  /opt/kafka/bin/kafka-topics.sh \
3
  --bootstrap-server kafka-1:9092 \
4
  --describe \
5
  --topic reliable-events

20.3 停止一个 Follower#

使用 acks=all 持续生产，然后停止一个非 Leader Broker：

1
docker stop kafka-r3

观察 ISR 缩小。RF=3、ISR=2、min ISR=2 时仍可写。

20.4 只剩一个 ISR#

再停止另一个 ISR Broker。如果 Leader 仍在线但 ISR 只剩一个，acks=all 应被拒绝。

恢复：

1
docker start kafka-r2 kafka-r3

等待 ISR 扩回。

20.5 Leader 切换#

Describe 找到某 Partition Leader，停止该 Broker，观察新 Leader、ISR 和已确认数据是否仍可消费。

20.6 Idempotent Producer#

1
Properties props = new Properties();
2
props.put(
3
    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
4
    "localhost:19092,localhost:29092,localhost:39092"
5
);
6
props.put(
7
    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
8
    StringSerializer.class.getName()
9
);
10
props.put(
11
    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
12
    StringSerializer.class.getName()
13
);
14
props.put(ProducerConfig.ACKS_CONFIG, "all");
15
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
16
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
17

18
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
19
    for (int i = 0; i < 1000; i++) {
20
        producer.send(
21
            new ProducerRecord<>(
22
                "reliable-events",
23
                "order-" + (i % 10),
24
                "event-" + i
25
            ),
26
            (metadata, error) -> {
27
                if (error != null) {
28
                    error.printStackTrace();
29
                }
30
            }
31
        );
32
    }
33
}

20.7 Transactional Producer#

1
Properties props = new Properties();
2
props.put(
3
    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
4
    "localhost:19092,localhost:29092,localhost:39092"
5
);
6
props.put(
7
    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
8
    StringSerializer.class.getName()
9
);
10
props.put(
11
    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
12
    StringSerializer.class.getName()
13
);
14
props.put(
15
    ProducerConfig.TRANSACTIONAL_ID_CONFIG,
16
    "transaction-lab-01"
17
);
18

19
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
20
    producer.initTransactions();
21

22
    try {
23
        producer.beginTransaction();
24

25
        producer.send(new ProducerRecord<>(
26
            "reliable-events", 0,
27
            "order-1001", "OrderCreated"
28
        ));
29

30
        producer.send(new ProducerRecord<>(
31
            "reliable-events", 1,
32
            "order-1001", "InventoryReserved"
33
        ));
34

35
        producer.commitTransaction();
36
    } catch (Exception e) {
37
        producer.abortTransaction();
38
        throw e;
39
    }
40
}

20.8 对比事务可见性#

分别启动：

1
--consumer-property isolation.level=read_uncommitted

和：

1
--consumer-property isolation.level=read_committed

将示例改为 abortTransaction()，比较结果。

20.9 Kafka → Kafka 事务模板#

1
consumer.subscribe(List.of("input-events"));
2
producer.initTransactions();
3

4
while (true) {
5
    ConsumerRecords<String, String> records =
6
        consumer.poll(Duration.ofSeconds(1));
7

8
    if (records.isEmpty()) {
9
        continue;
10
    }
11

12
    try {
13
        producer.beginTransaction();
14

15
        for (ConsumerRecord<String, String> record : records) {
16
            producer.send(new ProducerRecord<>(
17
                "output-events",
18
                record.key(),
19
                transform(record.value())
20
            ));
21
        }
22

23
        producer.sendOffsetsToTransaction(
24
            records.nextOffsets(),
25
            consumer.groupMetadata()
26
        );
27

28
        producer.commitTransaction();
29
    } catch (ProducerFencedException fatal) {
30
        throw fatal;
31
    } catch (KafkaException abortable) {
32
        producer.abortTransaction();
33
    }
34
}

生产代码还需处理 Rebalance、Wakeup、Fatal/Abortable Error、Backpressure 和监控。

20.10 实验验收#

应能观察：

一个 Follower 故障时 ISR 缩小但仍可写；
ISR 低于 min ISR 后 acks=all 失败；
Leader 故障后从安全副本选出新 Leader；
Abort Transaction 对 read_committed 不可见；
read_uncommitted 可以读取 Abort 数据；
同 Transactional ID 的旧 Producer 被 Fence；
长 Open Transaction 影响 LSO。

21. 常见错误认知#

21.1 `acks=all` 等全部配置副本#

错误。它等待当前 ISR，并要求满足 min ISR。

21.2 RF=3 一定有三份最新数据#

错误。要看 ISR、Follower Lag 和 Under Replicated 状态。

21.3 HW 等于 Leader LEO#

错误。Leader 尾部可能尚未形成安全复制条件。

21.4 Idempotence 去重所有业务重复#

错误。它识别 Producer Retry Batch，不理解业务身份。

21.5 Transaction 会删除 Abort 数据#

错误。数据仍在 Log，通过 Marker 和 Isolation 隐藏。

21.6 `read_committed` 总能读到 HW#

错误。还受 LSO 限制。

21.7 Kafka EOS 保证 MySQL 只更新一次#

错误。Kafka Transaction 不覆盖 MySQL Local Transaction。

22. 本篇验收清单#

22.1 画图#

能够画出：

1
Producer
2
→ Leader Append
3
→ Follower Fetch
4
→ ISR
5
→ HW
6
→ Consumer Visibility

22.2 状态#

准确区分：

1
Replica Set
2
ISR
3
ELR
4
Leader LEO
5
Follower LEO
6
HW
7
LSO
8
Committed Offset

22.3 幂等#

能够解释：

1
PID
2
Producer Epoch
3
Sequence Number
4
Duplicate Batch
5
OutOfOrderSequence
6
Fencing

22.4 事务#

能够画出：

1
transactional.id
2
→ Transaction Coordinator
3
→ PID/Epoch
4
→ Ongoing
5
→ Participating Partitions
6
→ Prepare Commit/Abort
7
→ Control Marker
8
→ Complete

22.5 Failure Boundary#

面对“丢消息”，先问：

1
应用是否调用 send？
2
Future 是否成功？
3
acks 是什么？
4
ISR 是否满足？
5
是否发生 Unclean Election？
6
Consumer Commit 在哪里？
7
Retention 是否删除？
8
业务异常是否被吞？

面对“重复”，先问：

1
Kafka Log 是否重复？
2
应用是否重复 Send？
3
Producer 是否启用 Idempotence？
4
Consumer 是否 Replay？
5
数据库是否幂等？

22.6 口述验收#

不用看文章，用 45 分钟回答：

一条 acks=all Record 进入 Leader 后，Follower 如何复制，ISR 和 HW 如何限制确认与可见性；Leader 故障后 KRaft Controller 如何从 ISR、ELR 或其他候选中选举；ACK 丢失时 Producer 如何用 PID、Epoch 和 Sequence 去重；多 Partition 原子写入又如何通过 Transaction Coordinator、事务状态与 Marker 实现；最后说明 Kafka Exactly Once 为什么不能自动扩展到 MySQL 和 HTTP。

23. 全文收束：Kafka 可靠性不是一个开关#

Kafka 可靠性可以收束为五层。

Confirm Boundary#

1
acks + min ISR

定义 Producer 何时认为成功。

Replication Boundary#

1
Leader + Follower Fetch + ISR + HW

定义哪些数据可安全恢复和对 Consumer 可见。

Election Boundary#

1
Leader Epoch + ISR + ELR + Unclean Policy

定义故障后哪一段 Log 成为合法历史。

Retry Boundary#

1
PID + Producer Epoch + Sequence

让网络不确定性下的 Retry 不直接变成重复与乱序。

Atomic Visibility Boundary#

1
Transaction Coordinator
2
+ __transaction_state
3
+ Marker
4
+ LSO
5
+ read_committed

把多个 Kafka Partition 和 Consumer Offset 组成原子可见事务。

最终结论是：

Kafka 没有消灭失败，而是把失败转换为一组可持久化、可比较、可隔离和可恢复的状态。

但当消息离开 Kafka，进入：

1
MySQL
2
Payment
3
HTTP API
4
Agent Tool

系统仍需要：

1
业务幂等
2
Outbox / Inbox
3
唯一约束
4
状态机
5
补偿
6
对账

下一篇进入 Spring Kafka 企业工程层：

1
KafkaTemplate
2
@KafkaListener
3
Listener Container
4
AckMode
5
Error Handler
6
Blocking Retry
7
Retry Topic
8
DLT
9
Idempotent Consumer
10
Transaction Manager
11
Outbox / CDC

回答：

如何把 Kafka 底层可靠性机制组织成一个不会因异常重试、数据库事务和框架默认行为而失控的 Java 应用。