Kafka Producer:在带回调的异步发送中处理异常

Question

我需要捕获异步发送到Kafka时的异常。Kafka producer Api附带一个函数send（ProducerRecord记录、回调）。但当我针对以下两种情况进行测试时：

• Kafka经纪人倒下
• 主题没有预创建回调没有被调用。相反，我在代码中收到发送不成功的警告（如下所示）。

问题：

>

那么回调是否只针对特定的异常调用？

Kafka客户端何时尝试在异步发送时连接到Kafka代理：每次批处理发送还是定期发送？

Kafka警告图像

注意：我也使用linger.ms设置25秒批量发送我的记录。

public class ProducerDemo {

    static KafkaProducer<String, String> producer;

    public static void main(String[] args) throws IOException {

         final Logger logger = LoggerFactory.getLogger(ProducerDemo.class);
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "1");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "30000");

        producer = new KafkaProducer<String, String>(properties);
        String topic = "first_topic";

        for (int i = 0; i < 5; i++) {
            String value = "hello world " + Integer.toString(i);
            String key = "id_" + Integer.toString(i);

            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, key, value);

              producer.send(record, new Callback() {
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        //execute everytime a record is successfully sent or exception is thrown
                        if(e == null){
                           // No Exception
                        }else{
                            //Exception Handling
                        }
                    }
                });
        }
        producer.close();
    }

Answer 1

总的来说，我注意到很多与KafkaProducer交付语义学和保证相关的问题。它肯定可以更好地记录下来。

还有一件事，自从你提到玲儿。ms:

请注意，及时到达的记录通常会批处理在一起，即使存在延迟。ms=0，因此在重负载下，无论玲珑配置如何，都会进行批处理

Answer 2

那么回调是否只针对特定的异常调用？

是的，这就是它的工作原理。从留档（2.5.0）：

     * Fully non-blocking usage can make use of the {@link Callback} parameter to provide a callback that
     * will be invoked when the request is complete.

请注意重要的部分：当请求完成时，什么意味着生产者必须接受记录并将产品请求发送给Kafka Broker。无需深入挖掘内部，这意味着代理元数据必须存在，分区必须存在。

谈到正式规范，您需要仔细了解send（）的Javadoc，可能还需要了解KafkaProducer对doSend方法的实现。在那里，您将看到在提交调用时可以抛出多个异常（而不是返回未来并调用回调），例如：

• 如果代理元数据在超时时不可用，
• 如果数据不能序列化，
• 如果序列化的表单太大等

Answer 3

对于第一个问题，这里是答案。根据apache kafka留档，当您实现回调接口时，您可以使用onComplments方法捕获以下异常

https://kafka.apache.org/25/javadoc/org/apache/kafka/clients/producer/Callback.html

对于第二个问题，以下属性的组合控制何时发送记录，据我所知，同步或异步调用也是如此。

徘徊马克斯·布洛克女士。太太

https://kafka.apache.org/documentation/#linger.ms