From 09d5ac4788d4440c03ebb52447abcf2f1ed6c5b0 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 11:00:27 +0800 Subject: [PATCH 1/9] Fix: send set configuration only to target nodes and harden compaction schedule interruption handling --- .../it/db/it/IoTDBSetConfigurationTableIT.java | 14 ++++++++++++++ .../confignode/manager/node/NodeManager.java | 2 +- .../db/storageengine/dataregion/DataRegion.java | 6 +++++- .../schedule/CompactionScheduleTaskManager.java | 9 +++++++++ .../schedule/CompactionScheduleTaskWorker.java | 15 +++++++++++++-- .../compaction/schedule/TTLScheduleTask.java | 13 +++++++++++-- 6 files changed, 53 insertions(+), 6 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetConfigurationTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetConfigurationTableIT.java index 2295b88f1bc52..8a4bc388a59fd 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetConfigurationTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBSetConfigurationTableIT.java @@ -109,6 +109,9 @@ public void testSetConfiguration() { statement.execute( "set configuration inner_compaction_candidate_file_num='1',max_cross_compaction_candidate_file_num='1' on " + dnId); + if (dnId == 5) { + statement.execute("set configuration compaction_schedule_thread_num='2' on 5"); + } } } catch (Exception e) { Assert.fail(e.getMessage()); @@ -131,6 +134,17 @@ public void testSetConfiguration() { "enable_cross_space_compaction=false", "inner_compaction_candidate_file_num=1", "max_cross_compaction_candidate_file_num=1")); + boolean scheduleThreadNumChanged = + checkConfigFileContains( + dnId, + EnvFactory.getEnv().getDataNodeWrapperList().get(i), + "compaction_schedule_thread_num=2"); + if (scheduleThreadNumChanged && dnId != 5) { + Assert.fail(); + } + if (!scheduleThreadNumChanged && dnId == 5) { + Assert.fail(); + } } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java index e3d775259d626..2e50c6b787b20 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java @@ -1052,7 +1052,7 @@ public List setConfiguration(TSetConfigurationReq req) { if (!targetDataNodes.isEmpty()) { DataNodeAsyncRequestContext clientHandler = new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.SET_CONFIGURATION, req, dataNodeLocationMap); + CnToDnAsyncRequestType.SET_CONFIGURATION, req, targetDataNodes); CnToDnInternalServiceAsyncRequestManager.getInstance() .sendAsyncRequestWithRetry(clientHandler); responseList.addAll(clientHandler.getResponseList()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index eb37cb1d5d21b..c65105ae54c8d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -3705,7 +3705,11 @@ public void executeTTLCheckForObjectFiles() throws InterruptedException { if (!regionObjectDir.isDirectory()) { continue; } - CompactionUtils.executeTTLCheckObjectFilesForTableModel(regionObjectDir, databaseName); + try { + CompactionUtils.executeTTLCheckObjectFilesForTableModel(regionObjectDir, databaseName); + } catch (Exception e) { + logger.error("Failed to execute object ttl check", e); + } } CompactionMetrics.getInstance() .updateTTLCheckForObjectFileCost(System.currentTimeMillis() - startTime); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java index 8348b87813768..516b1489a206d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskManager.java @@ -65,6 +65,7 @@ public class CompactionScheduleTaskManager implements IService { ConcurrentHashMap.newKeySet(); private ReentrantLock lock = new ReentrantLock(); private volatile boolean init = false; + private volatile boolean isStoppingAllScheduleTask = false; @Override public void start() throws StartupException { @@ -76,8 +77,13 @@ public void start() throws StartupException { logger.info("Compaction schedule task manager started."); } + public boolean isStoppingAllScheduleTask() { + return isStoppingAllScheduleTask; + } + public void stopCompactionScheduleTasks() throws InterruptedException { lock.lock(); + isStoppingAllScheduleTask = true; try { for (Future task : submitCompactionScheduleTaskFutures) { task.cancel(true); @@ -121,6 +127,7 @@ public void checkAndMayApplyConfigurationChange() throws InterruptedException { public void startScheduleTasks() { lock.lock(); + isStoppingAllScheduleTask = false; try { // compaction selector for (int workerId = 0; workerId < compactionSelectorNum; workerId++) { @@ -144,6 +151,7 @@ public void startScheduleTasks() { @Override public void stop() { lock.lock(); + isStoppingAllScheduleTask = true; try { if (!init) { return; @@ -160,6 +168,7 @@ public void stop() { @Override public void waitAndStop(long milliseconds) { lock.lock(); + isStoppingAllScheduleTask = true; try { if (!init) { return; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java index 17ad0dd4334e6..f5646fbed4a2c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java @@ -72,9 +72,20 @@ public Void call() { dataRegion.executeCompaction(); } } catch (InterruptedException ignored) { + boolean isStoppedByUser = + CompactionScheduleTaskManager.getInstance().isStoppingAllScheduleTask(); logger.info( - "[CompactionScheduleTaskWorker-{}] compaction schedule is interrupted", workerId); - return null; + "[CompactionScheduleTaskWorker-{}] compaction schedule is interrupted, isStopByUser: {}", + workerId, + isStoppedByUser); + if (isStoppedByUser) { + return null; + } + } catch (Exception e) { + logger.error( + "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task", + workerId, + e); } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java index e15a908ac24b5..c7b5271f352d0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java @@ -78,8 +78,17 @@ public Void call() throws Exception { } } } catch (InterruptedException ignored) { - logger.info("[TTLCheckTask-{}] TTL checker is interrupted", workerId); - return null; + boolean isStoppedByUser = + CompactionScheduleTaskManager.getInstance().isStoppingAllScheduleTask(); + logger.info( + "[TTLCheckTask-{}] TTL checker is interrupted, isStoppedByUser: {}", + workerId, + isStoppedByUser); + if (isStoppedByUser) { + return null; + } + } catch (Exception e) { + logger.error("[TTLCheckTask-{}] Failed to execute ttl check", workerId, e); } } } From b43d4a8d87f2d65e96597bd98edae00d94d8a80d Mon Sep 17 00:00:00 2001 From: shuwenwei <55970239+shuwenwei@users.noreply.github.com> Date: Tue, 14 Apr 2026 12:00:10 +0800 Subject: [PATCH 2/9] Update iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java Co-authored-by: Jackie Tien --- .../compaction/schedule/CompactionScheduleTaskWorker.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java index f5646fbed4a2c..7e0d730c35348 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java @@ -81,7 +81,7 @@ public Void call() { if (isStoppedByUser) { return null; } - } catch (Exception e) { + } catch (Throwable e) { logger.error( "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task", workerId, From b3c53037ce22a430cdacd711b6d9bf2fee3c211b Mon Sep 17 00:00:00 2001 From: shuwenwei <55970239+shuwenwei@users.noreply.github.com> Date: Tue, 14 Apr 2026 12:00:18 +0800 Subject: [PATCH 3/9] Update iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java Co-authored-by: Jackie Tien --- .../dataregion/compaction/schedule/TTLScheduleTask.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java index c7b5271f352d0..93481d6482aed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java @@ -87,7 +87,7 @@ public Void call() throws Exception { if (isStoppedByUser) { return null; } - } catch (Exception e) { + } catch (Throwable e) { logger.error("[TTLCheckTask-{}] Failed to execute ttl check", workerId, e); } } From 75630bc794583d6749031e49687ecccd548750c9 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 12:03:33 +0800 Subject: [PATCH 4/9] catch throwable --- .../compaction/schedule/CompactionScheduleTaskWorker.java | 7 ++++++- .../dataregion/compaction/schedule/TTLScheduleTask.java | 5 ++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java index 7e0d730c35348..0b28809a8e0ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java @@ -81,11 +81,16 @@ public Void call() { if (isStoppedByUser) { return null; } - } catch (Throwable e) { + } catch (Exception e) { logger.error( "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task", workerId, e); + } catch (Throwable t) { + logger.error( + "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task and cannot recover", + workerId, + t); } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java index 93481d6482aed..b05841df37343 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java @@ -87,8 +87,11 @@ public Void call() throws Exception { if (isStoppedByUser) { return null; } - } catch (Throwable e) { + } catch (Exception e) { logger.error("[TTLCheckTask-{}] Failed to execute ttl check", workerId, e); + } catch (Throwable t) { + logger.error( + "[TTLCheckTask-{}] Failed to execute ttl check and cannot recover", workerId, t); } } } From fd084f9009fc57bea01ddc78c11b410097397d8d Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 12:04:49 +0800 Subject: [PATCH 5/9] fix --- .../compaction/schedule/CompactionScheduleTaskWorker.java | 1 + 1 file changed, 1 insertion(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java index 0b28809a8e0ee..fb8f73ad80bf1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java @@ -91,6 +91,7 @@ public Void call() { "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task and cannot recover", workerId, t); + return null; } } } From 39282478ad1e7e88af34049fb2e3678403dc8f81 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 12:05:05 +0800 Subject: [PATCH 6/9] fix --- .../dataregion/compaction/schedule/TTLScheduleTask.java | 1 + 1 file changed, 1 insertion(+) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java index b05841df37343..902daa8d4e13f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java @@ -92,6 +92,7 @@ public Void call() throws Exception { } catch (Throwable t) { logger.error( "[TTLCheckTask-{}] Failed to execute ttl check and cannot recover", workerId, t); + return null; } } } From 05a79fa7f758395ff766246dc887440feb131d02 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 12:06:03 +0800 Subject: [PATCH 7/9] throw Throwable --- .../compaction/schedule/CompactionScheduleTaskWorker.java | 2 +- .../dataregion/compaction/schedule/TTLScheduleTask.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java index fb8f73ad80bf1..714d232d62538 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java @@ -91,7 +91,7 @@ public Void call() { "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task and cannot recover", workerId, t); - return null; + throw t; } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java index 902daa8d4e13f..565f7077cafaa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java @@ -92,7 +92,7 @@ public Void call() throws Exception { } catch (Throwable t) { logger.error( "[TTLCheckTask-{}] Failed to execute ttl check and cannot recover", workerId, t); - return null; + throw t; } } } From fd2087a0507833d9d1de9a3ebbaffa807490ec9e Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 15:00:02 +0800 Subject: [PATCH 8/9] add catch block for recursive method --- .../compaction/execute/utils/CompactionUtils.java | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java index a6358539f4fa2..fa4c5f595eadd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java @@ -595,7 +595,8 @@ private static void recursiveTTLCheckForTableDir( checkTTLAndDeleteExpiredObjectFile(currentFile, basicFileAttributes, lowerBoundInMS); return; } - } catch (IOException ignored) { + } catch (IOException e) { + logger.warn("Failed to read file attributes: {}", currentFile, e); } } File[] children = currentFile.listFiles(); @@ -606,8 +607,16 @@ private static void recursiveTTLCheckForTableDir( // block-aligned and reflects allocated directory entry blocks. acquireCompactionReadRate(currentFile.length()); for (File child : children) { - recursiveTTLCheckForTableDir( - child, depth + 1, maxObjectFileDepth, canDistinguishDirectoryByFileName, lowerBoundInMS); + try { + recursiveTTLCheckForTableDir( + child, + depth + 1, + maxObjectFileDepth, + canDistinguishDirectoryByFileName, + lowerBoundInMS); + } catch (Exception e) { + logger.warn("Failed to check table dir: {}", child, e); + } } } From 93777912f4d33e1c7eeb3ca36a5e8d1b667275fe Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Tue, 14 Apr 2026 15:11:39 +0800 Subject: [PATCH 9/9] add catch block for recursive method --- .../compaction/execute/utils/CompactionUtils.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java index fa4c5f595eadd..3cd21d0e0a7a8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java @@ -69,9 +69,11 @@ import javax.annotation.Nullable; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.Collection; @@ -595,6 +597,8 @@ private static void recursiveTTLCheckForTableDir( checkTTLAndDeleteExpiredObjectFile(currentFile, basicFileAttributes, lowerBoundInMS); return; } + } catch (FileNotFoundException | NoSuchFileException ignored) { + // may be deleted by other thread } catch (IOException e) { logger.warn("Failed to read file attributes: {}", currentFile, e); } @@ -646,7 +650,10 @@ private static void checkTTLAndDeleteExpiredObjectFile( FileMetrics.getInstance().decreaseObjectFileNum(1); FileMetrics.getInstance().decreaseObjectFileSize(attributes.size()); logger.info("Remove object file {}, size is {}(byte)", file.getPath(), attributes.size()); - } catch (Exception ignored) { + } catch (FileNotFoundException | NoSuchFileException ignored) { + // may be deleted by other thread + } catch (Exception e) { + logger.warn("Failed to delete expired object file: {}", file, e); } }