Reduce pipe load success log noise

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java

@@ -229,7 +229,11 @@ public Analysis analyzeFileByFile(Analysis analysis) {
       return analysis;
     }
 
-    LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed.");
+    if (isGeneratedByPipe) {
+      LOGGER.debug("Load - Analysis Stage: all tsfiles have been analyzed.");
+    } else {
+      LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed.");
+    }
 
     if (reconstructStatementIfMiniFileConverted()) {
       // All mini tsfiles are converted to tablets, so the analysis is finished.
@@ -287,22 +291,14 @@ private boolean doAnalyzeFileByFile(Analysis analysis) {
         if (LOGGER.isWarnEnabled()) {
           LOGGER.warn("TsFile {} is empty.", tsFile.getPath());
         }
-        if (LOGGER.isInfoEnabled()) {
-          LOGGER.info(
-              "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
-              i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
-        }
+        logAnalyzeProgress(i + 1, tsfileNum);
         continue;
       }
 
       final long startTime = System.nanoTime();
       try {
         analyzeSingleTsFile(tsFile, i);
-        if (LOGGER.isInfoEnabled()) {
-          LOGGER.info(
-              "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
-              i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
-        }
+        logAnalyzeProgress(i + 1, tsfileNum);
       } catch (AuthException e) {
         setFailAnalysisForAuthException(analysis, e);
         return false;
@@ -339,6 +335,26 @@ private boolean doAnalyzeFileByFile(Analysis analysis) {
     return true;
   }
 
+  private void logAnalyzeProgress(final int analyzedTsFileNum, final int totalTsFileNum) {
+    if (isGeneratedByPipe && !LOGGER.isDebugEnabled()) {
+      return;
+    }
+    if (!isGeneratedByPipe && !LOGGER.isInfoEnabled()) {
+      return;
+    }
+
+    final String progress = String.format("%.3f", analyzedTsFileNum * 100.00 / totalTsFileNum);
+    if (isGeneratedByPipe) {
+      LOGGER.debug(
+          "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
+          analyzedTsFileNum, totalTsFileNum, progress);
+    } else {
+      LOGGER.info(
+          "Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
+          analyzedTsFileNum, totalTsFileNum, progress);
+    }
+  }
+
   private void analyzeSingleTsFile(final File tsFile, int index) throws Exception {
     try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) {
       // can be reused when constructing tsfile resource

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java

@@ -69,7 +69,7 @@
 
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 
-public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
+public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCloseable {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsFileDispatcherImpl.class);
 
@@ -83,7 +83,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
   private final int localhostInternalPort;
   private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
       internalServiceClientManager;
-  private final ExecutorService executor;
+  private ExecutorService executor;
   private final boolean isGeneratedByPipe;
 
   public LoadTsFileDispatcherImpl(
@@ -92,36 +92,44 @@ public LoadTsFileDispatcherImpl(
     this.internalServiceClientManager = internalServiceClientManager;
     this.localhostIpAddr = IoTDBDescriptor.getInstance().getConfig().getInternalAddress();
     this.localhostInternalPort = IoTDBDescriptor.getInstance().getConfig().getInternalPort();
-    this.executor =
-        IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName());
     this.isGeneratedByPipe = isGeneratedByPipe;
   }
 
+  private synchronized ExecutorService getOrCreateExecutor() {
+    if (executor == null || executor.isShutdown()) {
+      executor =
+          IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName());
+    }
+    return executor;
+  }
+
   public void setUuid(String uuid) {
     this.uuid = uuid;
   }
 
   @Override
   public Future<FragInstanceDispatchResult> dispatch(
       SubPlan root, List<FragmentInstance> instances) {
-    return executor.submit(
-        () -> {
-          for (FragmentInstance instance : instances) {
-            try (SetThreadName threadName =
-                new SetThreadName(
-                    "load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) {
-              dispatchOneInstance(instance);
-            } catch (FragmentInstanceDispatchException e) {
-              return new FragInstanceDispatchResult(e.getFailureStatus());
-            } catch (Exception t) {
-              LOGGER.warn("cannot dispatch FI for load operation", t);
-              return new FragInstanceDispatchResult(
-                  RpcUtils.getStatus(
-                      TSStatusCode.INTERNAL_SERVER_ERROR, "Unexpected errors: " + t.getMessage()));
-            }
-          }
-          return new FragInstanceDispatchResult(true);
-        });
+    return getOrCreateExecutor()
+        .submit(
+            () -> {
+              for (FragmentInstance instance : instances) {
+                try (SetThreadName threadName =
+                    new SetThreadName(
+                        "load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) {
+                  dispatchOneInstance(instance);
+                } catch (FragmentInstanceDispatchException e) {
+                  return new FragInstanceDispatchResult(e.getFailureStatus());
+                } catch (Exception t) {
+                  LOGGER.warn("cannot dispatch FI for load operation", t);
+                  return new FragInstanceDispatchResult(
+                      RpcUtils.getStatus(
+                          TSStatusCode.INTERNAL_SERVER_ERROR,
+                          "Unexpected errors: " + t.getMessage()));
+                }
+              }
+              return new FragInstanceDispatchResult(true);
+            });
   }
 
   private void dispatchOneInstance(FragmentInstance instance)
@@ -147,7 +155,11 @@ private void dispatchOneInstance(FragmentInstance instance)
   }
 
   public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException {
-    LOGGER.info("Receive load node from uuid {}.", uuid);
+    if (isGeneratedByPipe) {
+      LOGGER.debug("Receive load node from uuid {}.", uuid);
+    } else {
+      LOGGER.info("Receive load node from uuid {}.", uuid);
+    }
 
     ConsensusGroupId groupId =
         ConsensusGroupId.Factory.createFromTConsensusGroupId(
@@ -352,6 +364,14 @@ private static void adjustTimeoutIfNecessary(Throwable e) {
 
   @Override
   public void abort() {
-    // Do nothing
+    close();
+  }
+
+  @Override
+  public synchronized void close() {
+    if (executor != null) {
+      executor.shutdownNow();
+      executor = null;
+    }
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java

@@ -242,11 +242,19 @@ public void start() {
 
           if (isLoadSingleTsFileSuccess) {
             node.clean();
-            LOGGER.info(
-                "Load TsFile {} Successfully, load process [{}/{}]",
-                filePath,
-                i + 1,
-                tsFileNodeListSize);
+            if (isGeneratedByPipe) {
+              LOGGER.debug(
+                  "Load TsFile {} Successfully, load process [{}/{}]",
+                  filePath,
+                  i + 1,
+                  tsFileNodeListSize);
+            } else {
+              LOGGER.info(
+                  "Load TsFile {} Successfully, load process [{}/{}]",
+                  filePath,
+                  i + 1,
+                  tsFileNodeListSize);
+            }
           } else {
             isLoadSuccess = false;
             failedTsFileNodeIndexes.add(i);
@@ -299,6 +307,7 @@ public void start() {
         }
       }
     } finally {
+      dispatcher.close();
       LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock();
     }
   }
@@ -387,7 +396,11 @@ private boolean dispatchOnePieceNode(
 
   private boolean secondPhase(
       boolean isFirstPhaseSuccess, String uuid, TsFileResource tsFileResource) {
-    LOGGER.info("Start dispatching Load command for uuid {}", uuid);
+    if (isGeneratedByPipe) {
+      LOGGER.debug("Start dispatching Load command for uuid {}", uuid);
+    } else {
+      LOGGER.info("Start dispatching Load command for uuid {}", uuid);
+    }
     final File tsFile = tsFileResource.getTsFile();
     final TLoadCommandReq loadCommandReq =
         new TLoadCommandReq(
@@ -594,7 +607,7 @@ private void convertFailedTsFilesToTabletsAndRetry() {
 
   @Override
   public void stop(Throwable t) {
-    // Do nothing
+    dispatcher.abort();
   }
 
   @Override

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

@@ -3244,10 +3244,17 @@ public void loadNewTsFile(
               0);
 
       if (!newFileName.equals(tsfileToBeInserted.getName())) {
-        logger.info(
-            "TsFile {} must be renamed to {} for loading into the unsequence list.",
-            tsfileToBeInserted.getName(),
-            newFileName);
+        if (isGeneratedByPipe) {
+          logger.debug(
+              "TsFile {} must be renamed to {} for loading into the unsequence list.",
+              tsfileToBeInserted.getName(),
+              newFileName);
+        } else {
+          logger.info(
+              "TsFile {} must be renamed to {} for loading into the unsequence list.",
+              tsfileToBeInserted.getName(),
+              newFileName);
+        }
         newTsFileResource.setFile(
             fsFactory.getFile(tsfileToBeInserted.getParentFile(), newFileName));
       }
@@ -3282,7 +3289,11 @@ public void loadNewTsFile(
       }
 
       onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader);
-      logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
+      if (isGeneratedByPipe) {
+        logger.debug("TsFile {} is successfully loaded in unsequence list.", newFileName);
+      } else {
+        logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
+      }
     } catch (final DiskSpaceInsufficientException e) {
       logger.error(
           "Failed to append the tsfile {} to database processor {} because the disk space is insufficient.",
@@ -3451,10 +3462,17 @@ private boolean loadTsFileToUnSequence(
       return false;
     }
 
-    logger.info(
-        "Load tsfile in unsequence list, move file from {} to {}",
-        tsFileToLoad.getAbsolutePath(),
-        targetFile.getAbsolutePath());
+    if (isGeneratedByPipe) {
+      logger.debug(
+          "Load tsfile in unsequence list, move file from {} to {}",
+          tsFileToLoad.getAbsolutePath(),
+          targetFile.getAbsolutePath());
+    } else {
+      logger.info(
+          "Load tsfile in unsequence list, move file from {} to {}",
+          tsFileToLoad.getAbsolutePath(),
+          targetFile.getAbsolutePath());
+    }
 
     LoadTsFileRateLimiter.getInstance().acquire(tsFileResource.getTsFile().length());
 

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java

@@ -523,7 +523,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
       return;
     }
 
-    LOGGER.info(
+    LOGGER.debug(
         "Receiver id = {}: Writing file {} is not existed or name is not correct, try to create it. "
             + "Current writing file is {}.",
         receiverId.get(),
@@ -541,7 +541,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
     // This may be useless, because receiver file dir is created when handshake. just in case.
     if (!receiverFileDirWithIdSuffix.get().exists()) {
       if (receiverFileDirWithIdSuffix.get().mkdirs()) {
-        LOGGER.info(
+        LOGGER.debug(
             "Receiver id = {}: Receiver file dir {} was created.",
             receiverId.get(),
             receiverFileDirWithIdSuffix.get().getPath());
@@ -556,7 +556,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
 
     writingFile = targetPath.toFile();
     writingFileWriter = new RandomAccessFile(writingFile, "rw");
-    LOGGER.info(
+    LOGGER.debug(
         "Receiver id = {}: Writing file {} was created. Ready to write file pieces.",
         receiverId.get(),
         writingFile.getPath());
@@ -730,7 +730,7 @@ protected final TPipeTransferResp handleTransferFileSealV1(final PipeTransferFil
       final TSStatus status = loadFileV1(req, fileAbsolutePath);
       if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
         shouldDeleteSealedFile = false;
-        LOGGER.info(
+        LOGGER.debug(
             "Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePath);
       } else {
         PipeLogger.log(
@@ -835,7 +835,7 @@ protected final TPipeTransferResp handleTransferFileSealV2(final PipeTransferFil
 
       final TSStatus status = loadFileV2(req, fileAbsolutePaths);
       if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-        LOGGER.info(
+        LOGGER.debug(
             "Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePaths);
       } else {
         PipeLogger.log(

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java

@@ -120,14 +120,14 @@ public static void cleanPipeReceiverDir(final File receiverFileDir) {
             FileUtils.deleteDirectory(receiverFileDir);
             return null;
           });
-      LOGGER.info("Clean pipe receiver dir {} successfully.", receiverFileDir);
+      LOGGER.debug("Clean pipe receiver dir {} successfully.", receiverFileDir);
     } catch (final Exception e) {
       LOGGER.warn("Clean pipe receiver dir {} failed.", receiverFileDir, e);
     }
 
     try {
       FileUtils.forceMkdir(receiverFileDir);
-      LOGGER.info("Create pipe receiver dir {} successfully.", receiverFileDir);
+      LOGGER.debug("Create pipe receiver dir {} successfully.", receiverFileDir);
     } catch (final IOException e) {
       LOGGER.warn("Create pipe receiver dir {} failed.", receiverFileDir, e);
     }