你的问题核心在于：消费者容器只关联了@Primary的事务管理器，未标记@Primary的生产者无法加入到容器启动的事务上下文，所以抛出“未处于事务进程中”的异常。要让多个生产者都支持EOS事务，需要用ChainedKafkaTransactionManager整合多个事务管理器，让容器能同时管理所有生产者的事务。

解决方案步骤 #

1. 配置链式事务管理器 #

创建一个新的配置类，把两个生产者的事务管理器整合到链式事务管理器中：

@Configuration
public class KafkaTransactionConfig {

    @Autowired
    @Qualifier("transactionalKafkaTransactionManagerNonAvro")
    private KafkaAwareTransactionManager<?, ?> nonAvroTxManager;

    @Autowired
    @Qualifier("transactionalKafkaTransactionManagerAvro")
    private KafkaAwareTransactionManager<?, ?> avroTxManager;

    @Bean
    public ChainedKafkaTransactionManager<?, ?> chainedKafkaTransactionManager() {
        // 顺序不影响事务一致性，按需求排列即可
        return new ChainedKafkaTransactionManager<>(nonAvroTxManager, avroTxManager);
    }
}

2. 修改消费者容器工厂，使用链式事务管理器 #

在KafkaConsumerConfig中，把原来绑定单个事务管理器的代码替换为链式事务管理器：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
    // 注入链式事务管理器
    ChainedKafkaTransactionManager<?, ?> chainedKafkaTransactionManager) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
    factory.setConsumerFactory(consumerFactory());
    factory.setAutoStartup(false);
    factory.setConcurrency(2);
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.BATCH);
    factory.getContainerProperties().setEosMode(EOSMode.ALPHA);
    // 替换为链式事务管理器
    factory.getContainerProperties().setTransactionManager(chainedKafkaTransactionManager);
    return factory;
}

3. 额外注意事项 #

• 确保Transactional ID唯一：你当前两个生产者的TRANSACTIONAL_ID_CONFIG分别是eventsannonavro和eventsanavro，这个是正确的，必须保持每个生产者的事务ID唯一，避免事务冲突。
• 移除手动flush操作：你的KafkaTopicProducer中的kafkaAvroFlush()和kafkaNonAvroFlush()可以移除，因为Spring Kafka在事务提交时会自动触发flush，手动调用可能导致事务外的消息发送，破坏EOS语义。
• 保持幂等性配置：继续保留ENABLE_IDEMPOTENCE_CONFIG=true和ACKS_CONFIG=all，这是EOS的基础配置。

原理说明 #

ChainedKafkaTransactionManager是Spring Kafka提供的专门用于管理多个生产者事务的组件，它会将多个事务管理器的操作串联起来：

1. 当容器启动事务时，链式管理器会同时启动所有关联生产者的事务。
2. 消息处理完成后，链式管理器会依次提交所有生产者的事务；如果处理过程中抛出异常，会依次回滚所有事务。
3. 这样就保证了两个生产者的发送操作都在同一个事务上下文里，实现Exactly Once语义。

验证方法 #

1. 重启应用，触发消息消费，同时调用两个生产者发送消息。
2. 查看Kafka的事务日志（通过Kafka命令行工具kafka-transactions.sh），确认两个生产者的事务都被正确提交/回滚。
3. 模拟异常场景（比如在消息处理中抛出RuntimeException），检查两个Topic中是否都没有产生消息（事务回滚成功）。

内容的提问来源于stack exchange，提问作者Aleshan