提交的到底是什么
先把语义钉死，后面所有结论都从它推出来。
消费者提交的位移，是下一条要读的位置，不是「最后一条已处理的位置」。消费了 offset 0–9 这十条，提交的值是 10。这样重启后从 10 开始，不会重复读 9。
提交动作本身在经典协议里由 ConsumerCoordinator 负责。它把 Map<TopicPartition, OffsetAndMetadata> 发给消费组协调器，协调器写进内部 topic __consumer_offsets。6.1 节已经讲过这个 topic 的存储格式，这里只关心客户端侧的三个问题：
- 什么时候提交；
- 提交的是哪个值；
- 提交失败会怎样。
三个问题的答案合起来，解释了一个高频事故：消费者「处理很慢」时丢消息。
自动提交只在 poll 里触发
先看配置文档给人的印象。ConsumerConfig 对 enable.auto.commit 的说明是：
If true the consumer's offset will be periodically committed in the background.
「periodically committed in the background」——周期性、在后台。很多人据此认为有一个后台线程每隔 auto.commit.interval.ms 提交一次，与应用线程无关。
实现不是这样。ConsumerCoordinator.java 第 1198–1206 行：
public void maybeAutoCommitOffsetsAsync(long now) {
if (autoCommitEnabled) {
nextAutoCommitTimer.update(now);
if (nextAutoCommitTimer.isExpired()) {
nextAutoCommitTimer.reset(autoCommitIntervalMs);
autoCommitOffsetsAsync();
}
}
}
这个方法只做三件事：更新计时器、判断是否到期、到期就发一次异步提交。它自己没有循环、没有线程。那谁调用它？
ConsumerCoordinator.poll(Timer, boolean) 的第 575 行：
maybeAutoCommitOffsetsAsync(timer.currentTimeMs());
return true;
而 ConsumerCoordinator.poll 又是被 ClassicKafkaConsumer.poll 里的 updateAssignmentMetadataIfNeeded 调用的。
所以真实链路是：consumer.poll() → ClassicKafkaConsumer.poll → updateAssignmentMetadataIfNeeded → ConsumerCoordinator.poll → maybeAutoCommitOffsetsAsync(now)。
没有 poll()，就没有自动提交。 所谓「后台」，指的是提交请求本身是异步发送的，不是指有一个独立的后台定时任务。
这个区别不是文字游戏。它意味着：如果你的应用在两次 poll() 之间处理了很久，这段时间里自动提交一次都不会发生。下一节用实测证明这一点。
提交的值是 position，不是「处理进度」
autoCommitOffsetsAsync 第 1229–1230 行：
private RequestFuture<Void> autoCommitOffsetsAsync() {
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
allConsumed() 在 SubscriptionState.java 第 786–794 行：
public synchronized Map<TopicPartition, OffsetAndMetadata> allConsumed() {
Map<TopicPartition, OffsetAndMetadata> allConsumed = new HashMap<>();
assignment.forEach((topicPartition, partitionState) -> {
if (partitionState.hasValidPosition())
allConsumed.put(topicPartition, new OffsetAndMetadata(partitionState.position.offset,
partitionState.position.offsetEpoch, ""));
});
return allConsumed;
}
它取的是 partitionState.position.offset。position 在什么时候变？
FetchCollector.fetchRecords 第 178–186 行：
if (nextInLineFetch.nextFetchOffset() > position.offset) {
SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
nextInLineFetch.nextFetchOffset(),
nextInLineFetch.lastEpoch(),
position.currentLeader);
log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`",
position, nextPosition, tp, partRecords.size());
subscriptions.position(tp, nextPosition);
positionAdvanced = true;
}
注意这一段发生在记录被返回给 poll() 调用者之前。日志里写得很直白：「Updating fetch position … and returning N records from poll()」。
于是得到一个关键事实：
position 在
poll()返回记录的那一刻就已经推进到这批记录的末尾，而不是在应用处理完之后。
自动提交取的就是这个 position。所以它提交的是「已经发给应用、但可能一条都没处理」的进度。
实测：处理 3 条，位移提交到 10
实验设计：
- 单分区 topic
b3.lab.autocommit，写入 30 条； - 消费者
enable.auto.commit=true、auto.commit.interval.ms=1000、max.poll.records=10； poll一次拿到 10 条，应用只「处理」其中 3 条；- 然后
sleep 3000，模拟慢处理，期间不调用poll； - 查询已提交位移；
- 再
poll一次，再查询； - 最后
close()，再查询。
查询用 AdminClient.listConsumerGroupOffsets，异步提交需要等响应，所以加了一个最多 5 秒的轮询等待。
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：
poll#1 返回 10 条, position=10, committed=null
应用只处理了 3 条, 进入慢处理 sleep 3000ms
sleep 结束(auto.commit.interval.ms=1000 早已过去), committed=null <- 期间没有 poll(), 自动提交根本没跑
poll#2 返回 10 条, position=20
等异步提交可见后 committed=10 <- 提交发生在 poll#2 进入时, 提交的是 poll#1 之后就已推进的 position
processed=3, committed=10 -> 若此刻崩溃, 位移 3~9 共 7 条从未处理却被标记为已消费
close() 之后 committed=20 <- close() 里的同步自动提交又把 poll#2 返回的 10 条也提交了
五行输出，四个事实：
- 第一次 poll 后 committed 是 null。 新组还没有任何提交。
- sleep 3 秒后仍是 null。
auto.commit.interval.ms=1000早就过了，但没有poll()，自动提交没有触发。这一行直接推翻了「后台每 N 秒提交一次」的理解。 - 第二次 poll 后 committed=10。 提交发生在
poll#2进入时（ConsumerCoordinator.poll里的maybeAutoCommitOffsetsAsync），提交的是poll#1结束时就已经推进到 10 的 position，而不是应用真正处理到的 3。 close()后 committed=20。 关闭时的同步自动提交（ConsumerCoordinator.close第 1021 行的maybeAutoCommitOffsetsSync(timer)）把poll#2返回的 10 条也提交了，尽管它们一条都没处理。
再用命令行确认一次。程序退出后：
docker exec mq-b3 /opt/kafka/bin/kafka-consumer-groups.sh \
--bootstrap-server 127.0.0.1:9092 --describe --group b3-g-autocommit
本机输出：
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
b3-g-autocommit b3.lab.autocommit 0 20 30 10 - - -
CURRENT-OFFSET=20，LOG-END-OFFSET=30，LAG=10。这 20 条里，应用真正处理过的只有 3 条。
丢消息的机制
把上面的数据整理成时间线：
| 时刻 | 动作 | position | committed | 实际处理 |
|---|---|---|---|---|
| t0 | poll#1 返回 10 条 | 10 | null | 0 |
| t1 | 处理 3 条 | 10 | null | 3 |
| t2 | sleep 3s，未 poll | 10 | null | 3 |
| t3 | poll#2 进入，触发自动提交 | 10 | 10 | 3 |
| t4 | poll#2 返回 10 条 | 20 | 10 | 3 |
| t5 | close()，同步自动提交 | 20 | 20 | 3 |
| t6 | 假设进程崩溃 | — | 20 | 3 |
如果崩溃发生在 t6，重启后从 20 开始消费，offset 3–19 这 17 条从未被业务处理却已被标记为消费完成。崩溃点不一定要在 close() 之后：t3 到 t5 之间任何时刻崩溃都会丢掉 offset 3–9 这 7 条，close() 只是把损失扩大到 17 条。
这也解释了为什么这类事故常常发生在「优雅关闭」场景：应用正常退出，以为没有数据损失，实际上关闭时的同步提交把最后一批未处理的数据也提交了。
同步与异步提交的语义差异
自动提交走的是异步路径。手动提交有两个 API，语义不同。
commitSync
ConsumerCoordinator.commitOffsetsSync 第 1142–1178 行：
public boolean commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets, Timer timer) {
invokeCompletedOffsetCommitCallbacks();
if (offsets.isEmpty()) {
// We guarantee that the callbacks for all commitAsync() will be invoked when
// commitSync() completes, even if the user tries to commit empty offsets.
return invokePendingAsyncCommits(timer);
}
long attempts = 0L;
do {
if (coordinatorUnknownAndUnreadySync(timer)) {
return false;
}
RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
client.poll(future, timer);
// We may have had in-flight offset commits when the synchronous commit began. If so, ensure that
// the corresponding callbacks are invoked prior to returning in order to preserve the order that
// the offset commits were applied.
invokeCompletedOffsetCommitCallbacks();
if (future.succeeded()) {
if (interceptors != null)
interceptors.onCommit(offsets);
return true;
}
if (future.failed() && !future.isRetriable())
throw future.exception();
timer.sleep(retryBackoff.backoff(attempts++));
} while (timer.notExpired());
return false;
}
三个特征：
- 阻塞，直到成功、超时或遇到不可重试错误；
- 可重试错误自动重试，间隔是
retryBackoff； - 不可重试错误直接抛异常，例如
CommitFailedException、FencedInstanceIdException。
commitAsync
异步路径在 doCommitOffsetsAsync 第 1099–1129 行。它发一次请求，挂一个监听器，成功或失败都通过回调通知：
private RequestFuture<Void> doCommitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, final OffsetCommitCallback callback) {
RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
inFlightAsyncCommits.incrementAndGet();
方法体后面挂一个监听器，成功时触发拦截器与回调，失败时把可重试异常包装成 RetriableCommitFailedException，遇到 FencedInstanceIdException 则置位 asyncCommitFenced。
关键差异：没有重试循环。失败只是把异常交给回调，是否重试由调用方决定。
为什么不自动重试？commitOffsetsAsync 第 1068–1070 行的注释给了理由：
// we don't know the current coordinator, so try to find it and then send the commit
// or fail (we don't want recursive retries which can cause offset commits to arrive
// out of order). Note that there may be multiple offset commits chained to the same
// coordinator lookup request. This is fine because the listeners will be invoked in
// the same order that they were added. Note also that AbstractCoordinator prevents
// multiple concurrent coordinator lookup requests.
提交乱序会导致小位移覆盖大位移，那比提交失败更危险。所以异步路径选择「不重试，交给调用方」。
混用的坑
commitSync 返回前会调用 invokePendingAsyncCommits，保证之前的异步提交回调都已执行，避免「异步晚到、覆盖同步」。但反过来不成立：先 commitSync(100) 再 commitAsync(50)，50 可能后到并把位移退回去。手动提交时要么全用同步，要么保证异步值单调递增。
提交失败怎么处理
失败分三类，处理方式完全不同。
| 异常 | 含义 | 处理 |
|---|---|---|
RetriableCommitFailedException | 可重试错误，如协调器迁移、网络抖动 | 退避后重试同一批位移 |
CommitFailedException | 再平衡已发生，当前成员不再拥有分区 | 不能重试，应放弃本轮处理，等待重新分配 |
FencedInstanceIdException | 静态成员被同 group.instance.id 的新实例取代 | 停止消费并退出，说明有重复实例 |
自动提交的失败处理最弱。autoCommitOffsetsAsync 第 1233–1245 行的回调：
return commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> {
if (exception != null) {
if (exception instanceof RetriableCommitFailedException) {
log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error.", offsets,
exception);
nextAutoCommitTimer.updateAndReset(rebalanceConfig.retryBackoffMs);
} else {
log.warn("Asynchronous auto-commit of offsets {} failed: {}", offsets, exception.getMessage());
}
} else {
log.debug("Completed asynchronous auto-commit of offsets {}", offsets);
}
});
可重试错误只打 debug 日志并把下次提交提前到 retryBackoffMs 之后；不可重试错误打 warn。两种情况都不抛异常，应用完全感知不到。 如果提交失败但应用继续跑，重启后会重复消费；如果应用以为已经提交并清理了本地状态，重启后可能丢状态。自动提交适合「重复消费可以接受」的场景，需要精确控制时必须关掉。
enable.auto.commit=true 的三种丢失形态
形态一：崩溃点在下一次 poll 之后、本批处理完成之前。 position 已推进但尚未提交时崩溃，只是重复消费——committed 还停在旧值。真正丢失需要等下一次 poll 触发提交。
形态二：下一次 poll 提前提交了上一批。 这正是实测展示的：处理 3 条，第二次 poll 把 10 提交了。只要应用在「处理完」之前调用下一次 poll()，就会发生。多线程处理、把记录丢进队列后立刻继续 poll，都属于这一类。
形态三：close() 的同步提交。 ConsumerCoordinator.close 第 1017–1021 行：
public void close(final Timer timer, CloseOptions.GroupMembershipOperation membershipOperation) {
// we do not need to re-enable wakeups since we are closing already
client.disableWakeups();
try {
maybeAutoCommitOffsetsSync(timer);
maybeAutoCommitOffsetsSync 会把当前 allConsumed() 同步提交一次。如果最后一批还没处理完就关闭，这批就被标记为已消费。
三种形态的共同点是：提交的值来自 position，而 position 与「业务处理完成」之间没有因果关系。 客户端不知道你处理到哪，它只知道发给你到哪。
要「至少一次」该怎么写
结论很简单：关掉自动提交，处理完之后同步提交处理过的位移。
Properties props = new Properties();
props.put("bootstrap.servers", "127.0.0.1:19213");
props.put("group.id", "b3-g-manual");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");
props.put("max.poll.records", "100");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(List.of("b3.lab.autocommit"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
if (records.isEmpty()) continue;
Map<TopicPartition, OffsetAndMetadata> done = new HashMap<>();
for (ConsumerRecord<String, String> record : records) {
handle(record); // 业务处理失败会抛异常，不会进入下面的 put
done.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1));
}
consumer.commitSync(done);
}
}
要点有三条：
- 提交
record.offset() + 1，因为提交的是下一条要读的位置； - 只把处理成功的分区放进
done，处理失败的分区不提交，下次会重读； commitSync在循环内，每批处理完提交一次。提交频率由max.poll.records决定，不由时间决定。
这个写法给出的是至少一次：崩溃会重复，但不会丢。要减少重复，可以把「业务写入」和「位移提交」放进同一个事务——那是卷二的话题，这里不展开。
如果吞吐要求高，可以用 commitAsync 加回调重试，但要自己保证提交值单调递增，并接受「关闭前最后一次异步提交可能没完成」。
一个容易忽略的细节
allConsumed() 只包含 hasValidPosition() 为真的分区。刚分配、还没拉取到位点信息的分区不会出现在提交里，所以再平衡刚完成时立刻提交，可能只提交了部分分区。另外，OffsetAndMetadata 里的 leaderEpoch 用于日志截断检测：如果 broker 返回的 epoch 比提交的小，说明发生了截断，客户端会触发 OffsetOutOfRange 或重置。14.3 节会用到。
本节判据
- 自动提交是后台定时任务吗？ 答：不是。它只在
poll()进入时由ConsumerCoordinator.poll调用，没有poll就不会提交。 - 自动提交提交的是哪个值？ 答：
position，即下一条要读的位置；而 position 在poll返回记录时就已推进。 commitSync和commitAsync的核心差异？ 答：同步会重试并可能抛异常；异步只发一次、通过回调报告，不自动重试，以避免提交乱序。- 自动提交在什么情况下丢消息？ 答：下一次
poll或close()把「已发出但未处理完」的记录提交掉，之后崩溃，这些记录不会再被消费。
下一节离开消费组，进入序列化与扩展点。9.1 会讲内置序列化器、Schema 边界，以及为什么不要在消息里放可变对象。
阅读导航：上一节 8.2 Fetcher 与拉取管线 ｜ 下一节 9.1 序列化器与 Schema 边界
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。