《Java 消息中间件·Kafka 内核与客户端》3.2 acks、重试与幂等

三个 acks 值，三种「成功」

acks 这个参数的名字容易让人以为它只影响性能：等一个副本、等全部副本、或者不等。它真正决定的是**「成功」这个词的含义**。

4.3.0 的 ProducerConfig.ACKS_DOC 写得比大多数教程准确：

<li><code>acks=0</code> If set to zero then the producer will not wait for any acknowledgment from the
server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be
made that the server has received the record in this case, and the <code>retries</code> configuration will not
take effect (as the client won't generally know of any failures). The offset given back for each record will
always be set to <code>-1</code>.
<li><code>acks=1</code> This will mean the leader will write the record to its local log but will respond
without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after
acknowledging the record but before the followers have replicated it then the record will be lost.
<li><code>acks=all</code> This means the leader will wait for the full set of in-sync replicas to
acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica
remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting.

三段话对应三个不同的失败窗口：

注意最后一列的差别。acks=0 和 acks=1 的危险不在于「可能丢」，而在于「丢了生产者不知道」。acks=all 也会失败，但失败是显式的，应用层有机会处理。

4.3.0 的默认值是 acks=all，retries=Integer.MAX_VALUE，delivery.timeout.ms=120000，enable.idempotence=true。这四个默认值是一套组合，拆开看容易理解错。

为什么单节点实验只能证明一半

完整的 acks=1 丢消息场景需要三个副本：leader 写本地日志、返回成功、还没复制到 follower 时 leader 崩溃。单节点只有 RF=1，leader 就是唯一副本，没有「还没复制」这个状态。

所以本节在单节点上做两件事：

1. 杀掉 broker，观察三种 acks 下生产者能不能发现失败；
2. 用落盘记录数对比「生产者认为成功」和「实际存在」。

至于「leader 确认后崩溃导致 acks=1 丢消息」，本机三节点集群的实测数据在 4.2 节，那里有 NOT_ENOUGH_REPLICAS 的真实输出。本节不会在单节点上假装三副本。

实验一：acks=0 的静默丢失

程序 AcksLab 以 20 ms 间隔发 400 条，acks=0、linger.ms=0、delivery.timeout.ms=8000、max.block.ms=5000。发到第 3 秒时 docker kill broker，第 6.5 秒 docker start。

结果：

sent=400 intervalMs=20 elapsedMs=27335
ok=393 fail=7 offsetMinusOne=393

生产者认为 393 条成功，7 条失败。所有成功记录的 RecordMetadata.offset() 都是 -1，这与 ACKS_DOC 的说明一致：没有服务端响应，就没有位移。

这些耗时都是单次运行，随调度波动，不作基准。程序按固定 20 ms 间隔发送，elapsedMs 主要反映多少次 send() 卡在等待上，不能用来比较吞吐。

再看 broker 重启后实际落盘多少：

topic=a2.acks0 total=388
p0=388

393 条「成功」里，只有 388 条真的在日志里。5 条消息被生产者报告成功，但 broker 上不存在，也不会抛任何异常。

失败的 7 条长这样：

FAIL[113] org.apache.kafka.common.errors.TimeoutException: Topic a2.acks0 not present in metadata after 5000 ms.
FAIL[109] org.apache.kafka.common.errors.TimeoutException: Expiring 4 record(s) for a2.acks0-0:8003 ms has passed since batch creation.

第一类是 broker 不可达时元数据拿不到，send() 在 max.block.ms 到期后失败；第二类是批次在 delivery.timeout.ms 到期后过期。两类都是显式失败，但只覆盖了没能写进 socket 的那部分记录。已经写进 socket 的 5 条，没有任何回调会告诉你有问题。

5 条这个数字不大，是因为 broker 停机期间生产者的 send() 大部分时间卡在元数据等待上，真正能写进 socket 的窗口很短。窗口越短，丢得越少；但机制不变：acks=0 的成功是一个本地动作，不是服务端确认。

实验二：acks=all 与 acks=1 的显式失败

同样的杀 broker 节奏，改成 200 条、20 ms 间隔、delivery.timeout.ms=8000：

# acks=all
sent=200 intervalMs=20 elapsedMs=22563
ok=182 fail=18 offsetMinusOne=0

# acks=1
sent=200 intervalMs=20 elapsedMs=20969
ok=181 fail=19 offsetMinusOne=0

失败样本：

FAIL[75] org.apache.kafka.common.errors.TimeoutException: Topic a2.acksall not present in metadata after 5000 ms.
FAIL[61] org.apache.kafka.common.errors.TimeoutException: Expiring 14 record(s) for a2.acksall-0:8004 ms has passed since batch creation.

落盘数与成功数完全对上：

topic=a2.acksall total=182
topic=a2.acks1 total=181

也就是说，在这台单节点、RF=1 的集群上，acks=1 和 acks=all 的表现几乎一样：能确认的都落盘了，不能确认的都显式失败。差异只有 1 条，属于调度噪声。

这不是说两个配置等价。它说明的是：没有副本时，acks=all 退化成 acks=1。ISR 里只有 leader，等全部 ISR 就是等 leader。真正的差别要在三副本上才出现，4.2 节会给出 docker stop k2 后 acks=all 仍成功、再停一个副本后报 NOT_ENOUGH_REPLICAS 的完整记录。

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

很多人同时设 retries=3 和 delivery.timeout.ms=60000，以为重试 3 次、总时限 60 秒。实际关系是：retries 是次数上限，delivery.timeout.ms 是时间上限，谁先到算谁。

DELIVERY_TIMEOUT_DOC 的原文：

An upper bound on the time to report success or failure after a call to send() returns.
This limits the total time that a record will be delayed prior to sending, the time to await acknowledgement
from the broker (if expected), and the time allowed for retriable send failures.

这段时限包含三部分：发送前的排队（包括 linger）、等待确认、以及重试。所以：

• retries=MAX_VALUE 不等于「永远重试」，delivery.timeout.ms 到了就失败；
• delivery.timeout.ms 必须大于 request.timeout.ms + linger.ms，否则第一批还没等到响应就过期；
• 把 retries 设成 3 而 delivery.timeout.ms 很大，实际是「3 次之后放弃」，后面不会再试。

还有一层容易被忽略：这两个限制都是按批次算的，不是按记录。计时起点是批次创建时刻（ProducerBatch.createdMs），重试次数记在批次上。所以一个批次里 100 条记录共享同一个截止时间，先到期的批次整批失败，回调会逐条收到同一个异常，日志里汇总成一条；同一个分区里后建的批次不受影响，还能继续重试。日志里 Expiring 14 record(s) 的「14」就是批次里的记录数，不是这个分区积压的总数。

本机实验里 delivery.timeout.ms=8000、request.timeout.ms=3000、linger.ms=0。broker 停机超过 8 秒后，批次过期，报的就是：

Expiring 14 record(s) for a2.acksall-0:8004 ms has passed since batch creation.
The request has not been sent, or no server response has been received yet.

这句话来自 4.3.0 的 Sender.failExpiredBatches，第 369 行：

String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition
    + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation. "
    + "The request has not been sent, or no server response has been received yet.";

它明确告诉你两种可能：请求没发出去，或发出去了但没有响应。生产者无法区分这两者——这正是重试可能造成重复的根源。

幂等生产者：PID 与 sequence

4.3.0 里 enable.idempotence 默认是 true。它不是一个独立开关，而是一组约束的入口。ProducerConfig.postProcessAndValidateIdempotenceConfigs 对 acks 的检查原文是：

final short acks = Short.parseShort(acksStr);
if (acks != (short) -1) {
    if (userConfiguredIdempotence) {
        throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " +
            "producer. Otherwise we cannot guarantee idempotence.");
    }
    log.info("Idempotence will be disabled because {} is set to {}, not set to 'all'.", ACKS_CONFIG, acks);
    shouldDisableIdempotence = true;
}

