完善group存在判定

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,