当Kafka服务器关闭几分钟时，Kafka制作人正在丢失消息

Question

我已经编写了一个Java程序，它利用了Kafka库，我听说Kafka制作人有内部缓冲区来保存消息，以便稍后重试。所以我创建了具有重试属性的幂等Kafka Producer。

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv(KafkaConstants.KAFKA_URL));
    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");
    props.put("linger.ms", 1000);
    props.put("acks", "all");
    props.put("request.timeout.ms",60000);
    props.put("retries",3);
    props.put("retry.backoff.ms",1000);
    props.put("max.in.flight.requests.per.connection",1);
    props.put("enable.idempotence",true);

在运行程序之前，我会关闭Kafka服务器（只有一个代理）。当我运行程序时，我遇到了一个异常“60000毫秒后更新元数据失败”。但是，当我重新启动Kafka服务器时，它应该将数据推送到Kafka主题，因为我已经给出了重试属性。

请在这方面提供帮助。

谢谢Priyam Saluja

Answer 1

我发现了问题所在，每次Kafka Producer尝试生成消息时，它都会首先更新元数据（以检查Kafka集群中的前导和分区）。如果无法获取信息，则会抛出错误，说明“60000毫秒后更新元数据失败”。

第二部分是重试，如果消息因暂时性错误而失败，Kafka制作人将重试。

Answer 2

Kafka客户端发送的第一个请求之一是获取元数据。请记住，客户端试图连接到引导服务器列表中的代理，但它可能想要发送的主题可能不是其中之一。例如，考虑有3个代理B01、B02、B03和引导服务器只是B01，但生产者想要将消息发送到以B02为主题的主题分区：生产者需要第一元数据请求来获取该信息，然后打开与B02的连接来发送消息。我猜重试机制在这一步之后开始发挥作用，因为生产者内部的批处理利用了已知分区及其所在位置。您应该检查重试工作是否在获取元数据步骤正确完成后关闭服务器，并且生产者知道谁是分区负责人。