This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new d3cebd25da3 [To dev/1.3] Fix thread pool lifecycle leaks (#18703)
(#18720)
d3cebd25da3 is described below
commit d3cebd25da389d328b5d7df5c6299d92fc4abd82
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 11:01:02 2026 +0800
[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
---
.../db/partition/DataPartitionTableGenerator.java | 8 +++-
.../impl/DataNodeInternalRPCServiceImpl.java | 50 ++++++++++++----------
.../metrics/IoTDBInternalLocalReporter.java | 1 +
.../reporter/iotdb/IoTDBSessionReporter.java | 1 +
4 files changed, 36 insertions(+), 24 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/partition/DataPartitionTableGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/partition/DataPartitionTableGenerator.java
index 0261cc34889..ee4517a5648 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/partition/DataPartitionTableGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/partition/DataPartitionTableGenerator.java
@@ -120,7 +120,13 @@ public class DataPartitionTableGenerator {
}
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() {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 55ee38253ea..8adbb3ba3bb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -2916,31 +2916,35 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
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");
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
index 3e566fd6df7..89ff842364f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java
@@ -148,6 +148,7 @@ public class IoTDBInternalLocalReporter extends
IoTDBInternalReporter {
currentServiceFuture.cancel(true);
currentServiceFuture = null;
}
+ service.shutdownNow();
clear();
LOGGER.info("IoTDBInternalReporter stop!");
return true;
diff --git
a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBSessionReporter.java
b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBSessionReporter.java
index 3603bac1a81..920bb213aa5 100644
---
a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBSessionReporter.java
+++
b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/reporter/iotdb/IoTDBSessionReporter.java
@@ -128,6 +128,7 @@ public class IoTDBSessionReporter extends IoTDBReporter {
currentServiceFuture.cancel(true);
currentServiceFuture = null;
}
+ service.shutdownNow();
if (sessionPool != null) {
sessionPool.close();
}
