Middleware/mq-starter
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mq-starter-rocketmq): 消费捕获 MqMessageInProgressException 返回 RECONSUME_LATER，杜绝处理中消息被耗尽落库+ACK 静默丢弃

Browse Source 
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
hongawen
2026-07-10 02:54:09 +01:00
parent ae0caafa63
commit f4793b10c0
2 changed files with 42 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java
View File

@@ -3,6 +3,7 @@ package com.njcn.mq.driver.rocketmq;
import com.alibaba.fastjson.JSON;
import com.njcn.mq.core.MqMessage;
import com.njcn.mq.spi.MqDriver;
import com.njcn.mq.spi.MqMessageInProgressException;
import com.njcn.mq.spi.MqSubscription;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -63,6 +64,11 @@ public class RocketMqDriver implements MqDriver {
String json = new String(ext.getBody(), StandardCharsets.UTF_8);
Object payload = JSON.parseObject(json, sub.getPayloadType());
sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
} catch (MqMessageInProgressException inProgress) {
// 去重"处理中"信号:不 ACK,交 broker 重投;绝不进耗尽落库分支(否则并发下静默丢消息)。
// 语义对齐 RedisStreamMqDriver#dispatch 对该异常的 return false(留 PEL)。
log.debug("[mq-rocket] in-progress, reconsume later key={}", ext.getKeys());
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
} catch (Exception e) {
if (ext.getReconsumeTimes() >= maxRetry) {
// 重试耗尽:回调 retryExhaustedHandler 落库并 ACK 终止,不再无限重投

36
mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java
View File

@@ -1,6 +1,7 @@
package com.njcn.mq.driver.rocketmq;
import com.njcn.mq.core.MqMessage;
import com.njcn.mq.spi.MqMessageInProgressException;
import com.njcn.mq.spi.MqSubscription;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
@@ -84,4 +85,39 @@ class RocketMqDriverConsumeTest {
assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status);
assertEquals(Collections.singletonList("hi"), received, "成功路径应把 payload 投给业务");
}
@Test
void consume_inProgress_reconsumesLaterWithoutLandingEvenIfExhausted() {
List<MqMessage> landed = new ArrayList<>();
MqSubscription sub = MqSubscription.builder()
.topic("t").group("g").payloadType(String.class).maxRetry(3)
.listener(m -> { throw new MqMessageInProgressException("k1"); })
.retryExhaustedHandler(landed::add)
.build();
// reconsumeTimes=3 >= maxRetry=3:普通异常会走"耗尽落库+ACK",而 InProgress 是控制流信号,
// 必须仍 RECONSUME_LATER 且不落库——否则并发中的消息被当耗尽 ACK,静默丢消息。
ConsumeConcurrentlyStatus status =
newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3);
assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status,
"处理中信号应保留消息交 broker 重投,绝不 ACK");
assertEquals(0, landed.size(), "处理中信号不是重试耗尽,不得回调 retryExhaustedHandler 落库");
}
@Test
void consume_inProgress_notExhausted_alsoReconsumesLater() {
List<MqMessage> landed = new ArrayList<>();
MqSubscription sub = MqSubscription.builder()
.topic("t").group("g").payloadType(String.class).maxRetry(3)
.listener(m -> { throw new MqMessageInProgressException("k1"); })
.retryExhaustedHandler(landed::add)
.build();
ConsumeConcurrentlyStatus status =
newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 0)), 3);
assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status);
assertEquals(0, landed.size());
}
}