代理关闭时，Kafka producer正在丢失消息

Question

鉴于以下情况：

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

然后，我运行一个简单的java程序，该程序每秒向“测试”主题生成一条消息。一段时间后，我关闭了本地的kafka代理，看到制作人继续生成消息，它没有抛出任何异常。最后，我再次启动kafka broker，producer可以重新连接到broker并继续生成消息，但是，在kafka broker宕机期间生成的所有消息都丢失了。当检测到健康的Kafka经纪人时，制作人不会重播它们。

我怎样才能防止这种情况？我希望kafka producer在检测到kafka broker重新联机时重播这些消息。这是我的制作人配置：

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

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

然而，奇怪的是，当你让你的经纪人下台时，你没有得到任何异常/警告/错误。

我预计会出现“更新元数据失败”或“消息过期”错误，因为当生产者将消息发送到引导过程中提到的代理时。servers属性，它首先与zookeeper检查活动控制器（或引线）和分区。因此，在您的情况下，因为您在独立模式下运行Kafka，并且当代理关闭时，制作人不应该收到领导信息并出错。

请检查以下属性设置为什么：

request.timeout.ms
max.block.ms

并利用这些价值观（减少，可能是）？然后检查结果？

您可能想要尝试的另一个选项是以同步方式向Kafka发送消息（在消息被接收之前阻止发送（）方法），这里有一个可能有帮助的代码片段（取自此留档参考）：

如果你想模拟一个简单的阻塞调用，你可以立即调用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应引发异常。

我希望这有帮助。

Answer 2

Kafka Producer库内置了重试机制，但默认情况下已关闭。将retriesProducer-config更改为大于0（默认值）的值以启用它。您还应该尝试重试。退避。ms和请求。timetout。ms以自定义生产者重试。

示例Kafka生产者配置与启用重试：

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

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