java - 如何暂停/启动/停止Kafka Producer/Kafka模板

Question

我正在使用带有 kafka 集成的 Spring Boot 应用程序，我想实现一个端点来停止和启动 kafka 发布消息。该消息由另一个端点以异步方式触发。

bean KafkaTemplate<String, String>或ProducerFactory<String, String> producerFactory()没有任何停止和暂停操作。

我的目标是能够模拟连接失败并确保这些消息存储在我现有的后备机制中。

有什么想法吗？

Answer 1

KafkaTemplate没有这些回调，因为它是一个被动组件，只有我们调用它时才能执行这些操作。

为了模拟连接失败，我建议您实现自定义ProducerFactory生产KafkaProducer被 mock 或覆盖Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); 。在那里，在那ProducerFactory您可以实现适当的生命周期回调并对提到的 send() 中的状态使用react实现。

org.apache.kafka.clients.producer.MockProducer可能有一些东西可供您重复使用或借用。例如查看其 close()或fenceProducer() .