“未能分配内存”和“过期记录”错误后，卡夫卡代理为5+ mins，然后重新加入集群

Question

我们正在对AWS EC2中的Kafka1.1代理进行一些测试。具体来说，我们正在干净地关闭代理5分钟或更长时间，然后重新启动它们，同时一直从集群中生产和使用它们。理论上，这对生产者和消费者来说都应该是相对无缝的。

但是，当测试代理停机超过5分钟时，我们在生产者应用程序上得到了这些错误：

 ERROR - 2019-02-27 04:34:24.946 - message size [2494] 
       -Expiring 7 record(s) for topic2-0: 30033 ms has passed since last append

然后是许多类似的错误中的第一个

 ERROR - 2019-02-27 04:35:13.098; Topic [topic2], message size [2494] 
       -Failed to allocate memory within the configured max blocking time 60000 ms. 

此时，长期运行的生产者变得没有响应能力，每个生产者的请求都会在相同的Failed to allocate memory中失败。我试图在网上搜索类似的问题，但我所能找到的只是0.10.1.1中解决的一张旧Kafka JIRA票证，所以它不应该申请我们正在使用的新的1.1版本。

https://issues.apache.org/jira/browse/KAFKA-3651

我们尝试了许多不同的场景，包括将生产者配置更改为Kafka默认值，以查看是否有任何帮助，但是每当代理在运行5分钟或更长时间后回到集群时，我们总是会遇到这个问题。

Answer 1

终于想出了解决办法。把它贴在这里，以防对别人有用。

我们的producer.send调用定义了一个回调，在发生错误时，将尝试使用相同的生产者将消息重新发送到另一个主题。这个机制已经建立，这样，如果发送调用导致错误，我们至少会将消息存储在错误主题中，在那里我们可以检查数据并找出正在发生的事情。

这种机制通常运行良好，但是每当我们暂时失去一个代理时(为了维护/升级/其他)，我们所有的应用程序都会被ELB卡住，并最终被ELB杀死。

从KafkaProducer.send()的官方Javadoc中，我发现了以下内容：

请注意，回调通常将在生产者的I/O线程中执行，因此应该相当快，否则将延迟从其他线程发送消息。如果您想执行阻塞或计算昂贵的回调，建议在回调体中使用您自己的执行器来并行处理。

https://kafka.apache.org/11/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

回调内部的producer.send可以说是一个昂贵的调用，IMHO。在进行了进一步的调试和分析之后，我想，当回调producer.send()调用等待陈旧的元数据时，所有的XNIO线程都被卡住了，这是从代理重新加入集群时开始的。这最终导致应用程序停止响应。

在我们的示例中，修复方法是简单地将错误消息排在ConcurrentLinkedDeque中，并让单独的线程处理这些消息，该线程的唯一任务是从deque发送take消息并尝试重新发送它们。

在将此更改部署到所有环境后，应用程序在多个代理重新启动和中断过程中表现良好。