Kafka - 如何在 Producer 类中获取失败消息的详细信息答案

【问题标题】：Kafka - How to obtain failed messages details in Producer classKafka - 如何在 Producer 类中获取失败消息的详细信息
【发布时间】：2018-01-13 16:29:57
【问题描述】：

Kafka 允许通过 Producer (KafkaProducer) 类的以下方法发送异步消息：

public java.util.concurrent.Future<RecordMetadata> send(ProducerRecord<K,V> record) 
public java.util.concurrent.Future<RecordMetadata> send(ProducerRecord<K,V> record, Callback callback)

可以通过
处理成功 1) Future<RecordMetaData> 对象或
2) 回调调用的onCompletion 方法。 onCompletion的完整方法签名和用法如下（taken from kafka docs）

ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);  
producer.send(record,
               new Callback() {
                   public void onCompletion(RecordMetadata metadata, Exception e) {
                       if(e != null)
                           e.printStackTrace();
                       System.out.println("The offset of the record we just sent is: " + metadata.offset());
                   }
               });

而失败需要通过Exception e传递给onCompletion方法来处理

到目前为止，一切看起来都不错。

但如果我做对了，可以从exception 或e 对象获得的任何合理信息都是堆栈跟踪和异常消息。我在这里要指出的是，e 不包含发送的实际记录的任何信息。或者换句话说，它不包含对发送给 kafka 代理的实际 record 的引用。那么如果record没有发送成功，生产者可以做哪些有用的处理或处理。真的不多。为什么我这么说 - 理想情况下，我想在某个地方记录失败的消息，然后尝试重新发送它。但是由于框架提供的信息很少（e），我觉得这是不可能的。

谁能指出我是对还是错？

【问题讨论】：

作为注释，我想提一下，我并不是说异常e 没有帮助。这对于调试和查找错误原因肯定是有帮助的。但我不确定它是否包含足够的信息来确定哪个record 失败了，所以我可以重新发送或记录record
试一试以下版本的回调类：class CallbackFunc implements Callback { private int recordId;公共 CallbackFunc(int recordId) { this.recordId = recordId; } @Override public void onCompletion(RecordMetadata metadata, Exception e) { if(e != null) { Log.error("向Kafka写入消息时出错。记录ID：" + recordId); } } }

标签： java apache-kafka kafka-producer-api

【解决方案1】：

您可以轻松地创建一个接收producerRecord 作为构造函数参数的回调。所以在onCompletion上，除了例外，你可以完全了解生产者记录，甚至可以尝试再次发送。

我处理了同样的问题。创建了一个获取producerRecord 的回调和一个使用执行程序服务再次发送记录的回调处理程序。所以最终，我可以容忍任意数量的故障（例如网络问题或 kafka 宕机），并从中恢复。

【讨论】：