集群的一个Kafka代理关闭时发生ConnectException

Question

我有两个Kafka代理：server1:9092和server2:9092我正在使用Java客户端向这个集群发送消息，代码如下：

@Test
public void sendRecordToTopic() throws InterruptedException, ExecutionException {

    //See at http://kafka.apache.org/documentation.html#newproducerconfigs
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
            "server1:9092,server2:9092");
    props.put(ProducerConfig.ACKS_CONFIG, "1");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

    ProducerRecord<String, String> myRecord =
            new ProducerRecord<String, String>("my-replicated-topic", "test", "someValue");

    boolean syncSend = true;

    if (syncSend) {
        //Synchronously send
        producer.send(myRecord).get();
    } else {
        //Asynchronously send
        producer.send(myRecord);
    }
    producer.close();
}

当其中一个代理关闭时，Test在某些情况下会抛出此异常（在此异常示例中'server1'已关闭）：

2015-11-02 17:59:29138警告[org.apache.kafka.common.network.Selector]服务器1/40.35.250.227 java的I/O错误。网ConnectException:连接被拒绝：sun没有进一步的信息。尼奥。索克坦林普尔。sun的checkConnect（本机方法）。尼奥。索克坦林普尔。finishConnect（socketchannelmpl.java:717）位于org。阿帕奇。Kafka。常见的网络选择器。在org上投票（Selector.java:238）。阿帕奇。Kafka。客户。网络客户端。在org上投票（NetworkClient.java:192）。阿帕奇。Kafka。客户。制作人内部。发件人。在org上运行（Sender.java:191）。阿帕奇。Kafka。客户。制作人内部。发件人。在java上运行（Sender.java:122）。朗。丝线。运行（Thread.java:745）

Answer 1

这就是我解决问题的方法：

>

将此参数添加到ZooKeeper属性文件：

滴答时间=200

使用这些其他参数时需要此参数：

initLimit=5
syncLimit=2

>

道具。setProperty（ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG，“10000”）；

使用"RECONNECT_BACKOFF_MS_CONFIG"属性，WARN只被抛出一次（不是无限循环），然后消息被发送

Answer 2

我遇到了这个问题，结果发现原因是对一个新配置属性的误解。

在从以前的ProducerAPI迁移时，我寻找了一个与“topic.metadata.refresh.interval.ms”等效的API，并选择了ProducerConfig。元数据获取超时配置。然而，事实证明，这是在尝试访问元数据被视为失败之前的超时，由于我将其设置为几分钟，它阻止了故障转移的发生。

将其设置为较低的值（我选择500毫秒）似乎解决了我的问题。

我相信我最初寻找的价值是ProducerConfig。METADATA_MAX_AGE_CONFIG作为元数据刷新前的超时，无论是否发生故障