当 kafka 服务器关闭时，Kafka 生产者无限期发送块

Question

我正在使用 Kafka 0.11.0.0。我有一个发布到 Kafka 主题的测试程序；如果 zookeeper 和 Kafka 服务器宕机（这在我的开发环境中是正常的；我会根据需要启动它们）然后对 KafkaProducer.send() 的调用将无限期挂起。

我要么需要 send() 返回，最好是指示错误；或者我需要一种方法来检查服务器是启动还是关闭。基本上，我希望我的测试工具能够告诉我，“嘿，笨蛋，启动 Kafka！”而不是挂起。

我的生产者任务有没有办法确定服务器是启动还是关闭？

我这样调用 send()：

kafkaProducer.send(new ProducerRecord<>(KAFKA_TOPIC, KAFKA_KEY,
    message), (rm, ex) -> {
        System.out.println("**** " + rm + "\n**** " +ex);
});

我有 linger.ms = 1;我尝试了 retries=0、1 和 2，但 send() 仍然阻塞。我从未见过回调被调用。

较早的消息建议将 metadata.fetch.timeout.ms 设置为一个较小的值，但在 0.11 中已取消。其他人建议调用命令行实用程序以查看服务器是否正常......但引用的实用程序似乎也不见了。

完成这项工作的优雅方式是什么？

Answer 1

这很奇怪。它应该返回一个错误，说明“更新元数据失败”或“正在过期 x 条记录”。

检查您的生产者的request.timeout.ms 和max.block.ms 设置。默认情况下request.timeout.ms 为 60 秒

Answer 2

我们可以通过三种方式向代理发送消息：

即发即弃： 我们向服务器发送一条消息，并不关心它是否成功到达。大多数情况下，它会成功到达，因为 Kafka 是高可用的，生产者会自动重试发送消息。但是，使用此方法会丢失一些消息。

异步发送 我们使用回调函数调用 send() 方法，当它收到来自 Kafka 代理的响应时会触发该方法。

同步发送 我们发送一条消息，send() 方法返回一个 Future 对象，我们使用 get() 等待未来，看看 send() 是否成功。

同步发送消息最简单的方法如下：

ProducerRecord<String, String> record =
            new ProducerRecord<>(KAFKA_TOPIC, KEY, message);
    try {
            producer.send(record).get();
    } catch (Exception e) {
             e.printStackTrace();
    }

在这里，我们使用 Future.get() 来等待来自 Kafka 的回复。如果记录没有成功发送到 Kafka，该方法会抛出异常。如果没有错误，我们将获得一个 RecordMetadata 对象，我们可以使用它来检索消息写入的偏移量。

希望这会有所帮助。