This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 81f59fa39f2 Reduce pipe load success log noise (#18106) (#18183)
81f59fa39f2 is described below
commit 81f59fa39f2bbddf67603adb5e13ce295180ad89
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 13 10:47:14 2026 +0800
Reduce pipe load success log noise (#18106) (#18183)
* Reduce pipe load success log noise
* Fix table cluster IT failures
---
.../plan/analyze/load/LoadTsFileAnalyzer.java | 38 ++++++++----
.../scheduler/load/LoadTsFileDispatcherImpl.java | 68 ++++++++++++++--------
.../plan/scheduler/load/LoadTsFileScheduler.java | 27 ++++++---
.../db/storageengine/dataregion/DataRegion.java | 36 +++++++++---
.../commons/pipe/receiver/IoTDBFileReceiver.java | 10 ++--
.../commons/pipe/receiver/IoTDBReceiverAgent.java | 4 +-
6 files changed, 125 insertions(+), 58 deletions(-)
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 d47953abe9b..56511e8acb2 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
@@ -229,7 +229,11 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
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 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
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 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
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
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 37f2fba832d..b734412ca9c 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
@@ -69,7 +69,7 @@ import java.util.concurrent.atomic.AtomicInteger;
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,11 +92,17 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher {
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;
}
@@ -104,24 +110,26 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher {
@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 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher {
}
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 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher {
@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 e61367ed896..8381e44f753 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
@@ -242,11 +242,19 @@ public class LoadTsFileScheduler implements IScheduler {
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 class LoadTsFileScheduler implements IScheduler {
}
}
} finally {
+ dispatcher.close();
LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock();
}
}
@@ -387,7 +396,11 @@ public class LoadTsFileScheduler implements IScheduler {
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 @@ public class LoadTsFileScheduler implements IScheduler {
@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 d06362ffd37..5568aa4015c 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
@@ -3244,10 +3244,17 @@ public class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
}
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 @@ public class DataRegion implements IDataRegionForQuery {
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());
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 3223305dc80..b1e29c8563d 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
@@ -523,7 +523,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
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 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
// 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 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
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 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
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 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
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(
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 7aaa66a8b75..ebf0898c019 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
@@ -120,14 +120,14 @@ public abstract class IoTDBReceiverAgent {
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);
}
