如何在具有事务的反应式生产者中从无法访问的Kafka中恢复？

Question

我有一个简单的反应式Kafka生产者，我需要它来手动提交事务。在Reactor Kafka for method begin（）中的javadoc之后，我创建了一个producer方法

@Autowired
private ReactiveKafkaProducerTemplate<String, String> kafkaTemplate;

public Mono<Void> send(String message) {
    return kafkaTemplate.transactionManager()
            .begin()
            .then(kafkaTemplate.send("my.test.topic", message))
            .then(kafkaTemplate.transactionManager().commit());
}

其配置为

@Bean
public ReactiveKafkaProducerTemplate<String, String> reactiveKafkaProducerTemplate() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "http://localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "producer-tx-1");

    return new ReactiveKafkaProducerTemplate<>(SenderOptions.create(props));
}

它运行良好，我正在向Kafka发送消息。但现在我需要保护它，使其免受我们公司潜在Kafka/网络中断的影响。为了模拟这种情况，我停止了我的Kafkadocker，发送了一条消息，一段时间后再次启动Kafka。但奇怪的事情正在发生。我遇到了三种可能的情况：

1. 当我在大约一分钟内启动Kafka时，制作人恢复了连接和事务，并很好地完成了发送

org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for my.test.topic-0:219057 ms has passed since batch creation

此状态不可恢复，每次后续尝试都会失败

org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state

我必须重新启动应用程序

TransactionalId producer-tx-1: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION

此状态也无法恢复，我必须重新启动应用程序。

如何从这种情况下恢复？我想实现的是生产者等待f. e.20秒，如果连接没有重新建立，那么就抛出异常并丢弃请求。最重要的是，系统不应该因为无法访问Kafka而崩溃。

起初，我认为需要使用kafkaTemplate手动中止事务。transactionManager（）。abort（）在某些中。doOnError（）但这不起作用（在场景3中为f.e.，代码不会引发任何异常）。然后，我玩了一下ProducerConfig，尝试了TRANSACTION\u TIMEOUT\u CONFIG、REQUEST\u TIMEOUT\u MS\u CONFIG或DELIVERY\u TIMEOUT\u MS\u CONFIG，但请求没有超时，不确定我做错了什么。

Answer 1

>

之后，您将需要创建一个新的生产者并处理旧的生产者。