kafka事务生产者示例

Question

我从源主题收到一条消息。然后我将消息分成3个部分，并将每个部分发送到3个不同的主题。现在有2条消息成功传递到第2个主题。但是在发送第3条消息时，我们会收到异常（例如ProducerFencedException|OutOfOrderSequenceException|AuthorizationException|RecordLengthException）

如何回滚/还原前2个主题中的其他2条消息？

完整Java代码示例将非常有用。不想使用producer.initTransaction（）类型的方法。

我也提到了这一点——事务性Kafka生产者，但我怀疑我们真的需要——为生产者、模板、工厂、Tx编写所有@Bean，因为这些可以很容易地在application.yml中提供。

Answer 1

请参阅文档https://docs.spring.io/spring-kafka/docs/current/reference/html/#transactions

当设置transactionIdPrefix属性时，Spring Boot将自动将KafkaTransactionManager bean配置到容器工厂中；然后，容器将启动事务，如果任何发布失败，它们都将回滚。

@Bean
@ConditionalOnProperty(name = "spring.kafka.producer.transaction-id-prefix")
@ConditionalOnMissingBean
public KafkaTransactionManager<?, ?> kafkaTransactionManager(ProducerFactory<?, ?> producerFactory) {
    return new KafkaTransactionManager<>(producerFactory);
}