《Java 消息中间件·可靠消息与消费治理》1.2 生产端：acks、重试与幂等

先把 4.3.0 的默认值打出来

讨论配置之前先看客户端自己的默认值。用 ProducerConfig.configDef().defaultValues() 和 ConsumerConfig 的同名方法读取，本机实测（Temurin 25.0.5+8 / kafka-clients 4.3.0 / 2026-10-29）：

producer.acks=all
producer.enable.idempotence=true
producer.retries=2147483647
producer.delivery.timeout.ms=120000
producer.request.timeout.ms=30000
producer.max.in.flight.requests.per.connection=5
producer.linger.ms=5
producer.batch.size=16384
consumer.enable.auto.commit=true
consumer.auto.commit.interval.ms=5000
consumer.max.poll.records=500
consumer.isolation.level=read_uncommitted
consumer.group.protocol=classic
consumer.session.timeout.ms=45000
consumer.auto.offset.reset=latest

生产端这一组默认值已经是「最强写入」：acks=all、幂等开启、重试次数等于 Integer.MAX_VALUE、投递总时限 120 秒。也就是说，一个什么都不配的生产者在发送语义上并不弱。

但默认值只说明客户端愿意等，不说明服务端一定收得下。acks=all 是否等于「不会丢」，取决于四个前提。

acks=all 的四个前提

前提一：topic 的副本数至少为 2

单副本 topic 的 ISR 只有 leader 自己，acks=all 退化成 acks=1。这一点在卷一 4.2 节已经用三节点集群验证过；本篇不再重复。

前提二：min.insync.replicas 不大于副本数

这是 4.3.0 里一个容易误判的细节。直觉上，min.insync.replicas=3 配在 2 副本 topic 上应该让所有写入失败，因为永远凑不齐 3 个同步副本。实测不是这样。

本机建了一个 2 副本、min.insync.replicas=3 的 topic：

Topic: r12.clamp	PartitionCount: 1	ReplicationFactor: 2	Configs: min.insync.replicas=3
	Topic: r12.clamp	Partition: 0	Leader: 2	Replicas: 2,3	Isr: 2,3

在 ISR 为 2 的情况下，acks=all 写入成功：

topic=r12.clamp acks=all retries=0
ok=1 fail=0 offsetMin=0 offsetMax=0

把 follower 冻结、等 ISR 收缩到 1 之后，同一个 topic 才拒绝写入：

Topic: r12.clamp	Partition: 0	Leader: 2	Replicas: 2,3	Isr: 2	Elr: 3

FAIL[2] java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.NotEnoughReplicasException: Messages are rejected since there are fewer in-sync replicas than required.
topic=r12.clamp acks=all retries=0
ok=0 fail=3 offsetMin=-1 offsetMax=-1
elapsedMs=5867

为什么 ISR=2 时能写？反编译 kafka-server-4.3.0 里的 kafka.cluster.Partition，produce 路径调用的 effectiveMinIsr 不是直接取配置，而是取配置与「远端副本数 + 1」的较小值：

invokevirtual  LogConfig.minInSyncReplicas:I
invokevirtual  ConcurrentHashMap.size:()I
iconst_1
iadd
invokevirtual  RichInt.min$extension:(II)I

远端副本数 + 1 在正常情况下就是副本数，所以有效值 = min(min.insync.replicas, 副本数)。2 副本配 3，实际按 2 执行；ISR 掉到 1 才低于有效值，才报 NotEnoughReplicasException。

这个行为有它的合理性：避免一个永远写不进去的 topic。但它也意味着把 min.insync.replicas 配得比副本数大，不会得到更强的保证，只会得到一张看起来更严的配置表。评审时要同时看副本数。

前提三：发送时 ISR 不少于有效 min.insync.replicas

acks=all 等的是当时 ISR 的所有副本，不是「所有副本」。一个副本被踢出 ISR 后，它就不再被等待，也不在确认范围内。等 ISR 收缩到低于有效 min.insync.replicas，写入会被显式拒绝。

