feat(mq-starter-rocketmq): 重试耗尽回调 retryExhaustedHandler 落库并 ACK，maxRetry 对齐 @MqListener

对齐 redis-stream 的 reclaim 耗尽语义，兑现"换 MQ 业务不动"契约：此前 rocketmq
只有 broker 默认 16 次重投、无 RETRY_EXHAUSTED 落库。

- setMaxReconsumeTimes 对齐 @MqListener(maxRetry)（缺省 3），不再用 broker 默认 16
- 抽出 consumeBatch：reconsumeTimes 达 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止；未达则 RECONSUME_LATER
- onRetryExhausted 落库 best-effort：毒消息 payload 反序列化失败时留 null、至少保留 key/tag，并隔离 handler 自身异常
- 补消费逻辑单测（耗尽落库 / 未耗尽重投 / 成功投递）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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

@@ -16,6 +16,7 @@ import org.springframework.messaging.support.MessageBuilder;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 
 @Slf4j
 public class RocketMqDriver implements MqDriver {
@@ -41,27 +42,56 @@ public class RocketMqDriver implements MqDriver {
     @Override
     public void subscribe(MqSubscription sub) {
         try {
+            int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : 3;
             DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(sub.getGroup());
             consumer.setNamesrvAddr(namesrvAddr);
+            consumer.setMaxReconsumeTimes(maxRetry); // 对齐 @MqListener(maxRetry),而非用 broker 默认 16 次
             String tag = sub.getTag();
             consumer.subscribe(sub.getTopic(), (tag == null || tag.isEmpty()) ? "*" : tag);
-            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) ->
+                    consumeBatch(sub, msgs, maxRetry));
+            consumer.start();
+            consumers.add(consumer);
+        } catch (Exception e) {
+            throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e);
+        }
+    }
+
+    ConsumeConcurrentlyStatus consumeBatch(MqSubscription sub, List<MessageExt> msgs, int maxRetry) {
         for (MessageExt ext : msgs) {
             try {
                 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 (Exception e) {
-                        log.warn("[mq-rocket] handle failed msgId={}", ext.getMsgId(), e);
+                if (ext.getReconsumeTimes() >= maxRetry) {
+                    // 重试耗尽:回调 retryExhaustedHandler 落库并 ACK 终止,不再无限重投
+                    //（对齐 redis-stream reclaim 耗尽语义,兑现"换 MQ 业务不动"契约)。
+                    onRetryExhausted(sub, ext, e);
+                } else {
+                    log.warn("[mq-rocket] handle failed msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e);
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                 }
             }
+        }
         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-            });
+    }
-            consumer.start();
+
-            consumers.add(consumer);
+    /** 重试耗尽落库(best-effort:毒消息时 payload 为 null，至少保留 key/tag),并隔离 handler 自身异常。 */
-        } catch (Exception e) {
+    private void onRetryExhausted(MqSubscription sub, MessageExt ext, Exception cause) {
-            throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e);
+        log.warn("[mq-rocket] retry exhausted, acked msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause);
+        Consumer<MqMessage> handler = sub.getRetryExhaustedHandler();
+        if (handler == null) return;
+        Object payload = null;
+        try {
+            payload = JSON.parseObject(new String(ext.getBody(), StandardCharsets.UTF_8), sub.getPayloadType());
+        } catch (Exception ignore) {
+            // 毒消息:payload 留 null，至少保留 key/tag 供落库(对齐 redis buildMessage)
+        }
+        try {
+            handler.accept(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
+        } catch (Exception ex) {
+            log.error("[mq-rocket] retryExhausted handler error msgId={}（不影响主流程)", ext.getMsgId(), ex);
         }
     }
 

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

@@ -0,0 +1,87 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqSubscription;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+/**
+ * RocketMQ 消费处理逻辑单测(纯 POJO，不启动真实 consumer / 不连 namesrv）。
+ *
+ * <p>守护跨 driver 契约一致性:与 redis-stream 的 reclaim 耗尽语义对齐 ——
+ * 重试次数达到 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止;未达到则 RECONSUME_LATER 交 broker 重投。
+ */
+class RocketMqDriverConsumeTest {
+
+    private RocketMqDriver newDriver() {
+        return new RocketMqDriver(mock(RocketMQTemplate.class), "localhost:9876");
+    }
+
+    private static MessageExt msg(String key, String body, int reconsumeTimes) {
+        MessageExt ext = new MessageExt();
+        ext.setKeys(key);
+        ext.setBody(body.getBytes(StandardCharsets.UTF_8));
+        ext.setReconsumeTimes(reconsumeTimes);
+        return ext;
+    }
+
+    @Test
+    void consume_retryExhausted_callsHandlerAndAcks() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new RuntimeException("biz boom"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status,
+                "重试耗尽应 ACK 终止,不再无限重投");
+        assertEquals(1, landed.size(), "重试耗尽应回调 retryExhaustedHandler 落库(对齐 redis)");
+        assertEquals("k1", landed.get(0).getKey(), "落库消息应保留原 key");
+    }
+
+    @Test
+    void consume_notExhausted_reconsumesLaterWithoutLanding() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new RuntimeException("biz boom"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 1)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status,
+                "未达 maxRetry 应交 broker 重投");
+        assertEquals(0, landed.size(), "未耗尽不应落库");
+    }
+
+    @Test
+    void consume_success_deliversPayloadAndAcks() {
+        List<Object> received = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> received.add(m.getPayload()))
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hi\"", 0)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status);
+        assertEquals(Collections.singletonList("hi"), received, "成功路径应把 payload 投给业务");
+    }
+}