如何为kafka-python kafka.KafkaProducer#send()添加失败回调？

Question

我想设置一个回调，以便在生成的记录失败时触发。最初，我只想记录失败的记录。

Confluent Kafka python 库提供了一种添加回调的机制：

produce(topic[, value][, key][, partition][, on_delivery][, timestamp])
...
    on_delivery(err,msg) (func) – Delivery report callback to call (from poll() or flush()) on successful or failed delivery

如何使用 kafka 实现类似的行为-python kafka.KafkaProducer#send() without having to use the deprecated SimpleClient using kafka.SimpleClient#send_produce_request()

Answer 1

虽然没有记录，但这相对简单。每当您发送消息时，您都会立即收到 Future 回复。您可以将 callbacks/errback 附加到 Future:

F = producer.send(topic=topic, value=message, key=key)
F.add_callback(callback, message=message, **kwargs_to_pass_to_callback_method)
F.add_errback(erback, message=message, **kwargs_to_pass_to_errback_method)

相关源码在这里： https://github.com/dpkp/kafka-python/blob/1937ce59b4706b44091bb536a9b810ae657c3225/kafka/future.py#L48-L64

我们真的应该记录下来，我提交了 https://github.com/dpkp/kafka-python/issues/1256 来跟踪它。

如何为kafka-python kafka.KafkaProducer#send()添加失败回调？

How to add a failure callback for kafka-python kafka.KafkaProducer#send()?

kafka-python