本机把 3 副本、min.insync.replicas=2 的 topic 的一个 follower 冻结 42 秒，ISR 从 2,3,1 收缩到 2,1。broker 侧这三个默认值先核实过：

replica.lag.time.max.ms=30000
unclean.leader.election.enable=false
min.insync.replicas=1    # broker 默认值；本 topic 覆盖为 2

随后一次 leader 故障中，acks=all 的 100 条全部保留。同一个实验里，如果再把 ISR 打到 1，写入会得到：

topic=r12.acksall acks=all retries=1
ok=0 fail=10 offsetMin=-1 offsetMax=-1
elapsedMs=7195

日志里对应的错误是 NOT_ENOUGH_REPLICAS。这是好事：生产者知道失败了，可以重试或补偿。

前提四：没有发生不干净选举

即使四条都满足，acks=all 确认过的记录仍可能在不干净选举后消失。不干净选举把不在 ISR 里的副本提为 leader，这个副本缺少的尾部就是永久缺失。它不是 acks 能控制的，属于 broker 段的选举策略。

所以完整的一句话是：acks=all 保证的是「当时 ISR 的确认」，不是「永久不丢」。 前者是客户端配置，后者要副本数、ISR 策略、选举策略一起配合。

实验：同一次 leader 故障，两种 acks

实验设置

环境是 Temurin 25.0.5+8、kafka-clients 4.3.0、apache/kafka:4.3.0，1 个独立 controller 加 3 个 broker（k1/k2/k3，宿主机端口 29091–29093）。

controller 独立出来是实验设计：combined 模式下 controller 与 broker 在同一个进程里，冻结一个节点再杀另一个，3 个 voter 只剩 1 个，元数据层会先于数据层失效，故障现象就混在一起了。把 controller 拆成单独进程后，冻结或杀掉 broker 只影响数据副本，不会连带 controller quorum。这不是生产拓扑推荐，只是为了把变量分开。

两个对照 topic：

kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create \
  --topic r12.acks1 --replica-assignment 1:3 --config min.insync.replicas=1
kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create \
  --topic r12.acksall --replica-assignment 2:3:1 --config min.insync.replicas=2

初始状态：

Topic: r12.acks1	Partition: 0	Leader: 1	Replicas: 1,3	Isr: 1,3
Topic: r12.acksall	Partition: 0	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1

然后 docker pause mqrel-k3，等 42 秒让 ISR 收缩：

Topic: r12.acks1	Partition: 0	Leader: 1	Replicas: 1,3	Isr: 1
Topic: r12.acksall	Partition: 0	Leader: 2	Replicas: 2,3,1	Isr: 2,1

r12.acks1 变成「leader 单副本」；r12.acksall 还有两个同步副本。这正是我们要对照的两个窗口。

acks=all：两次 leader 死亡后仍是 100 条

先向 r12.acksall 发 100 条，acks=all、retries=5、同步等待每条确认：

topic=r12.acksall acks=all retries=5
ok=100 fail=0 offsetMin=0 offsetMax=99
elapsedMs=1830

然后 docker kill mqrel-k2，杀掉该分区的 leader。新 leader 是 k1：

Topic: r12.acksall	Partition: 0	Leader: 1	Replicas: 2,3,1	Isr: 1	Elr: 2

从新 leader 从头消费：

100
v0
v1
...
v98
v99

100 条完整。把 k2 启动、等它重新入 ISR 后，再杀 k1（此时 r12.acksall 的 leader 是 k2），该分区仍有 100 条：

records=100
v0
v1
...
v98
v99

两次 leader 死亡、两次切换，acks=all 的数据都在。

acks=1：100 条确认，幸存副本 0 条

同一个冻结窗口里，向 r12.acks1 发 100 条，acks=1、retries=0：

topic=r12.acks1 acks=1 retries=0
ok=100 fail=0 offsetMin=0 offsetMax=99
elapsedMs=1024

