如何在Camel Kafka中实现已消费消息事务回滚并避免重复消费？

你遇到的问题核心是：发送到topic2失败时异常未被正确处理，导致broker2恢复后，消费者不仅重试未提交偏移量的目标消息，还意外重复消费了已提交偏移量的旧消息。结合你的路由代码和配置，以下是具体分析和修复方案：

问题根源 #

• 当前路由中，手动提交偏移量的逻辑放在to("kafka:topic2")之后，若发送到topic2抛出异常，提交逻辑根本不会执行，这两条消息的偏移量始终未提交——这部分重试是符合预期的。
• 已提交偏移量的旧消息被重复消费，大概率是因为发送失败时的异常触发了消费者重新平衡，或是客户端异常重启/初始化时，auto-offset-reset: latest配置未按预期生效，导致偏移量状态混乱。

修复方案 #

1. 完善路由的异常处理逻辑 #

修改路由，添加异常捕获，确保发送失败时不提交偏移量，同时通过Camel错误处理机制控制重试，避免异常扩散引发消费者异常重启：

from("kafka:topic1?groupId=topic1&broker=broker1")
    .process(<business-logic>)
    // 捕获发送到topic2的所有异常
    .onException(Exception.class)
        .handled(true) // 标记异常已处理，避免扩散到Kafka客户端
        .delay(5000) // 延迟5秒重试，避免频繁重试压垮系统
        .maximumRedeliveries(-1) // 无限重试直到发送成功
        .log("发送topic2失败，将重试: ${exception.message}")
    .end()
    .to("kafka:topic2?broker=broker2")
    // 仅当发送成功时提交偏移量
    .process(exchange -> {
        KafkaManualCommit kafkaManualCommit = exchange.getIn().getHeader(KafkaConstants.MANUAL_COMMIT, KafkaManualCommit.class);
        if (kafkaManualCommit != null) {
            kafkaManualCommit.commit();
        }
    });

2. 优化Kafka消费者配置 #

调整配置，确保消费者行为符合预期：

• 保留auto-commit-enable: false和allow-manual-commit: true，这是手动提交偏移量的基础。
• 补充配置避免不必要的重新平衡，同时限制单次拉取消息数量，缩小重复消费范围：

break-on-first-error: true
allow-manual-commit: true
auto-commit-enable: false
auto-offset-reset: latest
max-poll-records: 1
session.timeout.ms: 30000
heartbeat.interval.ms: 10000

3. 保证业务逻辑幂等性 #

由于你是跨Kafka broker操作（topic1在broker1，topic2在broker2），无法通过Kafka事务保证原子性，建议：

• 让业务处理逻辑具备幂等性：比如给消息添加唯一ID，处理前先检查该ID是否已被处理过（可记录到Redis或数据库），避免重复处理产生副作用。

验证步骤 #

1. 关闭broker2，向topic1发送2条消息，此时路由会捕获发送异常并进入重试等待。
2. 启动broker2后，路由自动重试发送这2条消息，成功后提交偏移量。
3. 确认已提交偏移量的旧消息不会被重复消费。

内容的提问来源于stack exchange，提问作者dark ninja