diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java index 398a7f0e1e4f..f57342633408 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadConfigurationTableIT.java @@ -41,6 +41,7 @@ import java.nio.channels.FileChannel; import java.nio.file.StandardOpenOption; import java.sql.Connection; +import java.sql.ResultSet; import java.sql.Statement; import java.util.HashMap; import java.util.Map; @@ -101,11 +102,13 @@ public void showConfigurationDisplaysEffectiveValue() throws Exception { + "conf" + File.separator + "iotdb-system.properties"; + int dataNodeId = findDataNodeIdByRpcPort(dataNodeWrapper.getPort()); // Case 1: non-positive guard params (`if (x > 0) setX(x)`) must display the effective // default value, not the raw file value (0 / -1). Map afterGuardReload = appendLinesLoadAndShow( + dataNodeId, confPath, "cte_buffer_size_in_bytes=0", "max_rows_in_cte_buffer=-1", @@ -116,20 +119,20 @@ public void showConfigurationDisplaysEffectiveValue() throws Exception { // Case 2: a valid value is not rewritten, so display must equal the file value (regression). Map afterValidReload = - appendLinesLoadAndShow(confPath, "cte_buffer_size_in_bytes=262144"); + appendLinesLoadAndShow(dataNodeId, confPath, "cte_buffer_size_in_bytes=262144"); Assert.assertEquals("262144", afterValidReload.get("cte_buffer_size_in_bytes")); // Case 3: params whose template default is 0 are always rewritten to a computed default, // so they must never display 0 (this was always wrong, even without a bad file value). - Map defaultsReload = appendLinesLoadAndShow(confPath); + Map defaultsReload = appendLinesLoadAndShow(dataNodeId, confPath); assertPositiveNonZero(defaultsReload, "sort_buffer_size_in_bytes"); assertPositiveNonZero(defaultsReload, "mods_cache_size_limit_per_fi_in_bytes"); } // Appends the given lines to the config file, runs `load configuration`, fetches the // `show configuration` result, and restores the file to its original length. - private Map appendLinesLoadAndShow(String confPath, String... lines) - throws Exception { + private Map appendLinesLoadAndShow( + int dataNodeId, String confPath, String... lines) throws Exception { long length = new File(confPath).length(); try { if (lines.length > 0) { @@ -145,7 +148,7 @@ private Map appendLinesLoadAndShow(String confPath, String... li Statement statement = connection.createStatement()) { statement.execute("LOAD CONFIGURATION"); } - return fetchShowConfiguration(); + return fetchShowConfiguration(dataNodeId); } finally { try (FileChannel fileChannel = FileChannel.open(new File(confPath).toPath(), StandardOpenOption.WRITE)) { @@ -154,11 +157,24 @@ private Map appendLinesLoadAndShow(String confPath, String... li } } - private Map fetchShowConfiguration() throws Exception { + private int findDataNodeIdByRpcPort(int rpcPort) throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("SHOW DATANODES")) { + while (resultSet.next()) { + if (resultSet.getInt(4) == rpcPort) { + return resultSet.getInt(1); + } + } + } + throw new AssertionError("Cannot find DataNode id for RPC port: " + rpcPort); + } + + private Map fetchShowConfiguration(int dataNodeId) throws Exception { Map result = new HashMap<>(); try (ITableSession tableSessionConnection = EnvFactory.getEnv().getTableSessionConnection()) { SessionDataSet sessionDataSet = - tableSessionConnection.executeQueryStatement("show configuration"); + tableSessionConnection.executeQueryStatement("show configuration on " + dataNodeId); SessionDataSet.DataIterator iterator = sessionDataSet.iterator(); while (iterator.next()) { String name = iterator.getString(1); diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 7978fce51136..0fd91703ff7f 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -1112,6 +1112,8 @@ public final class DataNodeQueryMessages { "cancel query {} on node {} failed."; public static final String CANNOT_DISPATCH_FI_FOR_LOAD_OPERATION = "cannot dispatch FI for load operation"; + public static final String MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800 = + "Unexpected errors: %s"; public static final String RECEIVE_LOAD_NODE_FROM_UUID = "Receive load node from uuid {}."; public static final String LOAD_TSFILE_NODE_ERROR = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 34a7fa084bf3..a29165dc56fd 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -1096,6 +1096,8 @@ public final class DataNodeQueryMessages { "在节点 {} 上取消查询 {} 失败。"; public static final String CANNOT_DISPATCH_FI_FOR_LOAD_OPERATION = "无法为加载操作分发 FI"; + public static final String MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800 = + "意外错误：%s"; public static final String RECEIVE_LOAD_NODE_FROM_UUID = "接收来自 uuid {} 的加载节点。"; public static final String LOAD_TSFILE_NODE_ERROR = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java index 89b2f8f2c523..158f38875e1d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java @@ -237,7 +237,11 @@ public IAnalysis analyzeFileByFile(IAnalysis analysis) { return analysis; } - LOGGER.info(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + if (isGeneratedByPipe) { + LOGGER.debug(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + } else { + LOGGER.info(DataNodeQueryMessages.LOAD_ANALYSIS_STAGE_ALL_TSFILES_HAVE_BEEN_ANALYZED); + } setTsFileModelInfoToStatement(); if (reconstructStatementIfMiniFileConverted()) { @@ -319,28 +323,14 @@ private boolean doAnalyzeFileByFile(IAnalysis analysis) { if (LOGGER.isWarnEnabled()) { LOGGER.warn(DataNodeQueryMessages.TSFILE_IS_EMPTY, tsFile.getPath()); } - if (LOGGER.isInfoEnabled()) { - LOGGER.info( - DataNodeQueryMessages - .LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT, - 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( - DataNodeQueryMessages - .LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT, - i + 1, - tsfileNum, - String.format("%.3f", (i + 1) * 100.00 / tsfileNum)); - } + logAnalyzeProgress(i + 1, tsfileNum); } catch (AuthException e) { setFailAnalysisForAuthException(analysis, e); return false; @@ -379,16 +369,49 @@ private boolean doAnalyzeFileByFile(IAnalysis 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( + DataNodeQueryMessages + .LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT, + analyzedTsFileNum, + totalTsFileNum, + progress); + } else { + LOGGER.info( + DataNodeQueryMessages + .LOAD_ANALYSIS_STAGE_ARG_ARG_TSFILES_HAVE_BEEN_ANALYZED_PROGRESS_ARG_PERCENT, + analyzedTsFileNum, + totalTsFileNum, + progress); + } + } + private void analyzeSingleTsFile(final File tsFile, int i) throws Exception { final SessionInfo sessionInfo = context.getSession(); try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) { // check whether the tsfile is tree-model or not final Map tableSchemaMap = reader.getTableSchemaMap(); final boolean isTableModelFile = Objects.nonNull(tableSchemaMap) && !tableSchemaMap.isEmpty(); - LOGGER.info( - DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE, - tsFile.getPath(), - isTableModelFile ? "table" : "tree"); + if (isGeneratedByPipe) { + LOGGER.debug( + DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE, + tsFile.getPath(), + isTableModelFile ? "table" : "tree"); + } else { + LOGGER.info( + DataNodeQueryMessages.TSFILE_ARG_IS_A_ARG_MODEL_FILE, + tsFile.getPath(), + isTableModelFile ? "table" : "tree"); + } // can be reused when constructing tsfile resource final TsFileSequenceReaderTimeseriesMetadataIterator timeseriesMetadataIterator = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java index 0037e2f1ac93..8a39ad9d266c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/SchemaPredicateUtil.java @@ -226,7 +226,8 @@ static List extractTagSingleMatchExpressionCases( final int tagCount = tableInstance.getTagNum(); for (int i = 0; i < index2FilterMapList.size(); i++) { final Map> filterMap = index2FilterMapList.get(i); - if (filterMap.size() == tagCount + if (tagCount > 0 + && filterMap.size() == tagCount && filterMap.values().stream() .allMatch( filterList -> diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java index 9ff496521cb1..3f5020cccf31 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java @@ -74,7 +74,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); @@ -88,7 +88,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher { private final int localhostInternalPort; private final IClientManager internalServiceClientManager; - private final ExecutorService executor; + private ExecutorService executor; private final boolean isGeneratedByPipe; public LoadTsFileDispatcherImpl( @@ -97,11 +97,17 @@ 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; } @@ -109,24 +115,28 @@ public void setUuid(String uuid) { @Override public Future dispatch( SubPlan root, List 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(DataNodeQueryMessages.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(DataNodeQueryMessages.CANNOT_DISPATCH_FI_FOR_LOAD_OPERATION, t); + return new FragInstanceDispatchResult( + RpcUtils.getStatus( + TSStatusCode.INTERNAL_SERVER_ERROR, + String.format( + DataNodeQueryMessages.MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800, + t.getMessage()))); + } + } + return new FragInstanceDispatchResult(true); + }); } private void dispatchOneInstance(FragmentInstance instance) @@ -152,7 +162,11 @@ private void dispatchOneInstance(FragmentInstance instance) } public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException { - LOGGER.info(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid); + if (isGeneratedByPipe) { + LOGGER.debug(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid); + } else { + LOGGER.info(DataNodeQueryMessages.RECEIVE_LOAD_NODE_FROM_UUID, uuid); + } ConsensusGroupId groupId = ConsensusGroupId.Factory.createFromTConsensusGroupId( @@ -270,7 +284,10 @@ public Future dispatchCommand( return immediateFuture( new FragInstanceDispatchResult( RpcUtils.getStatus( - TSStatusCode.INTERNAL_SERVER_ERROR, "Unexpected errors: " + t.getMessage()))); + TSStatusCode.INTERNAL_SERVER_ERROR, + String.format( + DataNodeQueryMessages.MESSAGE_UNEXPECTED_ERRORS_ARG_78EE0800, + t.getMessage())))); } } return immediateFuture(new FragInstanceDispatchResult(true)); @@ -365,6 +382,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; + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java index e304ed2310c5..d3beefc0802b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java @@ -255,11 +255,19 @@ public void start() { if (isLoadSingleTsFileSuccess) { node.clean(); - LOGGER.info( - DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG, - filePath, - i + 1, - tsFileNodeListSize); + if (isGeneratedByPipe) { + LOGGER.debug( + DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG, + filePath, + i + 1, + tsFileNodeListSize); + } else { + LOGGER.info( + DataNodeQueryMessages.LOAD_TSFILE_ARG_SUCCESSFULLY_LOAD_PROCESS_ARG_ARG, + filePath, + i + 1, + tsFileNodeListSize); + } } else { isLoadSuccess = false; failedTsFileNodeIndexes.add(i); @@ -313,6 +321,7 @@ public void start() { } } } finally { + dispatcher.close(); LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock(); } } @@ -412,7 +421,11 @@ private boolean dispatchOnePieceNode( private boolean secondPhase( boolean isFirstPhaseSuccess, String uuid, TsFileResource tsFileResource) { - LOGGER.info(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + if (isGeneratedByPipe) { + LOGGER.debug(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + } else { + LOGGER.info(DataNodeQueryMessages.START_DISPATCHING_LOAD_COMMAND_FOR_UUID, uuid); + } final File tsFile = tsFileResource.getTsFile(); final TLoadCommandReq loadCommandReq = new TLoadCommandReq( @@ -686,7 +699,7 @@ private LoadTsFileStatement buildRetryTreeLoadStatement( @Override public void stop(Throwable t) { - // Do nothing + dispatcher.abort(); } @Override 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 9f8f11cdc504..b37790d01f84 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 @@ -4196,11 +4196,19 @@ public void loadNewTsFile( 0); if (!newFileName.equals(tsfileToBeInserted.getName())) { - logger.info( - StorageEngineMessages - .STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619, - tsfileToBeInserted.getName(), - newFileName); + if (isGeneratedByPipe) { + logger.debug( + StorageEngineMessages + .STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619, + tsfileToBeInserted.getName(), + newFileName); + } else { + logger.info( + StorageEngineMessages + .STORAGE_LOG_TSFILE_MUST_BE_RENAMED_TO_FOR_LOADING_INTO_THE_UNSEQUENCE_70321619, + tsfileToBeInserted.getName(), + newFileName); + } newTsFileResource.setFile( fsFactory.getFile(tsfileToBeInserted.getParentFile(), newFileName)); } @@ -4249,7 +4257,11 @@ public void loadNewTsFile( } onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader); - logger.info(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName); + if (isGeneratedByPipe) { + logger.debug(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName); + } else { + logger.info(StorageEngineMessages.TSFILE_LOADED_IN_UNSEQ_LIST, newFileName); + } } catch (final DiskSpaceInsufficientException e) { logger.error( StorageEngineMessages @@ -4405,10 +4417,19 @@ private boolean loadTsFileToUnSequence( return false; } - logger.info( - StorageEngineMessages.STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB, - tsFileToLoad.getAbsolutePath(), - targetFile.getAbsolutePath()); + if (isGeneratedByPipe) { + logger.debug( + StorageEngineMessages + .STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB, + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath()); + } else { + logger.info( + StorageEngineMessages + .STORAGE_LOG_LOAD_TSFILE_IN_UNSEQUENCE_LIST_MOVE_FILE_FROM_TO_21E11AEB, + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath()); + } LoadTsFileRateLimiter.getInstance().acquire(tsFileResource.getTsFile().length()); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java index f05e5069eed4..54edcc14f68d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java @@ -476,7 +476,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool return; } - LOGGER.info( + LOGGER.debug( PipeMessages.RECEIVER_WRITING_FILE_NOT_EXIST, receiverId.get(), fileName, @@ -493,7 +493,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( PipeMessages.RECEIVER_FILE_DIR_CREATED, receiverId.get(), receiverFileDirWithIdSuffix.get().getPath()); @@ -508,7 +508,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool writingFile = targetPath.toFile(); writingFileWriter = new RandomAccessFile(writingFile, "rw"); - LOGGER.info( + LOGGER.debug( PipeMessages.RECEIVER_WRITING_FILE_CREATED, receiverId.get(), writingFile.getPath()); } @@ -665,7 +665,7 @@ protected final TPipeTransferResp handleTransferFileSealV1(final PipeTransferFil final TSStatus status = loadFileV1(req, fileAbsolutePath); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { shouldDeleteSealedFile = false; - LOGGER.info(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePath); + LOGGER.debug(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePath); } else { PipeLogger.log( LOGGER::warn, @@ -773,7 +773,7 @@ protected final TPipeTransferResp handleTransferFileSealV2(final PipeTransferFil final TSStatus status = loadFileV2(req, fileAbsolutePaths); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.info(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePaths); + LOGGER.debug(PipeMessages.RECEIVER_SEAL_FILE_SUCCESS, receiverId.get(), fileAbsolutePaths); } else { PipeLogger.log( LOGGER::warn, diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java index c51f36d78f00..e718300978b0 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBReceiverAgent.java @@ -117,14 +117,14 @@ public static void cleanPipeReceiverDir(final File receiverFileDir) { FileUtils.deleteDirectory(receiverFileDir); return null; }); - LOGGER.info(PipeMessages.CLEAN_RECEIVER_DIR_SUCCESS, receiverFileDir); + LOGGER.debug(PipeMessages.CLEAN_RECEIVER_DIR_SUCCESS, receiverFileDir); } catch (final Exception e) { LOGGER.warn(PipeMessages.CLEAN_RECEIVER_DIR_FAILED, receiverFileDir, e); } try { FileUtils.forceMkdir(receiverFileDir); - LOGGER.info(PipeMessages.CREATE_RECEIVER_DIR_SUCCESS, receiverFileDir); + LOGGER.debug(PipeMessages.CREATE_RECEIVER_DIR_SUCCESS, receiverFileDir); } catch (final IOException e) { LOGGER.warn(PipeMessages.CREATE_RECEIVER_DIR_FAILED, receiverFileDir, e); }