工程学习台

主题

主要章节

Kafka 客户端配置、确认与恢复：在真实 KRaft 集群上验证故障窗口 ​

本页固定 Kafka 4.1.2（客户端与服务端）与 spring-kafka 4.0.8，用一条订单事件回答“默认值是什么、显式改了什么、故障后还能证明什么”。机制权威仍在三篇既有正文（生产、追加与复制、消费位点与重平衡、重试与积压），它们的连续原码固定 Kafka 3.7.1；本页两部分是新增的：4.1.2 的配置默认值/显式值表，以及同 JVM 内 3 broker + 3 controller 真实 KRaft 集群上的 8 个故障窗口实验。两套版本分别登记，不把不同 tag 拼成一次运行。

实验入口是 python3 labs/kafka-reliability/run.py，一次完整运行 8 个测试、8 条观察行、exit 0，证据在 evidence/u57-2026-10-27.json。它不是 listener mock：KafkaClusterTestKit 运行的是真实 broker 代码；但它与客户端同进程，不能证明多机、网络分区或断电恢复。

一、默认值与显式值：4.1.2 客户端配置表 ​

配置问题的根源是很多人只记得“把 acks 设成 all”，却不知道其余关键项的默认值。下表来自固定 tag 的 ProducerConfig.java / ConsumerConfig.java（producer、consumer）与 4.1 producer 配置文档，2026-10-27 查询。

配置4.1.2 默认值作用域/单位改变了什么
acksall每批次确认条件3.0 起默认就是 all（KIP-679）；0 不等 broker 响应，1 只等 leader 本地追加
enable.idempotencetrueproducer 身份内去重默认开启；与 acks=1、retries=0、max.in.flight>5 的显式组合在构造期报错
retriesInteger.MAX_VALUE次数实际由 delivery.timeout.ms 截断，不是无限重试
delivery.timeout.ms120000（2 分钟）毫秒，从 send() 起算覆盖批等待、重试和请求超时；到达即失败，不再重试
request.timeout.ms30000毫秒，单次请求只约束一次 Produce 请求等待
max.block.ms60000毫秒限制 send() 等待元数据和缓冲分配的时长，不是交付 deadline
max.in.flight.requests.per.connection5每连接在途请求幂等开启时上限就是 5，超过构造期报错
batch.size / linger.ms16384 字节 / 5 毫秒批次攒批换吞吐；linger 是等待上限不是保证
buffer.memory33554432（32 MiB）字节客户端发送缓冲；耗尽时 send() 阻塞到 max.block.ms
enable.auto.commit（consumer）true位点提交自动提交不知道业务事务何时提交，见三种进度
auto.offset.resetlatest无位点/位点越界时本页实验证明它会把越界静默变成“从头/从新开始”
isolation.levelread_uncommitted消费者可见性事务消费者必须显式改 read_committed
max.poll.records500每条消息上限与 max.poll.interval.ms（默认 300000）共同决定单批处理预算
session.timeout.ms / heartbeat.interval.ms45000 / 3000毫秒前者看心跳可达，后者是心跳频率
fetch.max.bytes / max.partition.fetch.bytes / fetch.max.wait.ms52428800 / 1048576 / 500字节/毫秒单次拉取上限与等待，决定滞后消费者的追赶速度

一个容易读错的组合：acks=all 与 enable.idempotence 在 4.1.2 都是默认值，所以“我没有配置”不等于“没有可靠性”；反过来，显式把 acks 改成 1 会触发下一节的配置陷阱。topic 侧的关键默认也不同：min.insync.replicas 默认 1，副本因子不会替你补，必须在建 topic 时显式给。

二、acks 的三个确认方与 ISR 拒绝 ​

acks 描述的是谁确认了：0 表示不等任何 broker 响应（本地发出即认为成功），1 表示 leader 本地追加完成，all 表示分区内满足复制条件。all 不等同于“所有被分配的副本永远在线”：它等的是 requiredOffset 达到复制确认条件，而复制条件由 min.insync.replicas 与当前 ISR 共同裁决。源码侧的分支解释见等的是分区确认条件。

真实集群上把这条写成可回放的输入。建 1 分区 3 副本、min.insync.replicas=2 的 topic，ISR=3 时写入 v1；停掉一个 follower、等 ISR 收缩到 2 后写入 v2；再停一个 follower、ISR=1 时写入 v3：

text
U57 minIsr=PASS acks=all isr3=v1,v2 isr2=v1,v2 isr1=REJECTED-after-restart=2-records
  • ISR=3 时写入 v1、ISR=2 时写入 v2，两次都成功（acks=all 在 min ISR=2 下仍能确认）；
  • ISR=1 时第三条被拒：客户端拿到 NotEnoughReplicasException（本实验显式关掉幂等并设 retries=0，否则默认重试会把它掩盖成超时）；
  • 恢复两个 broker、ISR 回到 3 之后，从头读只看到 v1、v2，起点 offset 到终点的差是 2——被拒的消息确实不在日志里，不是“发成功但看不到”。

