Spring Cloud Stream Kafka Binder KafkaTransactionManager 在应用程序上下文中产生循环

Question

我正在使用 Kafka 设置一个基本的 Spring Cloud Stream 生产者。目的是接受 HTTP POST，使用 Spring Data JPA 将 post 的结果保存到数据库，然后使用 Spring Cloud 将结果写入 Kafka 主题流卡夫卡活页夹。我正在关注有关如何设置 KafkaTransactionManager 的最新活页夹 documentation，但此代码会导致应用程序启动时出错。

***************************
APPLICATION FAILED TO START
***************************

Description:

The dependencies of some of the beans in the application context form a cycle:

┌─────┐
|  kafkaTransactionManager defined in com.example.tx.Application
↑     ↓
|  org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration
└─────┘

我在我的应用程序 class 中定义了以下 Bean，这与文档相同。

@Bean
public KafkaTransactionManager kafkaTransactionManager(BinderFactory binders) {

    ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null, MessageChannel.class)).getTransactionalProducerFactory();
    KafkaTransactionManager tm = new KafkaTransactionManager<>(pf);
    tm.setTransactionIdPrefix("tx-test");
    return tm;
}

似乎调用 getBinder 导致 Spring 再次创建上下文。我该如何解决这种循环依赖？

依赖项：Spring引导父 2.4.6； Spring 云物料清单 2020.0.3

Answer 1

其中一层一定发生了变化；这是一个解决方法：

@Bean
SmartInitializingSingleton ktmProvider(BinderFactory binders, GenericApplicationContext context) {
    return () -> {
        context.registerBean("kafkaTransactionManager", KafkaTransactionManager.class,
                ((KafkaMessageChannelBinder) binders.getBinder(null, MessageChannel.class))
                        .getTransactionalProducerFactory());
        context.getBean(KafkaTransactionManager.class).setTransactionIdPrefix("tx-test");
    };
}

即在注册和配置 tm 之前等待创建其他 bean。

Spring Cloud Stream Kafka Binder KafkaTransactionManager 在应用程序上下文中产生循环

Spring Cloud Stream Kafka Binder KafkaTransactionManager results in a cycle in application context

spring-boot

spring-cloud-stream

spring-kafka

spring-cloud-stream-binder-kafka