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/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java index a6358539f4fa2..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,7 +597,10 @@ private static void recursiveTTLCheckForTableDir( checkTTLAndDeleteExpiredObjectFile(currentFile, basicFileAttributes, lowerBoundInMS); return; } - } catch (IOException ignored) { + } catch (FileNotFoundException | NoSuchFileException ignored) { + // may be deleted by other thread + } catch (IOException e) { + logger.warn("Failed to read file attributes: {}", currentFile, e); } } File[] children = currentFile.listFiles(); @@ -606,8 +611,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); + } } } @@ -637,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); } } 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..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 @@ -72,9 +72,26 @@ 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); + } catch (Throwable t) { + logger.error( + "[CompactionScheduleTaskWorker-{}] Failed to execute compaction schedule task and cannot recover", + workerId, + t); + 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 e15a908ac24b5..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 @@ -78,8 +78,21 @@ 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); + } catch (Throwable t) { + logger.error( + "[TTLCheckTask-{}] Failed to execute ttl check and cannot recover", workerId, t); + throw t; } } }