这条把“拒绝发生在追加前”从纸面变成了可核对的事实：追加前的 ISR 检查与追加后的 NOT_ENOUGH_REPLICAS_AFTER_APPEND 是两个分支，本实验覆盖的是追加前；追加后的分支需要“请求在 purgatory 中等待时 ISR 收缩”，本实验的关闭顺序没有构造出这个时序，如实列为未覆盖。

三、幂等的配置前提：显式冲突报错，隐式降级静默 ​

幂等 producer 的保证依赖三个配置前提。4.1.2 的客户端在构造期就把它们拦住（固定 tag 4.1.2，ProducerConfig.java L577–L611）：

原码java
    private void postProcessAndValidateIdempotenceConfigs(final Map<String, Object> configs) {
        final Map<String, Object> originalConfigs = this.originals();
        final String acksStr = parseAcks(this.getString(ACKS_CONFIG));
        configs.put(ACKS_CONFIG, acksStr);
        final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG);
        boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG);
        boolean shouldDisableIdempotence = false;

        // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation
        if (idempotenceEnabled) {
            final int retries = this.getInt(RETRIES_CONFIG);
            if (retries == 0) {
                if (userConfiguredIdempotence) {
                    throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer.");
                }
                log.info("Idempotence will be disabled because {} is set to 0.", RETRIES_CONFIG);
                shouldDisableIdempotence = true;
            }

            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;
            }

            final int inFlightConnection = this.getInt(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
            if (MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < inFlightConnection) {
                throw new ConfigException("To use the idempotent producer, " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION +
                                          " must be set to at most 5. Current value is " + inFlightConnection + ".");
            }
        }

这段原码定义了两种完全不同的结局：用户显式配置过 enable.idempotence（userConfiguredIdempotence 为真）时，retries=0 与 acks≠all 直接抛 ConfigException；没显式配置过时，同样的值只会把幂等静默关闭（shouldDisableIdempotence=true），日志里一行 Idempotence will be disabled 就过去了。幂等处于开启状态时（显式或隐式开启都一样），max.in.flight>5 永远报错；幂等被关闭后这条上限不再校验。

实验把四类输入都跑了一遍：

text
U57 idempotenceConfig=PASS acks1Explicit=ConfigException retries0=ConfigException inFlight6=ConfigException implicitAcks1=idempotence-disabled-silently

这解释了一个常见事故：有人只在配置文件里删掉了 enable.idempotence=true（以为默认值会接管），同时又保留了 acks=1——4.1.2 的默认值确实是 true，但“用户显式配置过幂等”变成了假，于是幂等被静默降级，没有任何报错。跨版本迁移时要核对的是“谁显式配置了什么”，不是“默认值是不是 true”。 幂等 producer 保护什么讲的 epoch/sequence 机制默认已经开启，正是这张配置表的前提。

四、事务可见性：中止不可见，未决挡住后来者 ​

事务的直觉是“commit 之后才可见、abort 之后看不到”，但真正常被忽略的是未决事务会挡住它之后的普通记录。实验在同一个分区里放三条记录：一条事务中止、一条未决事务记录、一条普通记录：

text
U57 transactionVisibility=PASS readCommitted=aborted-hidden,plain-blocked-until-commit readUncommitted=all-three
  • read_committed 消费者在未决事务提交前读不到任何东西：普通记录虽然已经追加、已经复制，但它在未决事务之后，LSO 停在未决事务起点，不能被 committed 读者越过；
  • 事务提交后，未决事务的记录与普通记录按顺序都可见，中止的那条永远不可见；
  • read_uncommitted 消费者能看到全部三条，包括那条被中止的——它不是“垃圾数据”，只是没有提交。

机制与 3.7.1 的 lastStableOffset 原码见HW 与 LSO。这里补的是 4.1.2 的运行行为：默认 isolation.level=read_uncommitted，忘记改配置就会读到中止事务的数据；改了 read_committed 之后，一个长事务会通过 LSO 阻塞后续所有普通记录的可见性——长事务的代价不仅是事务表大小，还包括消费者可见延迟。

五、位点与输出同事务：EOS 的边界在这里 ​

sendOffsetsToTransaction 把输入消费位点和输出记录放进同一个 Kafka 事务。实验读两条输入，先尝试“输出 + 位点”然后中止，再提交一次：

text
U57 eosOffsets=PASS abortedAttempt=offsets0,outputNone committedAttempt=offsets2,output1