同一个方法里还有两条约束：retries 必须大于 0；max.in.flight.requests.per.connection 必须不超过 5（常量 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE = 5）。违反前者时，显式开启幂等会抛 ConfigException；违反后者无论是否显式开启都会抛。

本机把 enable.idempotence=true 和 acks=1 放在一起构造，实际抛出的就是这段：

org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.

所以「幂等 + acks=1」不是「稍微弱一点的幂等」，而是配置非法（显式开启时）。

幂等的实现靠两个字段：生产者 ID（PID）和每个分区的序列号（sequence）。开启幂等后，broker 在批次头里能看到：

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

关闭幂等后，同样的位置变成：

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

这是本机 kafka-dump-log.sh 的真实输出。producerId=12 是 broker 分配给这次生产者会话的；baseSequence 从 0 开始，每个批次递增。前面 3.1 的 gzip 批次里，baseSequence: 0 lastSequence: 296 表示这一批 297 条记录的序列号是 0 到 296。

机制上，broker 对每个 (PID, 分区) 记录最近若干个批次的序列号。收到重复序列号的批次时，它丢弃数据但返回成功；收到序列号跳跃时，返回 OutOfOrderSequenceException。这样，生产者因为「响应丢失」而重发同一个批次时，broker 不会写入第二份。

