This is an automated email from the ASF dual-hosted git repository.
justinchen 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 4945fbc22d0 Pipe: Further reduce the pipe logs & Added configurations
for tsFile segment lock (#16315) (#16327)
4945fbc22d0 is described below
commit 4945fbc22d0840dd6ed97ce127e9c6e8622cec92
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 2 15:27:40 2025 +0800
Pipe: Further reduce the pipe logs & Added configurations for tsFile
segment lock (#16315) (#16327)
* logger-fix
* add-ons
* fix
* fix
* fix
* warn-fix
* further-fix
* fix
---
.../confignode/persistence/pipe/PipeTaskInfo.java | 2 +-
.../impl/pipe/AbstractOperatePipeProcedureV2.java | 10 +-
.../impl/pipe/runtime/PipeMetaSyncProcedure.java | 18 ++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 17 ++--
.../tsfile/PipeTsFileResourceSegmentLock.java | 63 +++++--------
.../apache/iotdb/commons/conf/CommonConfig.java | 23 ++++-
.../iotdb/commons/pipe/config/PipeConfig.java | 7 ++
.../iotdb/commons/pipe/config/PipeDescriptor.java | 5 +
.../commons/pipe/receiver/IoTDBFileReceiver.java | 101 ++++++++++++++-------
9 files changed, 148 insertions(+), 98 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
index 716be67dde0..3df50fbf3b7 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java
@@ -644,7 +644,7 @@ public class PipeTaskInfo implements SnapshotProcessor {
}
private TSStatus handleMetaChangesInternal(final PipeHandleMetaChangePlan
plan) {
- LOGGER.info("Handling pipe meta changes ...");
+ LOGGER.debug("Handling pipe meta changes ...");
pipeMetaKeeper.clear();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
index 07059d6a0d9..b860d7fcb82 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java
@@ -104,12 +104,12 @@ public abstract class AbstractOperatePipeProcedureV2
@Override
protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv
configNodeProcedureEnv) {
- LOGGER.info("ProcedureId {} try to acquire pipe lock.", getProcId());
+ LOGGER.debug("ProcedureId {} try to acquire pipe lock.", getProcId());
pipeTaskInfo = acquireLockInternal(configNodeProcedureEnv);
if (pipeTaskInfo == null) {
LOGGER.warn("ProcedureId {} failed to acquire pipe lock.", getProcId());
} else {
- LOGGER.info("ProcedureId {} acquired pipe lock.", getProcId());
+ LOGGER.debug("ProcedureId {} acquired pipe lock.", getProcId());
}
final ProcedureLockState procedureLockState =
super.acquireLock(configNodeProcedureEnv);
@@ -120,7 +120,7 @@ public abstract class AbstractOperatePipeProcedureV2
"ProcedureId {}: LOCK_ACQUIRED. The following procedure should
not be executed without pipe lock.",
getProcId());
} else {
- LOGGER.info(
+ LOGGER.debug(
"ProcedureId {}: LOCK_ACQUIRED. The following procedure should
be executed with pipe lock.",
getProcId());
}
@@ -129,7 +129,7 @@ public abstract class AbstractOperatePipeProcedureV2
if (pipeTaskInfo == null) {
LOGGER.warn("ProcedureId {}: LOCK_EVENT_WAIT. Without acquiring pipe
lock.", getProcId());
} else {
- LOGGER.info("ProcedureId {}: LOCK_EVENT_WAIT. Pipe lock will be
released.", getProcId());
+ LOGGER.debug("ProcedureId {}: LOCK_EVENT_WAIT. Pipe lock will be
released.", getProcId());
configNodeProcedureEnv
.getConfigManager()
.getPipeManager()
@@ -168,7 +168,7 @@ public abstract class AbstractOperatePipeProcedureV2
if (pipeTaskInfo == null) {
LOGGER.warn("ProcedureId {} release lock. No need to release pipe
lock.", getProcId());
} else {
- LOGGER.info("ProcedureId {} release lock. Pipe lock will be released.",
getProcId());
+ LOGGER.debug("ProcedureId {} release lock. Pipe lock will be released.",
getProcId());
if (this instanceof PipeMetaSyncProcedure) {
configNodeProcedureEnv
.getConfigManager()
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
index c8a734a9e35..282b540b3aa 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java
@@ -75,7 +75,7 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
if (System.currentTimeMillis() - LAST_EXECUTION_TIME.get() <
MIN_EXECUTION_INTERVAL_MS) {
// Skip by setting the pipeTaskInfo to null
pipeTaskInfo = null;
- LOGGER.info(
+ LOGGER.debug(
"PipeMetaSyncProcedure: acquireLock, skip the procedure due to the
last execution time {}",
LAST_EXECUTION_TIME.get());
return ProcedureLockState.LOCK_ACQUIRED;
@@ -91,7 +91,7 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
@Override
public boolean executeFromValidateTask(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: executeFromValidateTask");
+ LOGGER.debug("PipeMetaSyncProcedure: executeFromValidateTask");
LAST_EXECUTION_TIME.set(System.currentTimeMillis());
return true;
@@ -99,14 +99,14 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
@Override
public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: executeFromCalculateInfoForTask");
+ LOGGER.debug("PipeMetaSyncProcedure: executeFromCalculateInfoForTask");
// Do nothing
}
@Override
public void executeFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: executeFromWriteConfigNodeConsensus");
+ LOGGER.debug("PipeMetaSyncProcedure: executeFromWriteConfigNodeConsensus");
final List<PipeMeta> pipeMetaList = new ArrayList<>();
for (final PipeMeta pipeMeta : pipeTaskInfo.get().getPipeMetaList()) {
@@ -132,7 +132,7 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
@Override
public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
throws PipeException, IOException {
- LOGGER.info("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
+ LOGGER.debug("PipeMetaSyncProcedure: executeFromOperateOnDataNodes");
Map<Integer, TPushPipeMetaResp> respMap = pushPipeMetaToDataNodes(env);
if (pipeTaskInfo.get().recordDataNodePushPipeMetaExceptions(respMap)) {
@@ -145,28 +145,28 @@ public class PipeMetaSyncProcedure extends
AbstractOperatePipeProcedureV2 {
@Override
public void rollbackFromValidateTask(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: rollbackFromValidateTask");
+ LOGGER.debug("PipeMetaSyncProcedure: rollbackFromValidateTask");
// Do nothing
}
@Override
public void rollbackFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: rollbackFromCalculateInfoForTask");
+ LOGGER.debug("PipeMetaSyncProcedure: rollbackFromCalculateInfoForTask");
// Do nothing
}
@Override
public void rollbackFromWriteConfigNodeConsensus(ConfigNodeProcedureEnv env)
{
- LOGGER.info("PipeMetaSyncProcedure: rollbackFromWriteConfigNodeConsensus");
+ LOGGER.debug("PipeMetaSyncProcedure:
rollbackFromWriteConfigNodeConsensus");
// Do nothing
}
@Override
public void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) {
- LOGGER.info("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");
+ LOGGER.debug("PipeMetaSyncProcedure: rollbackFromOperateOnDataNodes");
// Do nothing
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index 6cd709a3799..44ba9eaedfb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBPipePattern;
import org.apache.iotdb.commons.pipe.receiver.IoTDBFileReceiver;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import
org.apache.iotdb.commons.pipe.sink.payload.airgap.AirGapPseudoTPipeTransferRequest;
import
org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferSliceReqHandler;
import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
@@ -376,7 +377,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
} catch (final Exception e) {
final String error =
String.format("Exception %s encountered while handling request %s.",
e.getMessage(), req);
- LOGGER.warn("Receiver id = {}: {}", receiverId.get(), error, e);
+ PipeLogger.log(LOGGER::warn, e, "Receiver id = %s: %s",
receiverId.get(), error);
return new TPipeTransferResp(RpcUtils.getStatus(TSStatusCode.PIPE_ERROR,
error));
}
}
@@ -533,7 +534,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
final TSStatus status =
((AlterLogicalViewNode)
req.getPlanNode()).checkPermissionBeforeProcess(username);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to check authority for statement {},
username = {}, response = {}.",
receiverId.get(),
StatementType.ALTER_LOGICAL_VIEW.name(),
@@ -681,7 +683,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
|| result.getCode() ==
TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
return result;
} else {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failure status encountered while executing
statement {}: {}",
receiverId.get(),
statement,
@@ -689,7 +692,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
return statement.accept(STATEMENT_STATUS_VISITOR, result);
}
} catch (final Exception e) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Exception encountered while executing statement
{}: ",
receiverId.get(),
statement,
@@ -733,8 +737,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
final TSStatus status = AuthorityChecker.checkAuthority(statement,
clientSession);
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
- "Receiver id = {}: Failed to check authority for statement {},
username = {}, response = {}.",
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Failed to check authority for statement %s,
username = %s, response = %s.",
receiverId.get(),
statement.getType().name(),
username,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceSegmentLock.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceSegmentLock.java
index d1bf7fef2ad..cd1be83e55f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceSegmentLock.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/tsfile/PipeTsFileResourceSegmentLock.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.resource.tsfile;
+import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.slf4j.Logger;
@@ -31,25 +32,29 @@ import java.util.concurrent.locks.ReentrantLock;
public class PipeTsFileResourceSegmentLock {
private static final Logger LOGGER =
LoggerFactory.getLogger(PipeTsFileResourceSegmentLock.class);
-
private static final int SEGMENT_LOCK_MIN_SIZE = 32;
private static final int SEGMENT_LOCK_MAX_SIZE = 128;
-
private volatile ReentrantLock[] locks;
private void initIfNecessary() {
if (locks == null) {
synchronized (this) {
+ int lockSegmentSize =
PipeConfig.getInstance().getPipeTsFileResourceSegmentLockNum();
if (locks == null) {
- int lockSegmentSize = SEGMENT_LOCK_MIN_SIZE;
- try {
- lockSegmentSize =
StorageEngine.getInstance().getAllDataRegionIds().size();
- } catch (final Exception e) {
- LOGGER.warn(
- "Cannot get data region ids, use default lock segment size:
{}", lockSegmentSize);
+ if (lockSegmentSize <= 0) {
+ try {
+ lockSegmentSize =
+ Math.min(
+ Math.max(
+
StorageEngine.getInstance().getAllDataRegionIds().size(),
+ SEGMENT_LOCK_MIN_SIZE),
+ SEGMENT_LOCK_MAX_SIZE);
+ } catch (final Exception e) {
+ LOGGER.warn(
+ "Cannot get data region ids, use default lock segment size:
{}", lockSegmentSize);
+ lockSegmentSize = SEGMENT_LOCK_MIN_SIZE;
+ }
}
- lockSegmentSize = Math.min(SEGMENT_LOCK_MAX_SIZE, lockSegmentSize);
- lockSegmentSize = Math.max(SEGMENT_LOCK_MIN_SIZE, lockSegmentSize);
final ReentrantLock[] tmpLocks = new ReentrantLock[lockSegmentSize];
for (int i = 0; i < tmpLocks.length; i++) {
@@ -58,6 +63,13 @@ public class PipeTsFileResourceSegmentLock {
// publish this variable
locks = tmpLocks;
+ } else if (locks.length < lockSegmentSize) {
+ final ReentrantLock[] tmpLocks = new ReentrantLock[lockSegmentSize];
+ System.arraycopy(locks, 0, tmpLocks, 0, locks.length);
+ for (int i = locks.length; i < lockSegmentSize; ++i) {
+ tmpLocks[i] = new ReentrantLock();
+ }
+ locks = tmpLocks;
}
}
}
@@ -74,39 +86,8 @@ public class PipeTsFileResourceSegmentLock {
return locks[Math.abs(file.hashCode()) % locks.length].tryLock(timeout,
timeUnit);
}
- public boolean tryLockAll(final long timeout, final TimeUnit timeUnit)
- throws InterruptedException {
- initIfNecessary();
- int alreadyLocked = 0;
- for (final ReentrantLock lock : locks) {
- if (lock.tryLock(timeout, timeUnit)) {
- alreadyLocked++;
- } else {
- break;
- }
- }
-
- if (alreadyLocked == locks.length) {
- return true;
- } else {
- unlockUntil(alreadyLocked);
- return false;
- }
- }
-
- private void unlockUntil(final int index) {
- for (int i = 0; i < index; i++) {
- locks[i].unlock();
- }
- }
-
public void unlock(final File file) {
initIfNecessary();
locks[Math.abs(file.hashCode()) % locks.length].unlock();
}
-
- public void unlockAll() {
- initIfNecessary();
- unlockUntil(locks.length);
- }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 51cf41a9c4f..a8c4c1758a8 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -331,6 +331,7 @@ public class CommonConfig {
private volatile double
pipeThresholdAllocationStrategyFixedMemoryHighUsageThreshold = 0.8d;
private volatile boolean pipeTransferTsFileSync = false;
private volatile long pipeCheckAllSyncClientLiveTimeIntervalMs = 5 * 60 *
1000L; // 5 minutes
+ private int pipeTsFileResourceSegmentLockNum = -1;
private long twoStageAggregateMaxCombinerLiveTimeInMs = 8 * 60 * 1000L; // 8
minutes
private long twoStageAggregateDataRegionInfoCacheTimeInMs = 3 * 60 * 1000L;
// 3 minutes
@@ -1926,14 +1927,28 @@ public class CommonConfig {
}
public void setPipeCheckAllSyncClientLiveTimeIntervalMs(
- long pipeCheckSyncAllClientLiveTimeIntervalMs) {
- if (this.pipeCheckAllSyncClientLiveTimeIntervalMs ==
pipeCheckSyncAllClientLiveTimeIntervalMs) {
+ long pipeCheckAllSyncClientLiveTimeIntervalMs) {
+ if (this.pipeCheckAllSyncClientLiveTimeIntervalMs ==
pipeCheckAllSyncClientLiveTimeIntervalMs) {
return;
}
- this.pipeCheckAllSyncClientLiveTimeIntervalMs =
pipeCheckSyncAllClientLiveTimeIntervalMs;
+ this.pipeCheckAllSyncClientLiveTimeIntervalMs =
pipeCheckAllSyncClientLiveTimeIntervalMs;
logger.info(
"pipeCheckSyncAllClientLiveTimeIntervalMs is set to {}",
- pipeCheckSyncAllClientLiveTimeIntervalMs);
+ pipeCheckAllSyncClientLiveTimeIntervalMs);
+ }
+
+ public int getPipeTsFileResourceSegmentLockNum() {
+ return pipeTsFileResourceSegmentLockNum;
+ }
+
+ public void setPipeTsFileResourceSegmentLockNum(int
pipeTsFileResourceSegmentLockNum) {
+ if (this.pipeTsFileResourceSegmentLockNum ==
pipeTsFileResourceSegmentLockNum) {
+ return;
+ }
+ this.pipeTsFileResourceSegmentLockNum = pipeTsFileResourceSegmentLockNum;
+ logger.info(
+ "pipeCheckSyncAllClientLiveTimeIntervalMs is set to {}",
+ pipeCheckAllSyncClientLiveTimeIntervalMs);
}
public double getPipeSendTsFileRateLimitBytesPerSecond() {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
index e43bc059d1f..ec21b8fe72e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeConfig.java
@@ -293,6 +293,10 @@ public class PipeConfig {
return COMMON_CONFIG.getPipeCheckAllSyncClientLiveTimeIntervalMs();
}
+ public int getPipeTsFileResourceSegmentLockNum() {
+ return COMMON_CONFIG.getPipeTsFileResourceSegmentLockNum();
+ }
+
/////////////////////////////// Meta Consistency
///////////////////////////////
public boolean isSeperatedPipeHeartbeatEnabled() {
@@ -525,6 +529,9 @@ public class PipeConfig {
LOGGER.info(
"PipeCheckAllSyncClientLiveTimeIntervalMs: {}",
getPipeCheckAllSyncClientLiveTimeIntervalMs());
+ LOGGER.info(
+ "PipeCheckAllSyncClientLiveTimeIntervalMs: {}",
+ getPipeCheckAllSyncClientLiveTimeIntervalMs());
LOGGER.info("PipeDynamicMemoryHistoryWeight: {}",
getPipeDynamicMemoryHistoryWeight());
LOGGER.info(
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
index 8009fa27dea..1beb4c67e25 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/PipeDescriptor.java
@@ -498,6 +498,11 @@ public class PipeDescriptor {
properties.getProperty(
"pipe_check_all_sync_client_live_time_interval_ms",
String.valueOf(config.getPipeCheckAllSyncClientLiveTimeIntervalMs()))));
+ config.setPipeTsFileResourceSegmentLockNum(
+ Integer.parseInt(
+ properties.getProperty(
+ "pipe_tsfile_resource_segment_lock_num",
+
String.valueOf(config.getPipeTsFileResourceSegmentLockNum()))));
config.setPipeRemainingTimeCommitRateAutoSwitchSeconds(
Long.parseLong(
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 cd0914c5015..ca9b0c8dbc7 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
@@ -133,7 +133,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
receiverId.get(),
receiverFileDirWithIdSuffix.get().getPath());
} catch (Exception e) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to delete original receiver file dir
{}, because {}.",
receiverId.get(),
receiverFileDirWithIdSuffix.get().getPath(),
@@ -161,8 +162,9 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
try {
receiverFileBaseDir = getReceiverFileBaseDir();
if (Objects.isNull(receiverFileBaseDir)) {
- LOGGER.warn(
- "Receiver id = {}: Failed to init pipe receiver file folder
manager because all disks of folders are full.",
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Failed to init pipe receiver file folder
manager because all disks of folders are full.",
receiverId.get());
return new
TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT));
}
@@ -177,7 +179,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
// Create a new receiver file dir
final File newReceiverDir = new File(receiverFileBaseDir,
Long.toString(receiverId.get()));
if (!newReceiverDir.exists() && !newReceiverDir.mkdirs()) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to create receiver file dir {}.",
receiverId.get(),
newReceiverDir.getPath());
@@ -212,8 +215,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
RpcUtils.getStatus(
TSStatusCode.PIPE_HANDSHAKE_ERROR,
"Receiver can not get clusterId from config node.");
- LOGGER.warn(
- "Receiver id = {}: Handshake failed, response status = {}.",
receiverId.get(), status);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = {}: Handshake failed, response status = {}.",
+ receiverId.get(),
+ status);
return new TPipeTransferResp(status);
}
@@ -224,8 +230,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
final TSStatus status =
RpcUtils.getStatus(
TSStatusCode.PIPE_HANDSHAKE_ERROR, "Handshake request does not
contain clusterId.");
- LOGGER.warn(
- "Receiver id = {}: Handshake failed, response status = {}.",
receiverId.get(), status);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Handshake failed, response status = %s.",
+ receiverId.get(),
+ status);
return new TPipeTransferResp(status);
}
@@ -237,8 +246,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
String.format(
"Receiver and sender are from the same cluster %s.",
clusterIdFromHandshakeRequest));
- LOGGER.warn(
- "Receiver id = {}: Handshake failed, response status = {}.",
receiverId.get(), status);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Handshake failed, response status = %s.",
+ receiverId.get(),
+ status);
return new TPipeTransferResp(status);
}
@@ -250,8 +262,11 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
RpcUtils.getStatus(
TSStatusCode.PIPE_HANDSHAKE_ERROR,
"Handshake request does not contain timestampPrecision.");
- LOGGER.warn(
- "Receiver id = {}: Handshake failed, response status = {}.",
receiverId.get(), status);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: Handshake failed, response status = %s.",
+ receiverId.get(),
+ status);
return new TPipeTransferResp(status);
}
@@ -267,7 +282,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
}
final TSStatus status = loginIfNecessary();
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Handshake failed because login failed, response
status = {}.",
receiverId.get(),
status);
@@ -328,7 +344,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
if (shouldLogin()) {
final TSStatus permissionCheckStatus = login();
if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to login, username = {}, response = {}.",
receiverId.get(),
username,
@@ -371,8 +388,9 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
String.format(
"Request sender to reset file reader's offset from %s to
%s.",
req.getStartWritingOffset(), writingFileWriter.length()));
- LOGGER.warn(
- "Receiver id = {}: File offset reset requested by receiver,
response status = {}.",
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = %s: File offset reset requested by receiver,
response status = %s.",
receiverId.get(),
status);
return PipeTransferFilePieceResp.toTPipeTransferResp(status,
writingFileWriter.length());
@@ -382,8 +400,12 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
return PipeTransferFilePieceResp.toTPipeTransferResp(
RpcUtils.SUCCESS_STATUS, writingFileWriter.length());
} catch (final Exception e) {
- LOGGER.warn(
- "Receiver id = {}: Failed to write file piece from req {}.",
receiverId.get(), req, e);
+ PipeLogger.log(
+ LOGGER::warn,
+ e,
+ "Receiver id = %s: Failed to write file piece from req {}.",
+ receiverId.get(),
+ req);
final TSStatus status =
RpcUtils.getStatus(
TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
@@ -457,7 +479,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
receiverId.get(),
writingFile == null ? "null" : writingFile.getPath());
} catch (final Exception e) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to close current writing file writer {},
because {}.",
receiverId.get(),
writingFile == null ? "null" : writingFile.getPath(),
@@ -495,7 +518,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
receiverId.get(),
file.getPath());
} catch (final Exception e) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to delete original writing file {},
because {}.",
receiverId.get(),
file.getPath(),
@@ -515,7 +539,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
private boolean isWritingFileOffsetCorrect(final long offset) throws
IOException {
final boolean offsetCorrect = writingFileWriter.length() == offset;
if (!offsetCorrect) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Writing file {}'s offset is {}, but request
sender's offset is {}.",
receiverId.get(),
writingFile.getPath(),
@@ -533,7 +558,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
String.format(
"Failed to seal file, because writing file %s is not
available.", writingFile));
- LOGGER.warn(status.getMessage());
+ PipeLogger.log(LOGGER::warn, status.getMessage());
return new TPipeTransferResp(status);
}
@@ -569,7 +594,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
LOGGER.info(
"Receiver id = {}: Seal file {} successfully.", receiverId.get(),
fileAbsolutePath);
} else {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, because {}.",
receiverId.get(),
fileAbsolutePath,
@@ -577,7 +603,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
}
return new TPipeTransferResp(status);
} catch (final Exception e) {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {} from req {}.",
receiverId.get(),
writingFile,
@@ -609,7 +636,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
String.format(
"Failed to seal file %s, because writing file %s is not
available.",
req.getFileNames(), writingFile));
- LOGGER.warn(status.getMessage());
+ PipeLogger.log(LOGGER::warn, status.getMessage());
return new TPipeTransferResp(status);
}
@@ -654,7 +681,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
LOGGER.info(
"Receiver id = {}: Seal file {} successfully.", receiverId.get(),
fileAbsolutePaths);
} else {
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, status is {}.",
receiverId.get(),
fileAbsolutePaths,
@@ -662,8 +690,13 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
}
return new TPipeTransferResp(status);
} catch (final Exception e) {
- LOGGER.warn(
- "Receiver id = {}: Failed to seal file {} from req {}.",
receiverId.get(), files, req, e);
+ PipeLogger.log(
+ LOGGER::warn,
+ "Receiver id = {}: Failed to seal file {} from req {}.",
+ receiverId.get(),
+ files,
+ req,
+ e);
return new TPipeTransferResp(
RpcUtils.getStatus(
TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
@@ -686,7 +719,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
RpcUtils.getStatus(
TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
String.format("Failed to seal file %s, the file does not
exist.", fileName));
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, because the file does not
exist.",
receiverId.get(),
fileName);
@@ -701,7 +735,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
"Failed to seal file %s, because the length of file is not
correct. "
+ "The original file has length %s, but receiver file
has length %s.",
fileName, fileLength, writingFileWriter.length()));
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, because the length of
file is not correct. "
+ "The original file has length {}, but receiver file has length
{}.",
receiverId.get(),
@@ -722,7 +757,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
TSStatusCode.PIPE_TRANSFER_FILE_ERROR,
String.format(
"Failed to seal file %s, because writing file is %s.",
fileName, writingFile));
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, because writing file is
{}.",
receiverId.get(),
fileName,
@@ -738,7 +774,8 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
"Failed to seal file %s, because the length of file is not
correct. "
+ "The original file has length %s, but receiver file
has length %s.",
fileName, fileLength, writingFileWriter.length()));
- LOGGER.warn(
+ PipeLogger.log(
+ LOGGER::warn,
"Receiver id = {}: Failed to seal file {}, because the length of
file is not correct. "
+ "The original file has length {}, but receiver file has length
{}.",
receiverId.get(),
