卡夫卡生产者超时异常答案

【问题标题】：Kafka Producer TimeOutException卡夫卡生产者超时异常
【发布时间】：2019-04-12 20:52:46
【问题描述】：

我正在运行将数据写入 Kafka 主题的 Samza 流作业。 Kafka 正在运行一个 3 节点集群。 Samza 作业部署在纱线上。我们在容器日志中看到了很多这样的异常：

 INFO [2018-10-16 11:14:19,410] [U:2,151,F:455,T:2,606,M:2,658] samza.container.ContainerHeartbeatMonitor:[ContainerHeartbeatMonitor:stop:61] - [main] - Stopping ContainerHeartbeatMonitor
ERROR [2018-10-16 11:14:19,410] [U:2,151,F:455,T:2,606,M:2,658] samza.runtime.LocalContainerRunner:[LocalContainerRunner:run:107] - [main] - Container stopped with Exception. Exiting process now.
org.apache.samza.SamzaException: org.apache.samza.SamzaException: Unable to send message from TaskName-Partition 15 to system kafka.
        at org.apache.samza.task.AsyncRunLoop.run(AsyncRunLoop.java:147)
        at org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:694)
        at org.apache.samza.runtime.LocalContainerRunner.run(LocalContainerRunner.java:104)
        at org.apache.samza.runtime.LocalContainerRunner.main(LocalContainerRunner.java:149)
Caused by: org.apache.samza.SamzaException: Unable to send message from TaskName-Partition 15 to system kafka.
        at org.apache.samza.system.kafka.KafkaSystemProducer$$anon$1.onCompletion(KafkaSystemProducer.scala:181)
        at org.apache.kafka.clients.producer.internals.RecordBatch.done(RecordBatch.java:109)
        at org.apache.kafka.clients.producer.internals.RecordBatch.maybeExpire(RecordBatch.java:160)
        at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortExpiredBatches(RecordAccumulator.java:245)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:212)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:135)
        at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.common.errors.TimeoutException: Expiring 5 record(s) for Topic3-16 due to 30332 ms has passed since last attempt plus backoff time

这 3 种类型的异常来了很多。

59088 org.apache.kafka.common.errors.TimeoutException: Expiring 115 record(s) for Topic3-1 due to 30028 ms has passed since last attempt plus backoff time

61015 org.apache.kafka.common.errors.TimeoutException: Expiring 60 record(s) for Topic3-1 due to 74949 ms has passed since batch creation plus linger time

62275 org.apache.kafka.common.errors.TimeoutException: Expiring 176 record(s) for Topic3-4 due to 74917 ms has passed since last append

请帮助我了解这里的问题。每当它发生时，Samza 容器就会重新启动。

【问题讨论】：

标签： apache-kafka kafka-producer-api apache-samza

【解决方案1】：

错误表明某些记录被放入队列的速度比它们从客户端发送的速度要快。

当您的生产者发送消息时，它们被存储在缓冲区中（在发送到目标代理之前），并且记录被分组到一起以增加吞吐量。将新记录添加到批次时，必须在由request.timeout.ms 控制的可配置时间窗口内发送（默认设置为 30 秒）。如果批次在队列中的时间较长，则会抛出TimeoutException，然后批次记录将从队列中删除，并且不会传递给代理。

增加request.timeout.ms 的值应该可以解决问题。

如果这不起作用，您还可以尝试减少 batch.size 以便更频繁地发送批次（但这次将包含更少的消息）并确保将 linger.ms 设置为 0（这是默认值）价值）。

请注意，您需要在更改任何配置参数后重新启动您的 kafka 代理。

如果您仍然收到错误消息，我认为您的网络出现问题。您启用 SSL 了吗？

【讨论】：

谢谢，Giorgos 我会试试这个属性。我也遇到了很多错误，比如org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. 这也是由于request.timeout.ms
@Anujjain 首先，您能否确认您的所有 Kafka 代理都已启动并运行？其次，你设置了多少次重试？
是的，所有 Kafka 代理都在运行。下面是我们的配置。 batch.size = 16384 buffer.memory = 33554432 linger.ms = 0 max.block.ms = 60000 max.in.flight.requests.per.connection = 1 max.request.size = 14000000 metadata.fetch.timeout.ms = 60000 metadata.max.age.ms = 300000 metrics.sample.window.ms = 30000 receive.buffer.bytes = 32768 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 20 retry.backoff.ms = 1000
@Anujjain 你试过增加request.timeout.ms吗？
由于此问题已得到解答，并且您目前面临另一个错误，我建议您提出一个新问题并在您的问题中包含新的错误和配置。