生产者拿到 100 个成功、位移 0–99。此刻数据只在 leader k1 上，follower k3 被冻结且已被踢出 ISR。

docker kill mqrel-k1 之后：

Topic: r12.acks1	Partition: 0	Leader: none	Replicas: 1,3	Isr: 	Elr: 1	LastKnownElr: 1

分区的 ISR 空了。默认 unclean.leader.election.enable=false 时，分区会一直离线，不会拿落后副本顶上。先 docker unpause mqrel-k3 让落后副本重新上线，再手工执行一次不干净选举，把「数据分歧」变成可观察的「数据丢失」：

kafka-leader-election.sh --bootstrap-server 127.0.0.1:9092 \
  --election-type UNCLEAN --topic r12.acks1 --partition 0

输出：

Successfully completed leader election (UNCLEAN) for partitions r12.acks1-0
Topic: r12.acks1	Partition: 0	Leader: 3	Replicas: 1,3	Isr: 3

新 leader 是原来被冻结的 k3。从头消费：

records=0

k3 的段文件也是 0 条：

r12.acks1-0
file=/var/lib/kafka/data/r12.acks1-0/00000000000000000000.log
0

而被杀的 k1 的段文件里有 100 条。把它的日志复制出来，按 20 位 offset 命名才能被 kafka-dump-log.sh 解析：

docker cp mqrel-k1:/var/lib/kafka/data/r12.acks1-0/00000000000000000000.log /tmp/k1.log
docker cp /tmp/k1.log mqrel-k2:/tmp/k1seg/00000000000000000000.log
docker exec mqrel-k2 sh -c 'unset JMX_PORT; /opt/kafka/bin/kafka-dump-log.sh \
  --files /tmp/k1seg/00000000000000000000.log --print-data-log | grep -c "^| offset:"'

100

把三个数字放在一起：

三个数字里没有一个是异常或错误日志。生产者的发送调用全部成功，broker 按要求完成了选举，消费者正常读到了一个空分区。这就是 acks=1 的静默丢失。

这个实验的边界

这个实验用 docker pause 冻结 follower，是为了让「follower 没追上」这个窗口稳定出现；用 UNCLEAN 选举，是为了让分歧变成可读的丢失。两件事都是显式的故障注入，不是默认行为。

如果不动用不干净选举，acks=1 的 100 条在 k1 恢复后仍在它的磁盘上，分区也会重新有 leader。真正的丢失条件是「k1 的磁盘或节点永久不可用」，或者在 k1 缺席时强行提升落后副本。前者是硬件事故，后者是运维决策。两者都会让 acks=1 的确认变成一张空头支票。

反过来，acks=all 在这个实验里能扛住两次 leader 死亡，前提是每次死亡时 ISR 里还有另一个有数据的副本。如果 ISR 掉到 1 再杀 leader，acks=all 也救不了 —— 它会在写入时先报 NOT_ENOUGH_REPLICAS，而不是先收下再丢。

retries 与 delivery.timeout.ms 不是两个独立旋钮

retries 是次数上限，delivery.timeout.ms 是时间上限，谁先到算谁。DELIVERY_TIMEOUT_DOC 的原文说它限制三段时间：发送前的排队（含 linger）、等待确认、以及可重试失败的等待。

这意味着：

• retries=2147483647 不等于永远重试，120 秒到了照样失败；
• retries=3 配一个很大的 delivery.timeout.ms，实际是「3 次之后放弃」，后面的时间窗用不上；
• delivery.timeout.ms 必须大于 request.timeout.ms + linger.ms，否则第一批可能还没发出去就过期。

在本机用 2 副本、min.insync.replicas=3 的 topic，冻结 follower 让 ISR=1，三种设置各发 3 条，全部失败：

第三行最能说明关系：重试次数几乎无限，但每条记录的投递时限是 5 秒，3 条顺序等待，总耗时约 15 秒。delivery.timeout.ms 是真正的兜底，retries 只是在这段时间里允许试几次。

