当代理关闭时，Kafka 生产者正在丢失消息

Question

鉴于以下情况：

我在本地启动 zookeeper 和单个 kafka 代理，并按照 kafka 快速入门中所述创建“测试”主题：https://kafka.apache.org/quickstart

然后，我运行一个简单的 java 程序，它每秒向“test”主题生成一条消息。一段时间后，我关闭了本地 kafka 代理并看到生产者继续生成消息，它没有抛出任何异常。最后，我再次启动了 kafka 代理，生产者能够重新连接到代理并继续生成消息，但是，在 kafka 代理停机期间生成的所有这些消息都丢失了。当检测到健康的 kafka 代理时，生产者不会重播它们。

如何防止这种情况发生？我希望 kafka 生产者在检测到 kafka 代理重新联机时重播这些消息。这是我的生产者配置：

props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("linger.ms", 0);
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", StringSerializer.class.getName());

Answer 1

Kafka Producer 库内置了重试机制，但默认情况下它是关闭的。将 retries Producer 配置更改为大于 0（默认值）的值以将其打开。您还应该尝试使用 retry.backoff.ms 和 request.timetout.ms 来自定义 Producer 重试。

启用重试的示例 Kafka Producer 配置：

retries=2147483647         //Integer.MAX_VALUE 
retry.backoff.ms=1000
request.timeout.ms=305000  //5 minutes
max.block.ms=2147483647    //Integer.MAX_VALUE 

您可以在Apache Kafka documentation 中找到有关这些属性的更多信息。

Answer 2

由于您只运行一个代理，恐怕当您的代理关闭时您将无法存储消息。

但是，奇怪的是，当您关闭代理时，您没有收到任何异常/警告/错误。

我预计会出现“更新元数据失败”或“消息过期”错误，因为当生产者向 bootstrap.servers 属性提到的代理发送消息时，它首先会与 Zookeeper 一起检查活动控制器（或领导者）和分区。因此，在您的情况下，由于您以独立模式运行 kafka，并且当代理关闭时，生产者不应收到领导者信息并出错。

能否请您检查以下属性设置为：

request.timeout.ms
max.block.ms

并玩弄（减少，可能）这些值？并检查结果？

您可能想尝试的另一个选项是以同步方式向 Kafka 发送消息（阻塞 send() 方法直到收到消息），这是一个可能有帮助的代码 sn-p（取自 this documentation reference ):

如果你想模拟一个简单的阻塞调用，你可以立即调用 get() 方法：

byte[] key = "key".getBytes();
byte[] value = "value".getBytes();
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("my-topic", key, value)
producer.send(record).get();

在这种情况下，如果消息由于某种原因没有成功发送，kafka 应该抛出异常。

我希望这会有所帮助。