Pipe: Allow idle Pipe worker threads to time out

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/IoTDBThreadPoolFactory.java

@@ -127,6 +127,23 @@ public static ExecutorService newFixedThreadPool(
         poolName);
   }
 
+  public static ExecutorService newFixedThreadPoolWithIdleThreadTimeout(
+      int nThreads, long keepAliveTime, TimeUnit unit, String poolName) {
+    logger.info(NEW_FIXED_THREAD_POOL_LOGGER_FORMAT, poolName, nThreads);
+
+    WrappedThreadPoolExecutor executor =
+        new WrappedThreadPoolExecutor(
+            nThreads,
+            nThreads,
+            keepAliveTime,
+            unit,
+            new LinkedBlockingQueue<>(),
+            new IoTThreadFactory(poolName),
+            poolName);
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
+  }
+
   /**
    * see {@link Executors#newSingleThreadExecutor(java.util.concurrent.ThreadFactory)}.
    *

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/execution/PipeSubtaskExecutor.java

@@ -38,11 +38,14 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 public abstract class PipeSubtaskExecutor {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeSubtaskExecutor.class);
 
+  private static final long WORKER_THREAD_KEEP_ALIVE_TIME_IN_SECONDS = 60L;
+
   private static final ExecutorService globalSubtaskCallbackListeningExecutor =
       IoTDBThreadPoolFactory.newSingleThreadExecutor(
           ThreadName.PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL.getName());
@@ -78,7 +81,11 @@ protected PipeSubtaskExecutor(
             : ThreadName.PIPE_SUBTASK_CALLBACK_EXECUTOR_POOL.getName();
     underlyingThreadPool =
         (WrappedThreadPoolExecutor)
-            IoTDBThreadPoolFactory.newFixedThreadPool(corePoolSize, workingThreadName);
+            IoTDBThreadPoolFactory.newFixedThreadPoolWithIdleThreadTimeout(
+                corePoolSize,
+                WORKER_THREAD_KEEP_ALIVE_TIME_IN_SECONDS,
+                TimeUnit.SECONDS,
+                workingThreadName);
     if (disableLogInThreadPool) {
       underlyingThreadPool.disableErrorLog();
     }

iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/IoTDBThreadPoolFactoryTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
 import org.apache.iotdb.commons.concurrent.WrappedRunnable;
+import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
 
 import org.apache.thrift.server.TThreadPoolServer;
 import org.apache.thrift.server.TThreadPoolServer.Args;
@@ -110,6 +111,21 @@ public void testNewCachedThreadPool() {
     }
   }
 
+  @Test
+  public void testNewFixedThreadPoolWithIdleThreadTimeout() throws Exception {
+    ExecutorService exec =
+        IoTDBThreadPoolFactory.newFixedThreadPoolWithIdleThreadTimeout(
+            1, 1, TimeUnit.MILLISECONDS, POOL_NAME);
+
+    exec.submit(() -> {}).get();
+    assertEquals(1, ((WrappedThreadPoolExecutor) exec).getLargestPoolSize());
+
+    Thread.sleep(100);
+    assertEquals(0, ((WrappedThreadPoolExecutor) exec).getPoolSize());
+
+    exec.shutdown();
+  }
+
   @Test
   public void testNewSingleThreadScheduledExecutor() throws InterruptedException {
     String reason = "(can be ignored in Tests) NewSingleThreadScheduledExecutor";