From 3cfb87f347d542761d759cfca247a4e52391b42d Mon Sep 17 00:00:00 2001
From: "review512jwy@163.com" <“review512jwy@163.com”>
Date: Sun, 12 Jul 2026 11:36:19 +0800
Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84group=E5=AD=98=E5=9C=A8?=
 =?UTF-8?q?=E5=88=A4=E5=AE=9A?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../listener/RedisStreamConsumer.java         | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
index c447477..6bf11ba 100644
--- a/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
+++ b/src/main/java/com/techsor/datacenter/receiver/listener/RedisStreamConsumer.java
@@ -61,7 +61,7 @@ public class RedisStreamConsumer {
     private final ExecutorService consumerExecutor =
             Executors.newFixedThreadPool(PARTITIONS, new CustomizableThreadFactory("redis-stream-consumer-"));
 
-    private final ExecutorService workerExecutor =
+    private final ThreadPoolExecutor workerExecutor =
             new ThreadPoolExecutor(
                     16,
                     64,
@@ -160,7 +160,9 @@ public class RedisStreamConsumer {
                     .createGroup(stream, ReadOffset.from("0-0"), GROUP);
             log.info("create group success stream={}", stream);
         } catch (Exception e) {
-            if (e.getMessage() != null && e.getCause().getMessage().contains("BUSYGROUP")) {
+            Throwable cause = e.getCause();
+            if ((e.getMessage() != null && e.getMessage().contains("BUSYGROUP"))
+                    || (cause != null && cause.getMessage() != null && cause.getMessage().contains("BUSYGROUP"))) {
                 log.info("group exists {}", stream);
             } else {
                 throw e;
@@ -360,6 +362,19 @@ public class RedisStreamConsumer {
         boolean reclaimTerminated = reclaimScheduler.awaitTermination(10, TimeUnit.SECONDS);
         boolean statsTerminated = statsScheduler.awaitTermination(10, TimeUnit.SECONDS);
 
+        if (!consumerTerminated) {
+            consumerExecutor.shutdownNow();
+        }
+        if (!workerTerminated) {
+            workerExecutor.shutdownNow();
+        }
+        if (!reclaimTerminated) {
+            reclaimScheduler.shutdownNow();
+        }
+        if (!statsTerminated) {
+            statsScheduler.shutdownNow();
+        }
+
         log.info("RedisStreamConsumer shutdown done, consumerTerminated={}, workerTerminated={}, reclaimTerminated={}, statsTerminated={}",
                 consumerTerminated,
                 workerTerminated,