Skip to content

Commit 4388ad3

Browse files
CaideyipiCaideyipi
committed
Reduce pipe load success log noise (#18106)
* Reduce pipe load success log noise * Fix table cluster IT failures
1 parent 427d402 commit 4388ad3

6 files changed

Lines changed: 125 additions & 58 deletions

File tree

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

Lines changed: 27 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -229,7 +229,11 @@ public Analysis analyzeFileByFile(Analysis analysis) {
229229
return analysis;
230230
}
231231

232-
LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed.");
232+
if (isGeneratedByPipe) {
233+
LOGGER.debug("Load - Analysis Stage: all tsfiles have been analyzed.");
234+
} else {
235+
LOGGER.info("Load - Analysis Stage: all tsfiles have been analyzed.");
236+
}
233237

234238
if (reconstructStatementIfMiniFileConverted()) {
235239
// All mini tsfiles are converted to tablets, so the analysis is finished.
@@ -287,22 +291,14 @@ private boolean doAnalyzeFileByFile(Analysis analysis) {
287291
if (LOGGER.isWarnEnabled()) {
288292
LOGGER.warn("TsFile {} is empty.", tsFile.getPath());
289293
}
290-
if (LOGGER.isInfoEnabled()) {
291-
LOGGER.info(
292-
"Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
293-
i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
294-
}
294+
logAnalyzeProgress(i + 1, tsfileNum);
295295
continue;
296296
}
297297

298298
final long startTime = System.nanoTime();
299299
try {
300300
analyzeSingleTsFile(tsFile, i);
301-
if (LOGGER.isInfoEnabled()) {
302-
LOGGER.info(
303-
"Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
304-
i + 1, tsfileNum, String.format("%.3f", (i + 1) * 100.00 / tsfileNum));
305-
}
301+
logAnalyzeProgress(i + 1, tsfileNum);
306302
} catch (AuthException e) {
307303
setFailAnalysisForAuthException(analysis, e);
308304
return false;
@@ -339,6 +335,26 @@ private boolean doAnalyzeFileByFile(Analysis analysis) {
339335
return true;
340336
}
341337

338+
private void logAnalyzeProgress(final int analyzedTsFileNum, final int totalTsFileNum) {
339+
if (isGeneratedByPipe && !LOGGER.isDebugEnabled()) {
340+
return;
341+
}
342+
if (!isGeneratedByPipe && !LOGGER.isInfoEnabled()) {
343+
return;
344+
}
345+
346+
final String progress = String.format("%.3f", analyzedTsFileNum * 100.00 / totalTsFileNum);
347+
if (isGeneratedByPipe) {
348+
LOGGER.debug(
349+
"Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
350+
analyzedTsFileNum, totalTsFileNum, progress);
351+
} else {
352+
LOGGER.info(
353+
"Load - Analysis Stage: {}/{} tsfiles have been analyzed, progress: {}%",
354+
analyzedTsFileNum, totalTsFileNum, progress);
355+
}
356+
}
357+
342358
private void analyzeSingleTsFile(final File tsFile, int index) throws Exception {
343359
try (final TsFileSequenceReader reader = new TsFileSequenceReader(tsFile.getAbsolutePath())) {
344360
// can be reused when constructing tsfile resource

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

Lines changed: 44 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@
6969

7070
import static com.google.common.util.concurrent.Futures.immediateFuture;
7171

72-
public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
72+
public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher, AutoCloseable {
7373

7474
private static final Logger LOGGER = LoggerFactory.getLogger(LoadTsFileDispatcherImpl.class);
7575

@@ -83,7 +83,7 @@ public class LoadTsFileDispatcherImpl implements IFragInstanceDispatcher {
8383
private final int localhostInternalPort;
8484
private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
8585
internalServiceClientManager;
86-
private final ExecutorService executor;
86+
private ExecutorService executor;
8787
private final boolean isGeneratedByPipe;
8888

8989
public LoadTsFileDispatcherImpl(
@@ -92,36 +92,44 @@ public LoadTsFileDispatcherImpl(
9292
this.internalServiceClientManager = internalServiceClientManager;
9393
this.localhostIpAddr = IoTDBDescriptor.getInstance().getConfig().getInternalAddress();
9494
this.localhostInternalPort = IoTDBDescriptor.getInstance().getConfig().getInternalPort();
95-
this.executor =
96-
IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName());
9795
this.isGeneratedByPipe = isGeneratedByPipe;
9896
}
9997

98+
private synchronized ExecutorService getOrCreateExecutor() {
99+
if (executor == null || executor.isShutdown()) {
100+
executor =
101+
IoTDBThreadPoolFactory.newCachedThreadPool(LoadTsFileDispatcherImpl.class.getName());
102+
}
103+
return executor;
104+
}
105+
100106
public void setUuid(String uuid) {
101107
this.uuid = uuid;
102108
}
103109

104110
@Override
105111
public Future<FragInstanceDispatchResult> dispatch(
106112
SubPlan root, List<FragmentInstance> instances) {
107-
return executor.submit(
108-
() -> {
109-
for (FragmentInstance instance : instances) {
110-
try (SetThreadName threadName =
111-
new SetThreadName(
112-
"load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) {
113-
dispatchOneInstance(instance);
114-
} catch (FragmentInstanceDispatchException e) {
115-
return new FragInstanceDispatchResult(e.getFailureStatus());
116-
} catch (Exception t) {
117-
LOGGER.warn("cannot dispatch FI for load operation", t);
118-
return new FragInstanceDispatchResult(
119-
RpcUtils.getStatus(
120-
TSStatusCode.INTERNAL_SERVER_ERROR, "Unexpected errors: " + t.getMessage()));
121-
}
122-
}
123-
return new FragInstanceDispatchResult(true);
124-
});
113+
return getOrCreateExecutor()
114+
.submit(
115+
() -> {
116+
for (FragmentInstance instance : instances) {
117+
try (SetThreadName threadName =
118+
new SetThreadName(
119+
"load-dispatcher" + "-" + instance.getId().getFullId() + "-" + uuid)) {
120+
dispatchOneInstance(instance);
121+
} catch (FragmentInstanceDispatchException e) {
122+
return new FragInstanceDispatchResult(e.getFailureStatus());
123+
} catch (Exception t) {
124+
LOGGER.warn("cannot dispatch FI for load operation", t);
125+
return new FragInstanceDispatchResult(
126+
RpcUtils.getStatus(
127+
TSStatusCode.INTERNAL_SERVER_ERROR,
128+
"Unexpected errors: " + t.getMessage()));
129+
}
130+
}
131+
return new FragInstanceDispatchResult(true);
132+
});
125133
}
126134

127135
private void dispatchOneInstance(FragmentInstance instance)
@@ -147,7 +155,11 @@ private void dispatchOneInstance(FragmentInstance instance)
147155
}
148156

149157
public void dispatchLocally(FragmentInstance instance) throws FragmentInstanceDispatchException {
150-
LOGGER.info("Receive load node from uuid {}.", uuid);
158+
if (isGeneratedByPipe) {
159+
LOGGER.debug("Receive load node from uuid {}.", uuid);
160+
} else {
161+
LOGGER.info("Receive load node from uuid {}.", uuid);
162+
}
151163

152164
ConsensusGroupId groupId =
153165
ConsensusGroupId.Factory.createFromTConsensusGroupId(
@@ -352,6 +364,14 @@ private static void adjustTimeoutIfNecessary(Throwable e) {
352364

353365
@Override
354366
public void abort() {
355-
// Do nothing
367+
close();
368+
}
369+
370+
@Override
371+
public synchronized void close() {
372+
if (executor != null) {
373+
executor.shutdownNow();
374+
executor = null;
375+
}
356376
}
357377
}

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

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -242,11 +242,19 @@ public void start() {
242242

243243
if (isLoadSingleTsFileSuccess) {
244244
node.clean();
245-
LOGGER.info(
246-
"Load TsFile {} Successfully, load process [{}/{}]",
247-
filePath,
248-
i + 1,
249-
tsFileNodeListSize);
245+
if (isGeneratedByPipe) {
246+
LOGGER.debug(
247+
"Load TsFile {} Successfully, load process [{}/{}]",
248+
filePath,
249+
i + 1,
250+
tsFileNodeListSize);
251+
} else {
252+
LOGGER.info(
253+
"Load TsFile {} Successfully, load process [{}/{}]",
254+
filePath,
255+
i + 1,
256+
tsFileNodeListSize);
257+
}
250258
} else {
251259
isLoadSuccess = false;
252260
failedTsFileNodeIndexes.add(i);
@@ -299,6 +307,7 @@ public void start() {
299307
}
300308
}
301309
} finally {
310+
dispatcher.close();
302311
LoadTsFileMemoryManager.getInstance().releaseDataCacheMemoryBlock();
303312
}
304313
}
@@ -387,7 +396,11 @@ private boolean dispatchOnePieceNode(
387396

388397
private boolean secondPhase(
389398
boolean isFirstPhaseSuccess, String uuid, TsFileResource tsFileResource) {
390-
LOGGER.info("Start dispatching Load command for uuid {}", uuid);
399+
if (isGeneratedByPipe) {
400+
LOGGER.debug("Start dispatching Load command for uuid {}", uuid);
401+
} else {
402+
LOGGER.info("Start dispatching Load command for uuid {}", uuid);
403+
}
391404
final File tsFile = tsFileResource.getTsFile();
392405
final TLoadCommandReq loadCommandReq =
393406
new TLoadCommandReq(
@@ -594,7 +607,7 @@ private void convertFailedTsFilesToTabletsAndRetry() {
594607

595608
@Override
596609
public void stop(Throwable t) {
597-
// Do nothing
610+
dispatcher.abort();
598611
}
599612

600613
@Override

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

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3244,10 +3244,17 @@ public void loadNewTsFile(
32443244
0);
32453245

32463246
if (!newFileName.equals(tsfileToBeInserted.getName())) {
3247-
logger.info(
3248-
"TsFile {} must be renamed to {} for loading into the unsequence list.",
3249-
tsfileToBeInserted.getName(),
3250-
newFileName);
3247+
if (isGeneratedByPipe) {
3248+
logger.debug(
3249+
"TsFile {} must be renamed to {} for loading into the unsequence list.",
3250+
tsfileToBeInserted.getName(),
3251+
newFileName);
3252+
} else {
3253+
logger.info(
3254+
"TsFile {} must be renamed to {} for loading into the unsequence list.",
3255+
tsfileToBeInserted.getName(),
3256+
newFileName);
3257+
}
32513258
newTsFileResource.setFile(
32523259
fsFactory.getFile(tsfileToBeInserted.getParentFile(), newFileName));
32533260
}
@@ -3282,7 +3289,11 @@ public void loadNewTsFile(
32823289
}
32833290

32843291
onTsFileLoaded(newTsFileResource, isFromConsensus, lastReader);
3285-
logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
3292+
if (isGeneratedByPipe) {
3293+
logger.debug("TsFile {} is successfully loaded in unsequence list.", newFileName);
3294+
} else {
3295+
logger.info("TsFile {} is successfully loaded in unsequence list.", newFileName);
3296+
}
32863297
} catch (final DiskSpaceInsufficientException e) {
32873298
logger.error(
32883299
"Failed to append the tsfile {} to database processor {} because the disk space is insufficient.",
@@ -3451,10 +3462,17 @@ private boolean loadTsFileToUnSequence(
34513462
return false;
34523463
}
34533464

3454-
logger.info(
3455-
"Load tsfile in unsequence list, move file from {} to {}",
3456-
tsFileToLoad.getAbsolutePath(),
3457-
targetFile.getAbsolutePath());
3465+
if (isGeneratedByPipe) {
3466+
logger.debug(
3467+
"Load tsfile in unsequence list, move file from {} to {}",
3468+
tsFileToLoad.getAbsolutePath(),
3469+
targetFile.getAbsolutePath());
3470+
} else {
3471+
logger.info(
3472+
"Load tsfile in unsequence list, move file from {} to {}",
3473+
tsFileToLoad.getAbsolutePath(),
3474+
targetFile.getAbsolutePath());
3475+
}
34583476

34593477
LoadTsFileRateLimiter.getInstance().acquire(tsFileResource.getTsFile().length());
34603478

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -523,7 +523,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
523523
return;
524524
}
525525

526-
LOGGER.info(
526+
LOGGER.debug(
527527
"Receiver id = {}: Writing file {} is not existed or name is not correct, try to create it. "
528528
+ "Current writing file is {}.",
529529
receiverId.get(),
@@ -541,7 +541,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
541541
// This may be useless, because receiver file dir is created when handshake. just in case.
542542
if (!receiverFileDirWithIdSuffix.get().exists()) {
543543
if (receiverFileDirWithIdSuffix.get().mkdirs()) {
544-
LOGGER.info(
544+
LOGGER.debug(
545545
"Receiver id = {}: Receiver file dir {} was created.",
546546
receiverId.get(),
547547
receiverFileDirWithIdSuffix.get().getPath());
@@ -556,7 +556,7 @@ protected final void updateWritingFileIfNeeded(final String fileName, final bool
556556

557557
writingFile = targetPath.toFile();
558558
writingFileWriter = new RandomAccessFile(writingFile, "rw");
559-
LOGGER.info(
559+
LOGGER.debug(
560560
"Receiver id = {}: Writing file {} was created. Ready to write file pieces.",
561561
receiverId.get(),
562562
writingFile.getPath());
@@ -730,7 +730,7 @@ protected final TPipeTransferResp handleTransferFileSealV1(final PipeTransferFil
730730
final TSStatus status = loadFileV1(req, fileAbsolutePath);
731731
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
732732
shouldDeleteSealedFile = false;
733-
LOGGER.info(
733+
LOGGER.debug(
734734
"Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePath);
735735
} else {
736736
PipeLogger.log(
@@ -835,7 +835,7 @@ protected final TPipeTransferResp handleTransferFileSealV2(final PipeTransferFil
835835

836836
final TSStatus status = loadFileV2(req, fileAbsolutePaths);
837837
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
838-
LOGGER.info(
838+
LOGGER.debug(
839839
"Receiver id = {}: Seal file {} successfully.", receiverId.get(), fileAbsolutePaths);
840840
} else {
841841
PipeLogger.log(

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -120,14 +120,14 @@ public static void cleanPipeReceiverDir(final File receiverFileDir) {
120120
FileUtils.deleteDirectory(receiverFileDir);
121121
return null;
122122
});
123-
LOGGER.info("Clean pipe receiver dir {} successfully.", receiverFileDir);
123+
LOGGER.debug("Clean pipe receiver dir {} successfully.", receiverFileDir);
124124
} catch (final Exception e) {
125125
LOGGER.warn("Clean pipe receiver dir {} failed.", receiverFileDir, e);
126126
}
127127

128128
try {
129129
FileUtils.forceMkdir(receiverFileDir);
130-
LOGGER.info("Create pipe receiver dir {} successfully.", receiverFileDir);
130+
LOGGER.debug("Create pipe receiver dir {} successfully.", receiverFileDir);
131131
} catch (final IOException e) {
132132
LOGGER.warn("Create pipe receiver dir {} failed.", receiverFileDir, e);
133133
}

0 commit comments

Comments
 (0)