当 Kafka 服务器关闭几分钟时，Kafka Producer 正在丢失消息答案

【问题标题】：Kafka Producer is losing the message when Kafka Server is down for few minutes当 Kafka 服务器关闭几分钟时，Kafka Producer 正在丢失消息
【发布时间】：2018-01-25 14:05:36
【问题描述】：

我编写了使用 Kafka 库的 Java 程序，我听说 Kafka Producer 有内部缓冲区来保存消息，以便稍后重试。所以我用重试属性创建了 Idempotent 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 主题，因为我已经给出了重试属性。

请在这方面提供帮助。

谢谢，普里亚姆萨卢加

【问题讨论】：

标签： apache-kafka kafka-producer-api

【解决方案1】：

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

【讨论】：

我在获取元数据的必要信息后尝试关闭服务器，然后我观察到它正在等待future.get方法60000毫秒并给出异常并使记录过期。在我重新启动服务器后，它没有将消息推送到 Kafka 主题。但是，如果我在等待 get 方法时重新启动服务器，那么它会将消息推送到 Kafka 主题。
这是因为我在 request.timeout 属性中给出了 60000 毫秒。它说客户端将最长等待 60000 毫秒才能获得响应，之后它将使请求失败。所以在这里我需要了解生产者何时会尝试重新发送消息，因为一旦 request.timeout 过去，它就不会发送消息。

【解决方案2】：

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

第二部分是重试，如果消息由于瞬时错误而失败，Kafka Producer 会尝试。

【讨论】：