feat: log stream bug fix

ozhera-log/log-stream/pom.xml

@@ -28,7 +28,7 @@ http://www.apache.org/licenses/LICENSE-2.0
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>log-stream</artifactId>
-    <version>2.2.7-SNAPSHOT</version>
+    <version>2.2.8-SNAPSHOT</version>
 
     <properties>
         <maven.compiler.source>21</maven.compiler.source>

ozhera-log/log-stream/src/main/java/org/apache/ozhera/log/stream/plugin/es/EsPlugin.java

@@ -227,8 +227,20 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
                     return;
                 }
                 if (failure.getMessage().contains("I/O reactor status: STOPPED")) {
-                    esServiceMap.remove(cacheKey(esInfo));
+                    EsService staleService = esServiceMap.remove(cacheKey(esInfo));
                     esProcessorMap.remove(cacheKey(esInfo));
+                    // Close the stale EsService in a background thread to release I/O dispatcher
+                    // threads. Without this, every reactor-stopped event leaks a set of threads
+                    // (visible as "I/O dispatcher N" in thread dumps).
+                    if (staleService != null) {
+                        Thread.ofVirtual().start(() -> {
+                            try {
+                                staleService.getEsClient().close();
+                            } catch (Exception ex) {
+                                log.warn("[EsPlugin] close stale EsClient error, addr:{}", esInfo.getAddr(), ex);
+                            }
+                        });
+                    }
                 }
                 log.error(String.format("fail send %s message to es,desc:%s,es addr:%s", request.numberOfActions(), request.getDescription(), esInfo.getAddr()), new RuntimeException(failure));
                 Class clazz = failure.getClass();