Spring Kafka 手动提交偏移量失效：同步提交未生效的解决方案-java教程-PHP中文网

Spring Kafka 手动提交偏移量失效：同步提交未生效的解决方案

当使用 Spring Kafka 配置 MANUAL_IMMEDIATE + syncCommits=true 且禁用自动提交时，若未调用 acknowledge()，预期应重复消费同一条消息，但实际却跳过——根本原因在于 Kafka Consumer 的偏移量管理机制：它默认在内存中维护当前消费位置，而非实时回查集群存储的 offset；未手动 seek 就不会重读旧消息。

当使用 spring kafka 配置 `manual_immediate` + `synccommits=true` 且禁用自动提交时，若未调用 `acknowledge()`，预期应重复消费同一条消息，但实际却跳过——根本原因在于 kafka consumer 的偏移量管理机制：它默认在内存中维护当前消费位置，而非实时回查集群存储的 offset；未手动 seek 就不会重读旧消息。

在 Spring Kafka 中，禁用自动提交（enable-auto-commit=false）并设置 AckMode.MANUAL_IMMEDIATE 并不等于“失败即重试”。它仅表示：是否提交 offset 完全由你显式调用 Acknowledgment.acknowledge() 控制；但 Kafka Consumer 实例本身仍会按 poll() 逻辑持续推进其内存中的拉取位置（position）。一旦下一次 poll() 被触发（例如因心跳或轮询超时），Consumer 将从内存 position 继续读取——即使上一条消息未被 acknowledge，也不会自动倒回。

✅ 正确做法是：在业务处理失败时，主动调用 Acknowledgment.seek() 强制重置消费位置。注意：seek() 必须在监听器方法内、且在当前 poll() 返回的记录范围内执行（即对当前批次中的分区-偏移量进行 seek）：

@KafkaListener(topics = "my-topic", groupId = "test")
public void listen(String data, Acknowledgment ack, Consumer, ?> consumer) {
    try {
        // 执行业务逻辑（可能失败）
        process(data);
        ack.acknowledge(); // 成功：提交 offset，继续下一条
    } catch (Exception e) {
        // 失败：获取当前记录的分区与偏移量，并 seek 回该位置
        ConsumerRecord, ?> record = ... // 需通过 ListenerUtils 或自定义参数获取（见下方说明）
        TopicPartition partition = new TopicPartition(record.topic(), record.partition());
        consumer.seek(partition, record.offset()); // 关键：强制重读本条
        // 注意：此时不应调用 ack.acknowledge()
    }
}

⚠️ 重要补充：

Spring Kafka 2.8+ 支持在监听器方法中直接注入 ConsumerRecord，从而安全获取 record.offset() 和 record.partition()：

@KafkaListener(topics = "my-topic")
public void listen(ConsumerRecord<string string> record, Acknowledgment ack, Consumer, ?> consumer)</string>

seek() 操作仅影响当前 Consumer 实例对该分区的下一次 poll()，且必须在本次监听方法执行期间完成（不能异步延迟调用）。
若使用 ConcurrentKafkaListenerContainerFactory 且 concurrency > 1，需确保 seek() 不跨线程误操作其他分区；单并发（concurrency=1）是最简可控场景，符合你的配置。
setSyncCommits(true) 仅保证 acknowledge() 底层调用的是 commitSync()（阻塞等待 Broker 确认），与是否重试无关；它解决的是提交可靠性，而非消费位置控制。