需要说清楚边界：这一段去重逻辑在 broker 侧，本节没有读 broker 源码，也没有做「注入重复批次」的实验。能验证的是客户端确实生成了 PID 和连续序列号，以及配置约束真实生效。broker 的去重行为按协议描述，不作为本机实测结论。

幂等能去掉哪一种重复

幂等生产者解决的是生产者重试导致的重复，也就是同一个批次被写了两次。它不解决：

• 应用层调用 send() 两次（两次是不同的批次，序列号不同）；
• 消费者处理完、提交位移前崩溃，重启后重新消费（消费侧重复，卷二）；
• 跨会话重复：PID 在生产者重启后会变，新会话不认识旧批次；
• 跨分区重复：序列号是按分区维护的，同一条业务记录写到两个分区，幂等管不了。

所以「开启幂等」不等于「业务不重复」。它只是把「网络重试」这个重复来源去掉了。业务幂等键、去重表、消费者位移管理，都是卷二的内容。

顺序：in-flight 与重试的相互作用

max.in.flight.requests.per.connection 默认 5，允许一个连接上同时有 5 个请求在途。没有幂等时，如果第 1 个请求失败重试、第 2 个请求成功，顺序就可能颠倒。

幂等开启后，这个值最多只能是 5（源码常量 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE = 5），并且 broker 会按序列号拒绝乱序批次。所以「幂等 + 5 个在途」仍然保序。

本机没有做「故意让第一个请求失败、观察乱序」的实验，这部分在第 7.3 节结合 Sender 与 NetworkClient 的源码展开。这里只给结论和它的前提。

一份可以直接用的生产者配置

Properties p = new Properties();
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19092");
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
p.put(ProducerConfig.ACKS_CONFIG, "all");
p.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
p.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "120000");
p.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "30000");
p.put(ProducerConfig.LINGER_MS_CONFIG, "5");
// retries 不显式设置，交给默认的 MAX_VALUE，由 delivery.timeout.ms 兜底

三个容易犯的错：

回到 libraryd

NaiveProducer 到 IdempotentProducer 的演进，在本卷里就是上面这份配置的差异：acks 从默认改为显式 all，幂等保持开启，delivery.timeout.ms 按业务能接受的最大延迟设置。

但 IdempotentProducer 仍然可能写出重复的借阅事件，只要应用层重试了整个 send 流程。要解决那个问题，需要业务键和去重表，那是卷二第 2 章的内容。本卷只负责把「网络层重复」这一种来源关掉。

下一节换一个方向：消息发出去之前，分区是怎么选的。分区选错，后面的 acks 和幂等都救不了顺序和负载。

阅读导航：上一节 3.1 批次、linger 与压缩 ｜ 下一节 3.3 分区器与键设计