Quarkus SmallRye Reactive Messaging：配置轮询频率规避SRMSG18231异常

问题背景 #

使用Quarkus 2.13.3.Final + quarkus-smallrye-reactive-messaging-kafka（SmallRye 3.21.0）在Kubernetes集群开发非阻塞消费流程：接收Kafka消息→调用外部API→处理逻辑→输出结果。单条消息处理耗时约8秒，但启动约1分钟后触发SRMSG18231（waitingForAckForTooLong）异常，导致Pod频繁重建进入崩溃循环。临时方案是通过mp.messaging.incoming.queue.throttled.unprocessed-record-max-age.ms=-1关闭超时检查，但需寻找合理的消息摄取速率限制方案。

核心原因 #

1. 消息预取过载：Kafka消费者一次性拉取大量消息，尽管单条处理时长低于默认的60秒超时阈值，但未启动处理的消息积压在队列中，KafkaThrottledLatestProcessedCommit的receivedOffsets持续增长。
2. 超时计时逻辑缺陷：超时检查以最早接收的未确认消息的时间为起点计算，而非单条消息的实际处理开始时间，导致积压消息触发超时。
3. 配置参数误用：之前尝试的max.poll.records未添加kafka.前缀，无法传递给底层Kafka消费者，因此未生效；max-inflight-messages等参数未匹配@Blocking(ordered=false)的多线程处理场景。

有效配置方案 #

1. 修正Kafka原生参数传递 #

将max.poll.records添加kafka.前缀，确保配置能作用于底层Kafka消费者，限制每次poll拉取的消息数量：

mp.messaging.incoming.queue.kafka.max.poll.records=20

2. 控制并发处理与消息 inflight 数量 #

结合@Blocking(ordered=false)的多线程处理特性，配置并发数和单订阅的最大 inflight 消息数，避免无限制积压：

# 设置并发处理器数量（根据Pod资源调整，例如4）
mp.messaging.incoming.queue.concurrency=4
# 限制每个订阅的最大未确认消息数
mp.messaging.incoming.queue.max-inflight-messages-per-subscription=40

3. 改用有界队列控制过载 #

将@OnOverflow的无界缓冲区改为有界队列，当队列满时暂停Kafka消息拉取，从根源控制消息摄取速率：

@Blocking(ordered = false)
@OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 40) // 替换为有界队列
@Acknowledgment(Acknowledgment.Strategy.MANUAL)
@Incoming("queue")
public void processMessage(Message<YourPayload> msg) {
    // 处理逻辑...
    msg.ack(); // 确保处理完成后手动确认
}

4. 调整超时检查窗口（可选） #

如果并发数增加后，消息积压的最大时长超过60秒，可适当调大超时阈值，保留超时检查的防护机制：

mp.messaging.incoming.queue.throttled.unprocessed-record-max-age.ms=120000 # 2分钟

额外优化建议 #

• 规范手动确认时机：必须在消息处理全流程（包括外部API调用、结果发送）完成后调用msg.ack()，确保receivedOffsets及时更新，避免无效的超时触发。
• 监控调优：在Kubernetes中监控Pod的CPU/内存、Kafka消费者的consumer_lag（消费延迟）、records_consumed_total（消费总量）等指标，根据实际负载动态调整并发数、poll数量和队列大小。
• 版本升级（可选）：Quarkus 2.13.x属于旧LTS版本，后续版本（如2.16.x或3.x）的SmallRye Reactive Messaging可能修复了超时计时逻辑的问题，可考虑升级并验证兼容性。

内容的提问来源于stack exchange，提问作者Andrew Cattermole