From 0e8e3762d228eab96dbb920f68625a24f3502105 Mon Sep 17 00:00:00 2001
From: hongawen <83944980@qq.com>
Date: Wed, 15 Jul 2026 02:55:04 +0100
Subject: [PATCH] =?UTF-8?q?fix(mq-starter-core):=20driver.stop()=20?=
 =?UTF-8?q?=E5=BC=82=E5=B8=B8=E4=B8=8D=E5=86=8D=E8=B7=B3=E8=BF=87=E5=81=9C?=
 =?UTF-8?q?=E6=9C=BA=E5=BC=BA=E5=88=B7=E5=B0=BE=E5=B7=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

review 发现 #8:stop() 里 driver.stop() 裸调用,抛异常(如 RocketMQ shutdown
出错/Redis 连接失败)会跳过定时器关闭、全部 forceFlush、running=false——
缓冲里已逐条 ACK 的批直接丢,且 SmartLifecycle 回调没走完会卡 Spring 关闭。

修复:driver.stop() 包 try/catch,异常打 error 日志后继续走完停机流程。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../njcn/mq/container/MqListenerRegistry.java |  8 +++++-
 .../mq/container/MqListenerRegistryTest.java  | 27 +++++++++++++++++++
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java
index 206db4e..0acd9bb 100644
--- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java
@@ -146,7 +146,13 @@ public class MqListenerRegistry implements SmartLifecycle {
 
     @Override public synchronized void stop() {
         // 停机顺序：driver 先停（不再有新消息入缓冲）→ 定时器停 → 强刷残留批 → 置 running。
-        driver.stop();
+        // driver.stop 异常必须吞掉：否则跳过后面整个强刷尾巴（缓冲里已 ACK 的批直接丢），
+        // 且 SmartLifecycle 回调没走完会卡 Spring 关闭。
+        try {
+            driver.stop();
+        } catch (Exception e) {
+            log.error("[mq] driver 停止失败（吞掉，继续停机流程强刷残留批）", e);
+        }
         if (flushScheduler != null) {
             // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。
             flushScheduler.shutdown();
diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java
index 0b12de8..387f676 100644
--- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java
+++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java
@@ -223,6 +223,33 @@ class MqListenerRegistryTest {
         assertNull(reg.getFlushScheduler(), "停机后定时器应已关闭置空");
     }
 
+    /** fake driver：stop 时抛异常。 */
+    static class FailingStopDriver implements MqDriver {
+        MqSubscription captured;
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() { throw new RuntimeException("driver stop fail"); }
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
+
+    @Test
+    void stop_whenDriverStopThrows_stillFlushesLeftoverAndCompletes() throws Exception {
+        List<String> events = new ArrayList<>();
+        LeftoverBiz biz = new LeftoverBiz(events);
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        FailingStopDriver driver = new FailingStopDriver();
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints());
+        reg.start();
+        driver.captured.getListener().onMessage(MqMessage.of("k1", "", "a"));
+        driver.captured.getListener().onMessage(MqMessage.of("k2", "", "b")); // batchSize=3，2 条不满批
+
+        reg.stop(); // driver.stop 抛异常不得中断停机流程
+
+        assertEquals(Arrays.asList("flush:2"), events, "driver.stop 抛异常也必须强刷残留批（否则已 ACK 的批静默丢）");
+        assertNull(reg.getFlushScheduler(), "driver.stop 抛异常也应关闭定时器");
+        assertFalse(reg.isRunning(), "driver.stop 抛异常也应完成停机置 running=false");
+    }
+
     @Test
     void stop_waitsForInFlightTimedFlushBeforeReturning() throws Exception {
         BatchBiz biz = new BatchBiz();