Spring Kafka 全局事务 ID 在程序结束后保持打开状态

Question

我正在 Spring Boot 下创建一个 Kafka Spring 生产者，它将数据发送到 Kafka，然后写入数据库；我希望所有这些工作都在一个事务中。我是 Kafka 的新手，也不是 Spring 方面的专家，并且遇到了一些困难。任何指针都非常感谢。

到目前为止，我的代码已成功循环写入 Kafka。我还没有设置 数据库，但已通过在配置中向 producerFactory 添加 transactionIdPrefix 来设置全局事务：

producerFactory.setTransactionIdPrefix("MY_SERVER");

并将@Transactional 添加到 Kafka 发送的方法中。最终我打算用同样的方法来做我的数据库工作。

问题：代码第一次运行良好。但是，如果我停止程序，即使是干净地，我发现代码在我第二次运行它时会挂起，因为它进入 @Transactional 方法。如果我注释掉@Transactional，它会进入方法但挂在kafa模板send()上。

问题似乎出在交易 ID 上。如果我更改前缀并重新运行，程序第一次运行良好，但再次运行时挂起，直到选择新前缀。由于重启后 trans ID 计数器从零开始，如果 trans ID 前缀没有改变，那么重启时将使用相同的 trans ID。

在我看来，原来的 transID 仍然在服务器上打开，并且从未提交过。 （我可以使用控制台消费者从主题中读取数据，但这将读取未提交）。但如果是这样的话，我如何让 spring 提交 trans？我想我的配置一定是错误的。或者——这个问题是否可能是反式 ID 永远无法重复使用？ （在这种情况下，如何解决？）

这是我的相关代码。配置是：

@SpringBootApplication
public class MYApplication {

@Autowired
private static ChangeSweeper changeSweeper;


@Value("${kafka.bootstrap-servers}")
private String bootstrapServers;

@Bean
public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        DefaultKafkaProducerFactory<String, String> producerFactory=new DefaultKafkaProducerFactory<>(configProps); 
        producerFactory.setTransactionIdPrefix("MY_SERVER"); 
        return  producerFactory;
}

@Bean
public KafkaTransactionManager<String, String> KafkaTransactionManager() {
    return new KafkaTransactionManager<String, String>((producerFactory()));
}

@Bean(name="kafkaProducerTemplate")
public KafkaTemplate<String, String> kafkaProducerTemplate() {
    return new KafkaTemplate<>(producerFactory());
}

而进行交易的方法是：

@Transactional
public void send( final List<Record> records) {
    logger.debug("sending {} records; batchSize={}; topic={}", records.size(),batchSize,  kafkaTopic);

    // Divide the record set into batches of size batchSize and send each batch with a kafka transaction:
    for (int batchStartIndex = 0; batchStartIndex < records.size(); batchStartIndex += batchSize ) {
        int batchEndIndex=Math.min(records.size()-1, batchStartIndex+batchSize-1);
        List<Record> nextBatch = records.subList(batchStartIndex, batchEndIndex);
        logger.debug("## batch is from " + batchStartIndex + " to " + batchEndIndex);           

        for (Record record : nextBatch) {

            kafkaProducerTemplate.send( kafkaTopic, record.getKey().toString(), record.getData().toString());   
            logger.debug("Sending> " + record);
        }

// I will put the DB writes here


}

Answer 1

无论我运行多少次，这对我来说都很好（但我必须在本地机器上运行 3 个代理实例，因为默认情况下事务需要这样做）...

@SpringBootApplication
@EnableTransactionManagement
public class So47817034Application {

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

    private final CountDownLatch latch = new CountDownLatch(2);

    @Bean
    public ApplicationRunner runner(Foo foo) {
        return args -> {
            foo.send("foo");
            foo.send("bar");
            this.latch.await(10, TimeUnit.SECONDS);
        };
    }

    @Bean
    public KafkaTransactionManager<Object, Object> KafkaTransactionManager(KafkaProperties properties) {
        return new KafkaTransactionManager<Object, Object>(kafkaProducerFactory(properties));
    }

    @Bean
    public ProducerFactory<Object, Object> kafkaProducerFactory(KafkaProperties properties) {
        DefaultKafkaProducerFactory<Object, Object> factory =
                new DefaultKafkaProducerFactory<Object, Object>(properties.buildProducerProperties());
        factory.setTransactionIdPrefix("foo-");
        return factory;
    }

    @KafkaListener(id = "foo", topics = "so47817034")
    public void listen(String in) {
        System.out.println(in);
        this.latch.countDown();
    }

    @Component
    public static class Foo {

        @Autowired
        private KafkaTemplate<Object, Object> template;

        @Transactional
        public void send(String go) {
            this.template.send("so47817034", go);
        }

    }

}