Fix: send set configuration only to target nodes and harden compaction schedule interruption handling

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();
+ }
  }
  }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java

@@ -1052,7 +1052,7 @@ public List<TSStatus> setConfiguration(TSetConfigurationReq req) {
  if (!targetDataNodes.isEmpty()) {
  DataNodeAsyncRequestContext<Object, TSStatus> clientHandler =
  new DataNodeAsyncRequestContext<>(
- CnToDnAsyncRequestType.SET_CONFIGURATION, req, dataNodeLocationMap);
+ CnToDnAsyncRequestType.SET_CONFIGURATION, req, targetDataNodes);
  CnToDnInternalServiceAsyncRequestManager.getInstance()
  .sendAsyncRequestWithRetry(clientHandler);
  responseList.addAll(clientHandler.getResponseList());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -2453,7 +2453,7 @@
  * lock if return true, the caller is responsible for unlocking all the already-acquiring lock
  * in needToUnLockList
  */
  private boolean tryGetFLushLock(
  long waitTimeInMs,
  IDeviceID singleDeviceId,
  Filter globalTimeFilter,
@@ -2891,7 +2891,7 @@
  }
  }
 
  public void deleteByTable(RelationalDeleteDataNode node) throws IOException {
  if (node.getDatabaseName() != null && !node.getDatabaseName().equals(databaseName)) {
  // not targeted on this database, return
  return;
@@ -3277,7 +3277,7 @@
  }
  }
 
  private void deleteDataInSealedFiles(Collection<TsFileResource> sealedTsFiles, ModEntry deletion)
  throws IOException {
  Set<ModificationFile> involvedModificationFiles = new HashSet<>();
  List<TsFileResource> deletedByMods = new ArrayList<>();
@@ -3705,7 +3705,11 @@
  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);

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);
  }
  }
 

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<Void> 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;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduleTaskWorker.java

@@ -49,7 +49,7 @@
 
  @Override
  @SuppressWarnings("java:S2142")
  public Void call() {
  while (true) {
  try {
  Thread.sleep(IoTDBDescriptor.getInstance().getConfig().getCompactionScheduleIntervalInMs());
@@ -72,9 +72,26 @@
  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;
  }
  }
  }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/TTLScheduleTask.java

@@ -78,8 +78,21 @@
  }
  }
  } 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;
  }
  }
  }