时间戳

首页 > 解决方案 > 无法使用ChainedKafkaTransaction 同步 Kafka 和 MQ 事务

java - 无法使用ChainedKafkaTransaction 同步 Kafka 和 MQ 事务

问题描述

我们有一个 Spring Boot 应用程序，它使用来自 IBM MQ 的消息进行一些转换并将结果发布到 Kafka 主题。我们为此使用https://spring.io/projects/spring-kafka 。我知道 Kafka 不支持 XA；但是，在文档中，我发现了一些关于使用 aChainedKafkaTransactionManager链接多个事务管理器并同步事务的输入。相同的文档还提供了一个示例，说明如何在从 Kafka 读取消息并将它们存储在数据库中时同步 Kafka 和数据库。

我在我的 se 案例中遵循相同的示例，并将JmsTransactionManagerwith链接到KafkaTransactionManagera 的保护伞下ChainedKafkaTransactionManager。bean定义如下：

@Bean({"mqListenerContainerFactory"})
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(this.connectionFactory());
    factory.setTransactionManager(this.jmsTransactionManager());
    return factory;
}

@Bean
public JmsTransactionManager jmsTransactionManager() {
    return new JmsTransactionManager(this.connectionFactory());
}

@Bean("chainedKafkaTransactionManager")
public ChainedKafkaTransactionManager<?, ?> chainedKafkaTransactionManager(
        JmsTransactionManager jmsTransactionManager, KafkaTransactionManager kafkaTransactionManager) {

    return new ChainedKafkaTransactionManager<>(kafkaTransactionManager, jmsTransactionManager);
}

@Transactional(transactionManager = "chainedKafkaTransactionManager", rollbackFor = Throwable.class)
@JmsListener(destination = "${myApp.sourceQueue}", containerFactory = "mqListenerContainerFactory")
public void receiveMessage(@Headers Map<String, Object> jmsHeaders, String message) {
    // Processing the message here then publishing it to Kafka using KafkaTemplate
    kafkaTemplate.send(sourceTopic,transformedMessage);

    // Then throw an exception just to test the transaction behaviour
    throw new RuntimeException("Not good Pal!");
}

运行应用程序时，发生的事情是消息不断回滚到 MQ 队列中，但消息在 Kafka 主题中不断增长，这对我来说意味着 kafkaTemplate 交互不会回滚。

如果我根据文档很好地理解了这不应该是这种情况。“如果事务处于活动状态，则在事务范围内执行的任何 KafkaTemplate 操作都使用事务的 Producer。”

在我们的 application.yaml 中，我们通过设置将 Kafka 生产者配置为使用事务spring.kafka.producer.transaction-id-prefix

问题是我在这里缺少什么以及我应该如何解决它。预先感谢您的意见。

标签： javaspring-bootapache-kafkatransactionsspring-kafka

解决方案

消费者默认可以看到未提交的记录；将isolation.level消费者属性设置read_committed为避免接收来自回滚事务的记录。

推荐阅读