fix(mq-starter-core): driver.stop() 异常不再跳过停机强刷尾巴

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>

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();

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();