Spring Integration - Kafka 消息驱动通道 - 自动确认

Question

我使用了 spring io 文档中列出的示例配置，它工作正常。

<int-kafka:message-driven-channel-adapter
        id="kafkaListener"
        listener-container="container1"
        auto-startup="false"
        phase="100"
        send-timeout="5000"
        channel="nullChannel"
        message-converter="messageConverter"
        error-channel="errorChannel" />

但是，当我使用下游应用程序测试它时，我从 kafka 消费并将其发布到下游。如果下游已关闭，则消息仍在被消耗且未重播。

或者说从 kafka 主题消费后，如果我在服务激活器中发现一些异常，我也想抛出一些异常，该异常应该回滚事务，以便可以重播 kafka 消息。

简而言之，如果消费应用程序有问题，那么我想回滚事务，以便消息不会被自动确认并一次又一次地重播，除非它被成功处理。

Answer 1

这不是 Apache Kafka 的工作方式。有类似于 JMS 的 TX 语义。 Kafka 主题中的偏移量与回调或重新交付无关。

我建议您从他们的官方资源中更深入地研究 Apache Kafka。

Spring Kafka 没有提供常规 Apache Kafka 协议，但是您可以考虑使用 Spring Kafka 中的重试功能在本地重新传递相同的记录：http://docs.spring.io/spring-kafka/docs/1.2.2.RELEASE/reference/html/_reference.html#_retrying_deliveries

是的，ack模式必须是MANUAL，消费后不要自动提交offset到Kafka中。