卡夫卡生产者TimeOutException [英] Kafka Producer TimeOutException

问题描述

我正在运行一个Samza流作业，该作业正在将数据写入Kafka主题. Kafka正在运行一个3节点群集. Samza作业部署在纱线上.我们在容器日志中看到了许多此类异常:

I am running a Samza stream job that is writing data to Kafka topic. Kafka is running a 3 node cluster. Samza job is deployed on yarn. We are seeing lot of these exceptions in container logs :

 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种异常类型来的很多.

These 3 types of exceptions are coming a lot.

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容器都会重新启动.

Please help me understand what is the issue here. Whenever its happened Samza container is getting restarted.

推荐答案

错误表明某些记录以比从客户端发送记录更快的速度放入队列.

The error indicates that some records are put into the queue at a faster rate than they can be sent from the client.

当生产者发送消息时，它们存储在缓冲区中(在将消息发送到目标代理之前)，并且记录被分为几批以提高吞吐量.将新记录添加到批处理时，必须在由request.timeout.ms控制的-configurable-time窗口中发送该记录(默认设置为30秒).如果批次在队列中放置的时间较长，则会抛出TimeoutException，并且批次记录将从队列中删除，并且不会传递给代理.

When your Producer sends messages, they are stored in buffer (before sending the to the target broker) and the records are grouped together into batches in order to increase throughput. When a new record is added to the batch, it must be sent within a -configurable- time window which is controlled by request.timeout.ms (the default is set to 30 seconds). If the batch is in the queue for longer time, a TimeoutException is thrown and the batch records will then be removed from the queue and won't be delivered to the broker.

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

Increasing the value of request.timeout.ms should do the trick for you.

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

In case this does not work, you can also try decreasing batch.size so that batches are sent more often (but this time will include fewer messages) and make sure that linger.ms is set to 0 (which is the default value).

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

Note that you need to restart your kafka brokers after changing any configuration parameter.

如果仍然出现错误，我认为您的网络出现了问题.您启用了SSL吗?

If you still get the error I assume that something wrong is going on with your network. Have you enabled SSL?

这篇关于卡夫卡生产者TimeOutException的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！