在 quarkus 中处理 kafka 消息时出现 TooManyMessagesWithoutAckException答案

【问题标题】：TooManyMessagesWithoutAckException while processing kafka message in quarkus在 quarkus 中处理 kafka 消息时出现 TooManyMessagesWithoutAckException
【发布时间】：2023-02-06 17:23:54
【问题描述】：

在 quarkus 进程中，一旦从 kafka 轮询消息，我们将执行以下步骤

Thread.sleep(30000) - 由于业务逻辑
调用第三方 API
调用另一个第 3 方 api

在数据库中插入数据

几乎每天都有一次该进程在抛出 TooManyMessagesWithoutAckException 后挂起。

2022-12-02 20:02:50 INFO  [2bdf7fc8-e0ad-4bcb-87b8-c577eb506b38,     ] : Going to sleep for 30 sec.....
2022-12-02 20:03:20 WARN  [                    kafka] : SRMSG18231: The record 17632 from topic-partition '<partition>' has waited for 60 seconds to be acknowledged. This waiting time is greater than the configured threshold (60000 ms). At the moment 2 messages from this partition are awaiting acknowledgement. The last committed offset for this partition was 17631. This error is due to a potential issue in the application which does not acknowledged the records in a timely fashion. The connector cannot commit as a record processing has not completed.
2022-12-02 20:03:20 WARN  [                     kafka] : SRMSG18228: A failure has been reported for Kafka topics '[<topic name>]': io.smallrye.reactive.messaging.kafka.commit.KafkaThrottledLatestProcessedCommit$TooManyMessagesWithoutAckException: The record 17632 from topic/partition '<partition>' has waited for 60 seconds to be acknowledged. At the moment 2 messages from this partition are awaiting acknowledgement. The last committed offset for this partition was 17631.
2022-12-02 20:03:20 INFO  [2bdf7fc8-e0ad-4bcb-87b8-c577eb506b38,     ] : Sleep over!

以下是我们如何使用消息的示例

@Incoming("my-channel")
@Blocking
CompletionStage<Void> consume(Message<Person> person) {
     String msgKey = (String) person
        .getMetadata(IncomingKafkaRecordMetadata.class).get()
        .getKey();
        // ...
      return person.ack();
}

根据日志，自事件被轮询以来仅过去了 30 秒，但抛出了 60 秒内未发送 kafka 确认的异常。当错误被抛出时，我检查了一整天的日志，看看 REST api 调用是否花费超过 30 秒来获取数据，但我找不到任何东西。

除了主题名称、通道名称、序列化程序、反序列化程序、组 ID 和托管的 kafka 连接详细信息之外，我们没有进行任何特定的 kafka 配置。

此主题中有 4 个分区，复制因子为 3。此进程运行 3 个 pod。我们无法在 Dev 和 UAT 环境中重现此问题。

我检查了配置选项，但找不到任何可能有帮助的配置：Quarkus Kafka Reference

mp:
  messaging:
    incoming:
      my-channel:
        topic: <topic>
        group:
          id: <group id>
        connector: smallrye-kafka
        value:
          serializer: org.apache.kafka.common.serialization.StringSerializer
          deserializer: org.apache.kafka.common.serialization.StringDeserializer

quarkus 是否有可能批量确认消息，到那时等待时间已经达到阈值？如果此问题还有其他可能性，请发表评论。

【问题讨论】：

Kafka 要求你在一定时间内定期轮询。你不应该休眠那个线程。如果你需要做很多工作，而是 pause() 消费者......这不是 Quarkus 独有的问题。否则，生成主题以调用 API 并使用一系列主题中的响应，而不是尝试调用多个 API 并在一个操作中全部写入数据库（和/或使用 Kafka Connect 实际执行数据库工作）

标签： java apache-kafka quarkus smallrye-reactive-messaging

【解决方案1】：

我在使用简单的 3-Node-Kafka-Cluster 运行不同 quarkus 服务的生产环境中遇到了类似的问题，我对这个问题进行了很多研究 - 没有明确的答案。目前，我有两种方法来解决这个问题：

请确保您确实在代码中确认或拒绝了 kafka 消息。真的每个异常都被捕获并用“person.nack(exception);”回答吗？（或“person.ack(()”——取决于你的失败策略）？确保它是。如果没有执行 ack() 或 nack()，则会抛出错误 Throttled-Exception。问题主要发生在，如果什么都没有发生。
当这没有帮助时，我将提交策略切换为“最新”： mp.messaging.incoming.my-channel.commit-strategy=latest 这有点慢，因为批处理提交被禁用，但在我的情况下运行稳定。如果你不知道提交策略和默认，赶上好article by Escoffier：
我知道，这并不能解决根本原因，但在绝望的时候有所帮助。问题一定是，一条或多条排队的消息没有及时得到确认，但我不能告诉你原因。也许应用程序逻辑太慢了，但我和你一样很难在本地重现它。您也可以尝试使用 throttled.unprocessed-record-max-age.ms 增加 60 秒的阈值，并亲自看看是否有帮助。就我而言，它没有。也许其他人可以分享他对这个问题的见解，并可以为您提供真正的解决方案。

【讨论】：