如何檢測Kafka中的連接錯誤

Question

我正在使用KafkaProducer和java。這是一個代碼示例：

package com.mypackage.kafka.producer; 

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerRecord; 

import java.util.Properties; 

public class ProducerToTest 
{ 
    public static void main(String[] args) 
    { 
     Properties props = new Properties(); 
     props.put("bootstrap.servers", "localhost:9092"); 
     props.put("retries", 0); 
     props.put("batch.size", 16384); 
     props.put("linger.ms", 1); 
     props.put("buffer.memory", 33554432); 
     props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 
     props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 

     KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); 
     for (int i = 0; i < 100; i++) 
     { 
      System.out.println(i); 
      producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i))); 
     } 

     producer.close(); 
    } 
} 

我遇到了問題。如果我的Kafka關閉了，並且我運行了我的代碼，它會在第一次發送時停止並等待Kafka重新啓動。

如何檢測連接錯誤或使用以不停止所有執行的方式發送？

Answer 1

製作人將嘗試重新連接到request.timeout.ms。它不會無限期地等待經紀人。請參閱request.timeout.ms的詳細信息。

該配置控制客戶端等待請求響應的最長時間。如果在超時過去之前未收到響應，客戶端將在必要時重新發送請求，或者如果重試耗盡，請求失敗。