向 kafka 主题发送消息时出现 TimeoutException

Question

我遇到了向 Kafka 主题生成消息的问题。

当向 Kafka 主题生成一些消息时，它会间歇性地失败并出现异常：

org.apache.kafka.common.errors.TimeoutException：some-topic-1 的 6 条记录已过期：自批处理创建以来已过去 30056 毫秒加上逗留时间

这个问题似乎与我在这里找到的其他问题不同。它发生在负载非常低的情况下。我很确定我没有超过缓冲区大小（默认值）。 linger.ms 设置为 0（默认）。 request.timeout.ms 也设置为默认值 30000。

实际上我的问题是：在调用 send() 方法和实际发送它之间，有什么可以让发送 kafka 消息的过程持续 30 秒吗？我正在寻找任何能让ProducerBatch 在极低负载（或无任何负载）下的生存时间超过 30 秒的东西。

我使用来自外部提供商的 Kafka 托管服务，所以我向他询问了代理状态，他说一切正常。顺便说一句，它发生在三个不同的 Kafka 实例上。 Kafka 客户端版本也无关紧要 - 它适用于 0.11.0.0 和 2.0.1。

Answer 1

我刚刚查看了Kafka documentation for version 2.0.X，我想说您正面临此错误，因为默认设置和低消息摄取负载。

让我们试着剖析你的错误信息：

org.apache.kafka.common.errors.TimeoutException: 6 条记录过期 对于 some-topic-1：自批量创建加上 linger 以来已经过去了 30056 毫秒 时间

1. 错误中的第一条线索是 30056 毫秒已过去 - 配置 request.timeout.ms 正在这里播放。此配置的默认值为 30000，即 30 秒。这 30 秒的等待是为了填充 Producer 端的缓冲区 - 我已经在下一点解释了这一点。
2. 错误消息中的第二条线索是自从批量创建 - 当您调用 send() 方法时 - 消息在 Producer 端被缓冲，它将等待 30秒（按照第一个配置属性的指示）填充批次。现在，batch.size 的默认值为 16384 字节，即 16 KB，因此如果消息摄取负载较低（如您的情况）并且缓冲区没有填满更多大小为 16 KB 的消息30 秒 - 您会看到此错误消息。因此，在您的情况下，Expiring 6 record(s) 似乎并不能真正构成填充批次所需的 16 KB。
3. 最后，错误还提到了 plus linger time - 这是linger.ms 配置，发现它在摄取负载的情况下使用很高，生产者希望限制对 Kafka 代理的 send() 调用。这是生产者在批处理准备好后将消息发送到代理之前等待的持续时间（即缓冲区已填满batch.size）。默认值为 0，因此在您的情况下，生产者将在批处理准备好后等待 0 毫秒。但是您的缓冲区实际上并没有在给定的时间内填满。

现在，回答您的问题：您可以增加linger.ms，以便在批处理准备好后能够更长时间地保留消息。如果您需要更多时间来填充批次，那么您需要增加request.timeout.ms。你不妨试试两者的结合。

在类似的行上，要修复此错误，您可以增加 request.timeout.ms 或减少 batch.size 或两者兼而有之。

希望这会有所帮助！