中止的那次事务里，输出记录不可见、位点没有前进（同组新消费者从 0 重放，是位点未提交的等价观察）；提交的那次把输出与位点 2 一起落定，输出 topic 只看到 1 条 attempt-committed。这就是 EOS 在 Kafka 原子域内的完整含义：输入位点、输出记录、事务标记共同提交或共同中止。

边界必须说清：原子域只覆盖 Kafka 资源。如果消费者还要写数据库（比如本课程 U79 的订单表），数据库提交不在这个事务里；数据库已提交而 Kafka 位点未提交会重放（下一节），Kafka 已提交而数据库失败也不会回滚 Kafka。跨资源的方案是 Outbox/Inbox，归 U61 票与 Outbox 页，不在本页预支。另外，Kafka 4.x 的 transaction.two.phase.commit.enable 提供了不同的提交路径，本实验没有启用，结论只适用于经典单阶段提交。

六、业务已完成、位点未提交：重放不是错误 ​

消费端的经典故障窗口：记录已交给业务、业务效果已提交，但位点在提交前进程退出。实验让消费者处理两条、只提交一条的位点后关闭，再用同组新实例启动：

text
U57 offsetReplay=PASS committed=1 replay=e2,e3 duplicateEffect=e2-without-idempotency

e2 被重新投递，如果业务没有幂等，效果就重复了一次。这不是 Kafka 的缺陷，而是“位点与业务效果是两个持久边界”的必然结果：先业务后位点会重放，先位点后业务会遗漏，二选一只能选择重放并让下游幂等。机制上的提交协议（generation、连续完成前缀、异步提交倒退）在三种进度和一个空洞与提交源码；本实验补的是真实重放行为：观察行里的 duplicateEffect=e2-without-idempotency 是故意不写幂等的结果，用来证明“重放确实会发生”，而不是证明“必须靠 Kafka 解决”。

七、cooperative 重平衡：成员退出后的分区转移 ​

2 分区、两个使用 CooperativeStickyAssignor 的成员各自拿到 1 个分区；关掉一个之后，幸存者接管两个：

text
U57 cooperativeRebalance=PASS initial=1+1 survivor=0,1 polls=7

这条实验能证明的是分区所有权确实转移，不能证明“没有停顿”（cooperative 协议的第一轮只撤销待转移分区，第二轮才授予；本实验没有测转移期间的空窗时长）。静态成员、session.timeout 与 max.poll.interval 的区别、撤销与优雅停机见Join、Sync、Heartbeat 到 cooperative与有界并行、撤销和优雅停机。这里要补一句本实验的边界：两个消费者在同一个 JVM 里，共享堆和 CPU，不是两个 OS 进程；转移时间不能外推为生产环境的 rebalance 时长。

八、保留删除先于消费：OffsetOutOfRange 与静默跳位 ​

Kafka 的保留删除不等待消费者。实验建 1 分区 topic，retention.ms=4000、segment.ms=1000，写入三条后让一个消费者的位点停在 1；等到日志起点推进到 3：

text
U57 retention=PASS beginningAdvancedTo=3 resetNone=OffsetOutOfRange resetEarliest=silentResetTo3 lagDoesNotPreventDeletion=true

消费者的位点 1 落在已删除区间。此时两种配置给出两种完全不同的结局：

  • auto.offset.reset=none：poll 抛 OffsetOutOfRangeException，错误显式、可观测；
  • auto.offset.reset=earliest（默认是 latest）：消费者静默把位置跳到日志起点 3，什么都不报。

静默跳位在生产里最危险：数据缺口被“跳过”了，而应用日志里没有任何异常。保留窗口与恢复时间的算例见恢复窗口必须小于保留窗口；本实验补的是消费端的可观测性选择：关键链路应显式设置 auto.offset.reset=none 并在异常分支告警，而不是依赖默认值。4 秒保留只用于让实验在分钟级完成，不代表任何生产保留策略。

九、任期接管：旧实例被 fence 的实际异常 ​

同一个 transactional.id 被两个实例先后 initTransactions，新实例接管后旧实例继续发消息。微妙的细节是实际抛出的异常类型：

text
U57 producerFencing=PASS old=InvalidProducerEpochException current=committed visible=1

旧实例的 send().get() 得到 InvalidProducerEpochException（“attempted to produce with an old epoch”），而不是很多人预期的 ProducerFencedException。4.1.2 的客户端在 TransactionManager.maybeFailWithError 里把两种错误分开包装，broker 侧的 epoch 比较在 ProducerAppendInfo.checkProducerEpoch 里抛 InvalidProducerEpochException。恢复策略因此不同：InvalidProducerEpoch 表示“这次请求用了旧 epoch”，客户端可以 abort 当前事务后重试；ProducerFenced 是致命错误，旧实例必须停止。把两者混为一谈会导致“遇到 fence 就重启整个进程”或“遇到 epoch 错误就无限重试”这两种错误处理。

