不应该确认消息会导致它被重新传递给 Kafka 消费者吗？

Question

我正在运行一个 Spring Cloud Stream 应用程序，我在其中从 Kafka 主题读取事务，处理事务，然后将它们发送到 IBM MQ。当与 IBM MQ 没有连接以防止事务丢失时，我正在尝试处理该错误。在这种情况下，jms 模板将抛出异常，并且流侦听器不会提交事务。预期的行为是事务保留在 Kafka 主题中，并且流侦听器再次读取它。然而，消息似乎只被消费一次，并且没有发生“回滚”。为此，这是我的配置：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          input:
            consumer:
              auto-commit-offset: false
      bindings:
        input:
          destination:  kafka_topic
          brokers: localhost:9092

这里是代码：

    public void handleMessage(Message<TransactionMessage> request,  @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {
        TransactionMessage message = request.getPayload();
        System.out.println("Consumed a message");
        try {
            executionFlow.execute(message); // here the jmsTemplate throws an exception
            System.out.println("doing the ack");
            acknowledgment.acknowledge();
        }
        catch (RuntimeException e) {
            System.out.println("did not send to MQ");
        }
    }

executionFlow调用的jmsTemplate代码：

    public void sendMessage(String messageTarget) {

        System.out.println("i am trying to send to MQ");
        try {
            jmsTemplate.convertAndSend(destinationTopicQueue, messageTarget);
        } catch (Exception e) {
            throw new RuntimeException("jmsTemplate failed to send to IBM MQ");
        }
    }

这是我关闭与 IBM MQ 的连接时的输出：

Consumed a message
i am trying to send to MQ
did not send to MQ

Answer 1

   catch (RuntimeException e) {
        System.out.println("did not send to MQ");
    }

你需要重新抛出异常才能导致回滚。

您还需要在活页夹中启用 Kafka 事务。

见Kafka Binder Properties。

spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix

在活页夹中启用事务。请参阅 Kafka 文档中的 transaction.id 和 spring-kafka 文档中的 Transactions。启用事务后，将忽略单个生产者属性，所有生产者都使用 spring.cloud.stream.kafka.binder.transaction.producer.* 属性。

默认 null（无交易）

如果你也向kafka发送数据，你需要一个事务生产者

spring.cloud.stream.kafka.binder.transaction.producer.*

事务绑定器中生产者的全局生产者属性。请参阅 spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix 和 Kafka Producer Properties 以及所有 binder 支持的一般 producer properties。

默认值：查看各个生产者属性。