如何在 Spring Boot 应用程序中使用 Spring 事务配置 kafka 事务管理器答案

【问题标题】：How to configure kafka transaction manager with spring transaction in spring boot app如何在 Spring Boot 应用程序中使用 Spring 事务配置 kafka 事务管理器
【发布时间】：2022-01-27 02:53:30
【问题描述】：

我在 Spring Boot 应用程序中使用 Kafka。我想在一个事务中执行操作，如下所示。

listen(){
 produce()
 saveInDb()
}

和

operation(){
 saveInDB()
 produce()
}

我已使用以下配置启用 Kafka 事务

spring:
  kafka:
    bootstrap-servers: localhost:19092,localhost:29092,localhost:39092
    producer:
      transaction-id-prefix: tx-
    consumer:
      enable-auto-commit: false
      isolation-level: read_committed

并使用此配置

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(props);
        factory.setTransactionIdPrefix("tx-");
        return factory;
    }

    @Bean
    public KafkaTransactionManager kafkaTransactionManager() {
        KafkaTransactionManager manager = new KafkaTransactionManager(producerFactory());
        return manager;
    }
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

但是尝试使用 spring @Transactional 注释时出现错误

@Transactional
operation(){
 saveInDB()
 produce()
}

No bean named 'transactionManager' available: No matching TransactionManager bean found for qualifier 'transactionManager' - neither qualifier match nor bean name match!

我在这里关注了春季文档 https://docs.spring.io/spring-kafka/reference/html/#using-kafkatransactionmanager

我在配置中遗漏了什么？

【问题讨论】：

您是否尝试过使用@Transactional("kafkaTransactionManager")？
是的，但出现同样的错误

标签： java spring-boot apache-kafka spring-kafka

【解决方案1】：

我错过了在配置中定义 transactionManager bean。 Spring 找不到它，因为 KafkaTransactionManager 扩展了 AbstractPlatformTransactionManager 和 JpaTransactionManager 也扩展了同一个类。

将此 bean 定义为主 bean 解决了这个问题。

    @Bean
    @Primary
    public JpaTransactionManager transactionManager(EntityManagerFactory entityManagerFactory) {
        return new JpaTransactionManager(entityManagerFactory);
    }

Kafka 事务与其链接，KafkaTemplate 将与事务管理器同步事务。

参考https://docs.spring.io/spring-kafka/reference/html/#transactions

【讨论】：