broker 侧 3.7.1 的协调器任期校验原码见结束事务先校验任期；本页新增的 4.1.2 客户端侧原码如下（TransactionManager.java L1154–L1174）：

原码java
    private void maybeFailWithError() {
        if (!hasError()) {
            return;
        }
        // for ProducerFencedException, do not wrap it as a KafkaException
        // but create a new instance without the call trace since it was not thrown because of the current call
        if (lastError instanceof ProducerFencedException) {
            throw new ProducerFencedException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " has been fenced by another producer " +
                    "with the same transactionalId");
        }
        if (lastError instanceof InvalidProducerEpochException) {
            throw new InvalidProducerEpochException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " attempted to produce with an old epoch");
        }
        if (lastError instanceof IllegalStateException) {
            throw new IllegalStateException("Producer with transactionalId '" + transactionalId
                    + "' and " + producerIdAndEpoch + " cannot execute transactional method because of previous invalid state transition attempt", lastError);
        }
        throw new KafkaException("Cannot execute transactional method because we are in an error state", lastError);
    }

四类错误对应四种恢复动作：ProducerFencedException 不包装、必须停止旧实例；InvalidProducerEpochException 表示旧 epoch，可 abort 后重试；IllegalStateException 是非法状态转移（例如没有 init 就 begin）；其余包装成 KafkaException。本实验实际触发的是第二种，所以观察行里如实记录 old=InvalidProducerEpochException，不把它改写成 ProducerFenced。

十、确认矩阵：每种确认的参与者与不保证项 ​

把本页与前几节的确认点收拢成一张表。每一行是一个独立的成功点，右边是它能证明的与不能证明的：

确认参与者能证明不能证明
send() 返回客户端缓冲区消息已被接受、进入发送流程已经发出、broker 已收、不会因缓冲失败
Future 成功（acks=all）leader + 达到条件的 ISR复制条件满足（requiredOffset 达标）所有副本都在线、已 fsync 到磁盘、消费者读到
acks=1 成功只有 leaderleader 本地追加follower 是否复制；leader 故障时可能丢
事务 commitTransaction 返回协调器 + 参与分区输出记录与位点按事务可见外部数据库/支付已提交
commitSync 位点成功组协调器位点已持久到 __consumer_offsets业务效果已提交（可能先提交位点后崩溃）
业务事务提交应用数据库业务效果持久化Kafka 位点已提交（下一轮会重放）

矩阵的用途是故障排查的顺序：先问哪个确认缺失，再问业务效果。例如“消息发了但消费者没处理”，要按 acks → Future → 位点 → 业务日志的顺序逐层排查，而不是直接怀疑 Kafka。这一顺序与既有正文故障时先问哪个确认缺失一致，本页把它与 4.1.2 的实验观察合并成可引用的表格。

十一、证明边界：真实 KRaft 集群能证明什么 ​

类别本页有什么不能推出什么
真实组件真实 KRaft broker（3 broker + 3 controller，KafkaClusterTestKit 4.1.2）、真实 KafkaProducer/KafkaConsumer/AdminClient 4.1.2、真实 topic 配置多机部署、网络分区、断电、磁盘故障
合成的部分随机 topic 名、4 秒保留、合成消息、合成事务 id生产命名、生产保留策略、真实业务数据
运行版本4.1.2 运行行为 + 4.1.2 源码摘录3.7.1 的运行行为（机制原码仍是 3.7.1）
未运行RabbitMQ（U58）、RocketMQ（U59）、生产 TLS/SASL、容量压测、跨数据库 EOS（U61/U80）、追加后 ISR 收缩分支不能标为已通过

一个必须写明的实验边界：本实验在构造 KafkaClusterTestKit 时用独立 controller 节点（setCombined(false)）。原因是一次真实的踩坑：最初用 combined 模式（broker 同时是 controller），停掉两个 broker 来制造 ISR 收缩时，controller 仲裁也跟着丢掉了多数派，Admin 请求全部超时——ISR 故障实验和 controller 故障实验不能在同一拓扑上叠加。这不是 Kafka 的缺陷，而是“合并部署的故障域重叠”的现场演示；生产上 controller 与 broker 分离的理由之一就是这种重叠。

教材完成不等于学习者掌握：下面的题目检查“改变条件后能否给出正确裁决”，不是记忆题。

闭卷验收 ​

本页 12 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

工程学习台　阅读讲解，学后自主自测。记录保存在本机浏览器。