[To dev/1.3] Fix thread pool lifecycle leaks (#18703) (#18720)

* Fix thread pool lifecycle leaks (#18703)

* Adapt thread pool cherry-pick for dev/1.3

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/partition/DataPartitionTableGenerator.java

@@ -120,7 +120,13 @@ public CompletableFuture<Void> startGeneration() {
     }
 
     status = TaskStatus.IN_PROGRESS;
-    return CompletableFuture.runAsync(this::generateDataPartitionTableByMemory);
+    return CompletableFuture.runAsync(this::generateDataPartitionTableByMemory)
+        .whenComplete((ignored, throwable) -> close());
+  }
+
+  /** Close the executor owned by this generator after generation has finished. */
+  public void close() {
+    executor.shutdownNow();
   }
 
   private void generateDataPartitionTableByMemory() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java

@@ -2916,31 +2916,35 @@ private void processDataRegionForEarliestTimeslots(Map<String, Long> earliestTim
             ThreadName.FIND_EARLIEST_TIME_SLOT_PARALLEL_POOL.getName(),
             new ThreadPoolExecutor.CallerRunsPolicy());
 
-    for (DataRegion dataRegion : StorageEngine.getInstance().getAllDataRegions()) {
-      CompletableFuture<Void> regionFuture =
-          CompletableFuture.runAsync(
-              () -> {
-                TsFileManager tsFileManager = dataRegion.getTsFileManager();
-                String databaseName = dataRegion.getDatabaseName();
-                if (ignoreDatabase.contains(databaseName)) {
-                  return;
-                }
+    try {
+      for (DataRegion dataRegion : StorageEngine.getInstance().getAllDataRegions()) {
+        CompletableFuture<Void> regionFuture =
+            CompletableFuture.runAsync(
+                () -> {
+                  TsFileManager tsFileManager = dataRegion.getTsFileManager();
+                  String databaseName = dataRegion.getDatabaseName();
+                  if (ignoreDatabase.contains(databaseName)) {
+                    return;
+                  }
 
-                Set<Long> timePartitionIds = tsFileManager.getTimePartitions();
-                if (timePartitionIds.isEmpty()) {
-                  return;
-                }
-                final long earliestTimeSlotId = Collections.min(timePartitionIds);
-                earliestTimeslots.compute(
-                    databaseName,
-                    (k, v) -> v == null ? earliestTimeSlotId : Math.min(earliestTimeSlotId, v));
-              },
-              findEarliestTimeSlotExecutor);
-      futures.add(regionFuture);
-    }
+                  Set<Long> timePartitionIds = tsFileManager.getTimePartitions();
+                  if (timePartitionIds.isEmpty()) {
+                    return;
+                  }
+                  final long earliestTimeSlotId = Collections.min(timePartitionIds);
+                  earliestTimeslots.compute(
+                      databaseName,
+                      (k, v) -> v == null ? earliestTimeSlotId : Math.min(earliestTimeSlotId, v));
+                },
+                findEarliestTimeSlotExecutor);
+        futures.add(regionFuture);
+      }
 
-    // Wait for all tasks to complete
-    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+      // Wait for all tasks to complete
+      CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+    } finally {
+      findEarliestTimeSlotExecutor.shutdown();
+    }
     LOGGER.info("Process data directory for earliestTimeslots completed successfully");
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java

@@ -148,6 +148,7 @@ public boolean stop() {
       currentServiceFuture.cancel(true);
       currentServiceFuture = null;
     }
+    service.shutdownNow();
     clear();
     LOGGER.info("IoTDBInternalReporter stop!");
     return true;

iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBSessionReporter.java

@@ -128,6 +128,7 @@ public boolean stop() {
       currentServiceFuture.cancel(true);
       currentServiceFuture = null;
     }
+    service.shutdownNow();
     if (sessionPool != null) {
       sessionPool.close();
     }