Spring Kafka 事务 - 没有事务在处理，在 template.executeInTransaction() 范围内运行模板操作

Question

我正在尝试使用 KafkaTemplate 将消息从事务发布到 Kafka：

@Autowired
KafkaTemplate<GenericRecord, GenericRecord> kafkaTemplate;

@Transactional
@RabbitListener(queues = "queueName")
void input(final List<Message> messages) {
     for (Message msg : messages) {
          PublishRequest request = prepareRequest(msg);
          kafkaTemplate.sendDefault(request.getKey(), reguest.getValue());
     }
     transactionalDatabaseInserts();
}

但是当我这样做时，我得到了这个异常：

原因：java.lang.IllegalStateException：没有事务在 过程;可能的解决方案：在 template.executeInTransaction() 操作的范围，开始一个 在调用模板方法之前使用@Transactional 进行事务， 在使用一个侦听器容器启动的事务中运行 记录

KafkaTemplate 的配置：

@EnableTransactionManagement
@Configuration
public class KafkaConfig{
    @Bean
    KafkaTransactionManager<GenericRecord, GenericRecord> kafkaTransactionManager(final ProducerFactory<GenericRecord, GenericRecord> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }

    @Bean
    KafkaTemplate<GenericRecord, GenericRecord> kafkaTemplate(final ProducerFactory<GenericRecord, GenericRecord> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

在我的application.yaml 中，我包含了：

spring.kafka.producer.transaction-id-prefix: tx-

我希望我的方法适用于 @Transactional 而不是 kafkaTemplate.executeInTransaction()。为什么我会遇到这个异常？

Answer 1

您一定有一些错误配置 - 这按预期工作......

@SpringBootApplication
@EnableTransactionManagement
public class So63596919Application {

    public static void main(String[] args) {
        SpringApplication.run(So63596919Application.class, args);
    }

    @Autowired
    private KafkaTemplate<String, String> template;

    private final CountDownLatch latch = new CountDownLatch(1);

    @Transactional
    @RabbitListener(queues = "so63596919")
    public void listen(List<String> in) throws InterruptedException {
        System.out.println(in);
        in.forEach(str -> this.template.send("so63596919", str));
        System.out.println("Hit enter to exit listener and commit transaction");
        this.latch.await();
    }

    @KafkaListener(id = "so63596919", topics = "so63596919")
    public void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public Queue queue() {
        return new Queue("so63596919");
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so63596919").partitions(1).replicas(1).build();
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate template, AbstractRabbitListenerContainerFactory<?> factory) {
        factory.setBatchListener(true);
        factory.setContainerCustomizer(container -> {
                ((SimpleMessageListenerContainer) container).setConsumerBatchEnabled(true);
                container.setDeBatchingEnabled(true);
        });
        return args -> {
            template.convertAndSend("so63596919", "foo");
            template.convertAndSend("so63596919", "bar");
            System.in.read();
            this.latch.countDown();
        };
    }

}

spring.kafka.producer.transaction-id-prefix: tx-
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.isolation.level=read_committed

spring.rabbitmq.listener.simple.batch-size=2

如果你能把你的项目精简成这样一个小例子，我可以看看有什么问题。