如何在 Kafka Producer 中启用 onFailure 回调？答案

【问题标题】：How to enable onFailure call back in Kafka Producer?如何在 Kafka Producer 中启用 onFailure 回调？
【发布时间】：2022-01-26 22:13:14
【问题描述】：

我正在写一个 kafka 生产者，下面是我的课程。

public class KafkaPublisher {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @Autowired
    KafkaProducer producer;

    public void pushKafkaNotification(String partitionKey, String serializedKafkaNotification, String topic)
    {
        try {
            producer.beginTransaction();
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String,String>(topic,partitionKey,serializedKafkaNotification);
            Future<RecordMetadata> result = producer.send(producerRecord);
            producer.commitTransaction();
        }
        catch (Exception e)
        {
            producer.abortTransaction();
            log.error("Exception while pushing notification to kafka for user = {}", partitionKey,  e);
        }
    }
}

如何在此启用 onFailure 回调，以便我可以通过日志知道是否有故障？

【问题讨论】：

标签： apache-kafka callback spring-kafka kafka-producer-api

【解决方案1】：

替代解决方案，假设您不需要将参数记录到推送方法中。您应该从 RecordMetadata 中获得所需的所有信息。

producer.send(producerRecord, this);

public class KafkaPublisher implements Callback {

   ... 

   @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        // ... 
   } 
}

【讨论】：

【解决方案2】：

这就是你的做法。

            producer.send(producerRecord, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e!=null)
                    {
                        log.error("Callback : Failed to push event to kafka for partition key, notification {] {}", partitionKey, serializedKafkaNotification, e);
                    }
                }
            });

【讨论】：