第一行 5.9 秒的绝对值偏高，是宿主机高负载下单次运行的结果，不能当作「无重试的固定开销」。三行耗时的递增关系（5.9 s < 8.5 s < 15.7 s）与重试设置一致，这个方向性结论可以复用。

还有一层：这两个限制按批次计时，不是按记录。计时起点是批次创建时刻，一个批次里 100 条记录共享同一个截止时间；先到期的批次整批失败，回调逐条收到同一个异常。所以 Expiring N record(s) 里的 N 是批次里的记录数，不是分区积压总数。

幂等生产者去重的是哪一种重复

enable.idempotence=true 在 4.3.0 默认开启。它不是独立开关，而是一组约束的入口。显式把 acks=1 和幂等放在一起，构造阶段直接失败：

Exception in thread "main" org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.
	at org.apache.kafka.clients.producer.ProducerConfig.postProcessAndValidateIdempotenceConfigs(ProducerConfig.java:613)

幂等的实现靠生产者 ID（PID）和每分区的序列号。开启幂等后，段文件里能看到：

baseOffset: 0 lastOffset: 0 count: 1 baseSequence: 0 lastSequence: 0 producerId: 2001 producerEpoch: 0

broker 按 <PID, 分区, sequence> 记住最近写入的批次。生产者因为「响应丢了」重发同一个批次时，序列号相同，broker 丢弃重复数据但返回成功。这一段是协议行为，本节只验证了 PID 与连续序列号的生成，没有注入重复批次；能实测的是下面「它不覆盖什么」。这就是它覆盖的重复类型：同一个批次的重试。

它不覆盖「业务层调了两次 send」。本机用幂等生产者把同一条业务记录连发两次：

send#1 -> p=0 off=0
send#2 -> p=0 off=1

段文件里是两条真实记录，同一个 PID，序列号从 0 到 1：

baseOffset: 0 lastOffset: 0 count: 1 baseSequence: 0 lastSequence: 0 producerId: 3001
| offset: 0 ... key: book-1 payload: loan-duplicate
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: 1 lastSequence: 1 producerId: 3001
| offset: 1 ... key: book-1 payload: loan-duplicate

两次调用生成两个不同序列号的批次，broker 没有理由去重。所以幂等的边界要写成：

「开启幂等」只关掉了第一行。业务幂等键、去重表、消费端位移管理，是第 2 章的内容。

一份生产端配置与检查表

acks=all
enable.idempotence=true
delivery.timeout.ms=120000
request.timeout.ms=30000
linger.ms=5
max.in.flight.requests.per.connection=5
# retries 不显式设置，交给默认的 MAX_VALUE，由 delivery.timeout.ms 兜底

上线前逐条检查：

本节判据

回答四个问题：

1. acks=all 是否足以保证不丢？——不够。还要副本数、有效 min.insync.replicas、ISR 当时够、以及不发生不干净选举。本机 acks=1 的 100 条在 leader 死后读到 0 条；acks=all 的 100 条扛住了两次 leader 死亡。
2. min.insync.replicas=3 配在 2 副本 topic 上会怎样？——4.3.0 的有效值是 min(3, 2) = 2，ISR=2 时仍可写，ISR=1 时才拒绝。不要用大于副本数的配置制造安全感。
3. retries=1000 是否意味着会重试 1000 次？——不。delivery.timeout.ms 先到就失败。本机把时限设为 5 秒，3 条记录约 15 秒后全部失败。
4. 开启幂等后，同一条业务消息发两次会去重吗？——不会。两次 send() 生成两个序列号，段文件里是两条记录；幂等只去掉同一批次的重试重复。

下一节转向消费端：位移到底代表什么，自动提交在 close() 时会提交到哪，以及手动提交失败后哪些异常可以重试、哪些必须重新加入消费组。

阅读导航：上一节 1.1 端到端可靠性的责任边界 ｜ 下一节 1.3 消费端：手动提交与位移管理