This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5aad9323c71 Reduce pipe type conversion log frequency (#18161)
5aad9323c71 is described below
commit 5aad9323c71ef5fdfd25b924dba010f37fb94ed9
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 9 17:49:59 2026 +0800
Reduce pipe type conversion log frequency (#18161)
---
.../iotdb/confignode/i18n/ManagerMessages.java | 17 +++++++
.../iotdb/confignode/i18n/ManagerMessages.java | 17 +++++++
.../iotdb/confignode/manager/ClusterManager.java | 3 +-
.../iotdb/confignode/manager/node/NodeManager.java | 2 +-
.../manager/partition/PartitionManager.java | 26 ++++++----
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 8 +--
.../manager/schema/ClusterSchemaManager.java | 4 +-
.../procedure/impl/cq/CreateCQProcedure.java | 2 +-
.../impl/schema/DeactivateTemplateProcedure.java | 2 +-
.../impl/schema/DeleteLogicalViewProcedure.java | 2 +-
.../impl/schema/DeleteTimeSeriesProcedure.java | 2 +-
.../impl/schema/SetTemplateProcedure.java | 2 +-
.../impl/sync/AuthOperationProcedure.java | 2 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 57 ++++++++++++++++++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 52 ++++++++++++++++++++
.../subtask/processor/PipeProcessorSubtask.java | 4 +-
.../common/tsfile/PipeTsFileInsertionEvent.java | 7 ++-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 23 +++++----
.../statement/PipeConvertedInsertRowStatement.java | 30 ++++++------
.../PipeConvertedInsertTabletStatement.java | 14 +++---
...leStatementDataTypeConvertExecutionVisitor.java | 16 ++++--
...eeStatementDataTypeConvertExecutionVisitor.java | 9 +++-
.../client/IoTDBDataNodeAsyncClientManager.java | 11 ++---
.../iotconsensusv2/IoTConsensusV2AsyncSink.java | 16 +++---
.../handler/IoTConsensusV2DeleteEventHandler.java | 2 +-
.../IoTConsensusV2TabletBatchEventHandler.java | 2 +-
.../IoTConsensusV2TabletInsertionEventHandler.java | 2 +-
.../IoTConsensusV2TsFileInsertionEventHandler.java | 2 +-
.../db/pipe/sink/protocol/opcua/OpcUaSink.java | 3 +-
.../protocol/opcua/client/IoTDBOpcUaClient.java | 2 +-
.../sink/protocol/opcua/server/OpcUaNameSpace.java | 2 +-
.../thrift/async/IoTDBDataRegionAsyncSink.java | 6 +--
.../PipeTransferTabletBatchEventHandler.java | 2 +-
.../PipeTransferTabletInsertionEventHandler.java | 2 +-
.../handler/PipeTransferTrackableHandler.java | 4 +-
.../async/handler/PipeTransferTsFileHandler.java | 8 +--
.../sink/protocol/writeback/WriteBackSink.java | 2 +-
37 files changed, 273 insertions(+), 94 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 082fcd9439b..20316edc397 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -49,6 +49,17 @@ public final class ManagerMessages {
"[CreateRegionGroups] Starting to create the following RegionGroups:";
public static final String CREATE_DATAPARTITION_FAILED_BECAUSE =
"Create DataPartition failed because: ";
+ public static final String
+ DATAPARTITIONTABLEINTEGRITYCHECKPROCEDURE_IS_ALREADY_SUBMITTED =
+ "DataPartitionTableIntegrityCheckProcedure is already submitted.";
+ public static final String
+
LACKED_DATAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_DATA_PARTITIONS_FOR_DATABASES
=
+ "Lacked %d/%d DataPartition allocation result when get or create
data partitions for databases: %s";
+ public static final String
NO_RUNNING_DATAPARTITIONTABLE_INTEGRITY_CHECK_PROCEDURE =
+ "No running DataPartitionTable integrity check procedure";
+ public static final String
+
LACKED_SCHEMAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_SCHEMA_PARTITIONS_FOR_DATABASES
=
+ "Lacked %d/%d SchemaPartition allocation result when get or create
schema partitions for databases: %s";
public static final String CREATE_SCHEMAPARTITION_FAILED_BECAUSE =
"Create SchemaPartition failed because: ";
public static final String DATABASE_DOESN_T_EXIST = "Database: {} doesn't
exist";
@@ -277,6 +288,8 @@ public final class ManagerMessages {
"LoadStatistics service is stopped successfully.";
public static final String
MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE
=
"[MigrateRegion] Submit RegionMigrateProcedure successfully, Region: {},
Origin DataNode: {}, Dest DataNode: {}, Add Coordinator: {}, Remove
Coordinator: {}";
+ public static final String
SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =
+ "Submit RegionMigrateProcedure failed, because RegionGroup: %s doesn't
exist";
public static final String
MISMATCHED_CRC32_CODE_WHEN_DESERIALIZING_SERVICE_INFO =
"Mismatched CRC32 code when deserializing service info.";
public static final String
NETWORK_ERROR_WHEN_SEAL_CONFIG_REGION_SNAPSHOT_BECAUSE =
@@ -357,6 +370,10 @@ public final class ManagerMessages {
"Receiver id = {}: Failure status encountered while executing plan {}:
{}";
public static final String
RECEIVER_ID_PERMISSION_CHECK_FAILED_WHILE_EXECUTING_PLAN =
"Receiver id = {}: Permission check failed while executing plan {}: {}";
+ public static final String UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE =
+ "Unsupported PipeRequestType on ConfigNode %s.";
+ public static final String
EXCEPTION_ENCOUNTERED_WHILE_HANDLING_PIPE_TRANSFER_REQUEST =
+ "Exception encountered while handling pipe transfer request. Root cause:
%s";
public static final String
RECEIVER_ID_UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE_RESPONSE_STATUS =
"Receiver id = {}: Unsupported PipeRequestType on ConfigNode, response
status = {}.";
public static final String
RECONSTRUCTREGION_SUBMIT_RECONSTRUCTREGIONPROCEDURE_SUCCESSFULLY =
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 072c2b728a2..8dc9afd1e8b 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -49,6 +49,17 @@ public final class ManagerMessages {
"[CreateRegionGroups] 开始创建以下 RegionGroup：";
public static final String CREATE_DATAPARTITION_FAILED_BECAUSE =
"创建 DataPartition 失败，原因：";
+ public static final String
+ DATAPARTITIONTABLEINTEGRITYCHECKPROCEDURE_IS_ALREADY_SUBMITTED =
+ "DataPartitionTableIntegrityCheckProcedure 已提交。";
+ public static final String
+
LACKED_DATAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_DATA_PARTITIONS_FOR_DATABASES
=
+ "获取或创建数据库的数据分区时缺少 %d/%d 个 DataPartition 分配结果，数据库：%s";
+ public static final String
NO_RUNNING_DATAPARTITIONTABLE_INTEGRITY_CHECK_PROCEDURE =
+ "没有正在运行的 DataPartitionTable 完整性检查流程";
+ public static final String
+
LACKED_SCHEMAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_SCHEMA_PARTITIONS_FOR_DATABASES
=
+ "获取或创建数据库的模式分区时缺少 %d/%d 个 SchemaPartition 分配结果，数据库：%s";
public static final String CREATE_SCHEMAPARTITION_FAILED_BECAUSE =
"创建 SchemaPartition 失败，原因：";
public static final String DATABASE_DOESN_T_EXIST = "Database: {} 不存在";
@@ -275,6 +286,8 @@ public final class ManagerMessages {
"LoadStatistics 服务已成功停止。";
public static final String
MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE
=
"[MigrateRegion] 成功提交 RegionMigrateProcedure，Region：{}，原 DataNode：{}，目标
DataNode：{}，新增 Coordinator：{}，移除 Coordinator：{}";
+ public static final String
SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =
+ "提交 RegionMigrateProcedure 失败，因为 RegionGroup：%s 不存在";
public static final String
MISMATCHED_CRC32_CODE_WHEN_DESERIALIZING_SERVICE_INFO =
"反序列化 service info 时 CRC32 码不匹配。";
public static final String
NETWORK_ERROR_WHEN_SEAL_CONFIG_REGION_SNAPSHOT_BECAUSE =
@@ -352,6 +365,10 @@ public final class ManagerMessages {
"Receiver id = {}: 执行计划 {} 时遇到失败状态：{}";
public static final String
RECEIVER_ID_PERMISSION_CHECK_FAILED_WHILE_EXECUTING_PLAN =
"Receiver id = {}: 执行计划 {} 时权限检查失败：{}";
+ public static final String UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE =
+ "ConfigNode 上不支持的 PipeRequestType %s。";
+ public static final String
EXCEPTION_ENCOUNTERED_WHILE_HANDLING_PIPE_TRANSFER_REQUEST =
+ "处理 pipe transfer 请求时遇到异常。根因：%s";
public static final String
RECEIVER_ID_UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE_RESPONSE_STATUS =
"Receiver id = {}: ConfigNode 上不支持的 PipeRequestType，响应状态 = {}。";
public static final String
RECONSTRUCTREGION_SUBMIT_RECONSTRUCTREGIONPROCEDURE_SUCCESSFULLY =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
index 7faec232cab..790f09c08b9 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterManager.java
@@ -40,6 +40,7 @@ import
org.apache.iotdb.confignode.client.async.handlers.ConfigNodeAsyncRequestC
import
org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import
org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateClusterIdPlan;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.persistence.ClusterInfo;
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -64,7 +65,7 @@ public class ClusterManager {
private final ClusterInfo clusterInfo;
public static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public ClusterManager(IManager configManager, ClusterInfo clusterInfo) {
this.configManager = configManager;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 6db027af546..13e476c0037 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -140,7 +140,7 @@ public class NodeManager {
private final ReentrantLock removeConfigNodeLock;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public static final int APPLY_CONFIG_LOCALLY = -2;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index 1cf416bedd7..156493f80e3 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
@@ -72,6 +72,7 @@ import
org.apache.iotdb.confignode.consensus.response.partition.SchemaPartitionR
import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.exception.NoAvailableRegionGroupException;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.manager.IManager;
import org.apache.iotdb.confignode.manager.ProcedureManager;
@@ -140,10 +141,10 @@ public class PartitionManager {
private SeriesPartitionExecutor executor;
private static final String CONSENSUS_READ_ERROR =
- "Failed in the read API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_READ_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
// Monitor for leadership change
private final Object scheduleMonitor = new Object();
@@ -347,8 +348,11 @@ public class PartitionManager {
final String errMsg =
String.format(
- "Lacked %d/%d SchemaPartition allocation result when get or
create schema partitions for databases: %s",
- unassignedSlotNum.get(), totalSlotNum.get(), errDatabases);
+ ManagerMessages
+
.LACKED_SCHEMAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_SCHEMA_PARTITIONS_FOR_DATABASES,
+ unassignedSlotNum.get(),
+ totalSlotNum.get(),
+ errDatabases);
LOGGER.error(errMsg);
resp.setStatus(
new
TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
@@ -511,8 +515,11 @@ public class PartitionManager {
String errMsg =
String.format(
- "Lacked %d/%d DataPartition allocation result when get or create
data partitions for databases: %s",
- unassignedSlotNum.get(), totalSlotNum.get(), errDatabases);
+ ManagerMessages
+
.LACKED_DATAPARTITION_ALLOCATION_RESULT_WHEN_GET_OR_CREATE_DATA_PARTITIONS_FOR_DATABASES,
+ unassignedSlotNum.get(),
+ totalSlotNum.get(),
+ errDatabases);
LOGGER.error(errMsg);
resp.setStatus(
new
TSStatus(TSStatusCode.LACK_PARTITION_ALLOCATION.getStatusCode()).setMessage(errMsg));
@@ -529,7 +536,7 @@ public class PartitionManager {
||
!dataPartitionTableIntegrityCheckProcedureRunning.compareAndSet(false, true)) {
return RpcUtils.getStatus(
TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
- "DataPartitionTableIntegrityCheckProcedure is already submitted.");
+
ManagerMessages.DATAPARTITIONTABLEINTEGRITYCHECKPROCEDURE_IS_ALREADY_SUBMITTED);
}
synchronized (this) {
@@ -555,7 +562,8 @@ public class PartitionManager {
RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS),
RepairDataPartitionTableProgressState.IDLE.name(),
0.0)
- .setMessage("No running DataPartitionTable integrity check
procedure"));
+ .setMessage(
+
ManagerMessages.NO_RUNNING_DATAPARTITIONTABLE_INTEGRITY_CHECK_PROCEDURE));
}
private TSStatus consensusWritePartitionResult(ConfigPhysicalPlan plan) {
@@ -1077,7 +1085,7 @@ public class PartitionManager {
}
String msg =
String.format(
- "Submit RegionMigrateProcedure failed, because RegionGroup: %s
doesn't exist",
+
ManagerMessages.SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST,
regionId);
LOGGER.warn(msg);
return Optional.empty();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
index f68f1f71ed8..281867d0ec5 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
@@ -258,7 +258,8 @@ public class IoTDBConfigNodeReceiver extends
IoTDBFileReceiver {
final TSStatus status =
RpcUtils.getStatus(
TSStatusCode.PIPE_TYPE_ERROR,
- String.format("Unsupported PipeRequestType on ConfigNode %s.",
rawRequestType));
+ String.format(
+ ManagerMessages.UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE,
rawRequestType));
LOGGER.warn(
ManagerMessages.RECEIVER_ID_UNSUPPORTED_PIPEREQUESTTYPE_ON_CONFIGNODE_RESPONSE_STATUS,
receiverId.get(),
@@ -266,8 +267,9 @@ public class IoTDBConfigNodeReceiver extends
IoTDBFileReceiver {
return new TPipeTransferResp(status);
} catch (final Exception e) {
final String error =
- "Exception encountered while handling pipe transfer request. Root
cause: "
- + e.getMessage();
+ String.format(
+
ManagerMessages.EXCEPTION_ENCOUNTERED_WHILE_HANDLING_PIPE_TRANSFER_REQUEST,
+ e.getMessage());
LOGGER.warn(ManagerMessages.RECEIVER_ID, receiverId.get(), error, e);
return new TPipeTransferResp(RpcUtils.getStatus(TSStatusCode.PIPE_ERROR,
error));
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
index da8467ee900..a66b88f0a10 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -150,10 +150,10 @@ public class ClusterSchemaManager {
private final ReentrantLock createDatabaseLock = new ReentrantLock();
private static final String CONSENSUS_READ_ERROR =
- "Failed in the read API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_READ_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public ClusterSchemaManager(
final IManager configManager,
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java
index 680b960e8e6..7066815ef14 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/cq/CreateCQProcedure.java
@@ -70,7 +70,7 @@ public class CreateCQProcedure extends
AbstractNodeProcedure<CreateCQState> {
private long firstExecutionTime;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public CreateCQProcedure(ScheduledExecutorService executor) {
super();
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
index 9fa3c49d6d0..4ca00eb46aa 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java
@@ -82,7 +82,7 @@ public class DeactivateTemplateProcedure
private Map<String, List<Integer>> dataNodeRequest; // transient
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public DeactivateTemplateProcedure(boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java
index 343f658601d..d97b0270e58 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteLogicalViewProcedure.java
@@ -71,7 +71,7 @@ public class DeleteLogicalViewProcedure
private transient String requestMessage;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public DeleteLogicalViewProcedure(final boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
index a417ed55d98..d875636cc2d 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteTimeSeriesProcedure.java
@@ -80,7 +80,7 @@ public class DeleteTimeSeriesProcedure
private transient boolean isAllLogicalView;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public DeleteTimeSeriesProcedure(final boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
index 04ab1d34963..5a16b49c1f1 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java
@@ -83,7 +83,7 @@ public class SetTemplateProcedure
private String templateSetPath;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
public SetTemplateProcedure(final boolean isGeneratedByPipe) {
super(isGeneratedByPipe);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
index 59e4f50b40b..7ae95858ac6 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/sync/AuthOperationProcedure.java
@@ -66,7 +66,7 @@ public class AuthOperationProcedure extends
AbstractNodeProcedure<AuthOperationP
private long timeoutMS;
private static final String CONSENSUS_WRITE_ERROR =
- "Failed in the write API executing the consensus layer due to: ";
+
ProcedureMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE;
private static final int RETRY_THRESHOLD = 2;
private static final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index bcdab68abfd..9963d029094 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -108,6 +108,9 @@ public final class DataNodePipeMessages {
"Error occurred when collecting events from processor.";
public static final String
EXCEPTION_IN_PIPE_EVENT_PROCESSING_IGNORED_BECAUSE =
"Exception in pipe event processing, ignored because pipe is dropped.{}";
+ public static final String
TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING =
+ "Temporarily out of memory in pipe event processing, will wait for the
memory to release. "
+ + "Message: {}";
public static final String
EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_CONNECTOR_SUBTASK =
"Exception occurred when closing pipe connector subtask {}, root cause:
{}";
public static final String
EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_PROCESSOR_SUBTASK =
@@ -515,6 +518,9 @@ public final class DataNodePipeMessages {
"Interrupted waiting for processor to stop";
public static final String
INTERRUPTED_WHEN_WAITING_FOR_PARSING_PRIVILEGE_FOR_TSFILE =
"Interrupted when waiting for parsing privilege for TsFile %s.";
+ public static final String INTERRUPTED_WHEN_WAITING_FOR_CLOSING_TSFILE =
+ "Interrupted when waiting for closing TsFile %s.";
+ public static final String PARSE_TSFILE_ERROR_BECAUSE = "Parse TsFile %s
error. Because: %s";
public static final String PARSE_TSFILE_WHEN_CHECKING_PRIVILEGE_ERROR =
"Parse TsFile %s when checking privilege error. Because: %s";
public static final String READ_TSFILE_ERROR = "Read TsFile %s error.";
@@ -710,6 +716,13 @@ public final class DataNodePipeMessages {
"Failed to adjust timeout when failed to transfer file.";
public static final String FAILED_TO_BORROW_CLIENT_FOR_CACHED_LEADER =
"failed to borrow client {}:{} for cached leader.";
+ public static final String HANDSHAKE_ERROR_WITH_RECEIVER =
+ "Handshake error with receiver {}:{}, code: {}, message: {}.";
+ public static final String HANDSHAKE_ERROR_WITH_RECEIVER_1 =
+ "Handshake error with receiver {}:{}.";
+ public static final String HANDSHAKE_ERROR_BY_HANDSHAKE_V2_RETRY_WITH_V1 =
+ "Handshake error by PipeTransferHandshakeV2Req with receiver {}:{} retry
to handshake by "
+ + "PipeTransferHandshakeV1Req.";
public static final String FAILED_TO_BUILD_AND_STARTUP_OPCUASERVER =
"Failed to build and startup OpcUaServer";
public static final String
FAILED_TO_CLOSE_ASYNCPIPEDATATRANSFERSERVICECLIENTMANAGER_FOR_RECEIVER_ATTRIBUTE
=
@@ -756,12 +769,23 @@ public final class DataNodePipeMessages {
"Failed to transfer dataValue after successfully created nodes";
public static final String
FAILED_TO_TRANSFER_PIPEDELETENODEEVENT_COMMITTER_KEY_REPLICATE =
"Failed to transfer PipeDeleteNodeEvent {} (committer key={}, replicate
index={}).";
+ public static final String FAILED_TO_TRANSFER_SLICE_RETRY_WHOLE_TRANSFER =
+ "Failed to transfer slice. Origin req: {}-{}. Retry the whole transfer.";
public static final String
FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_REPLICATE =
"Failed to transfer TabletInsertionEvent {} (committer key={}, replicate
index={}).";
+ public static final String
FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
+ "Failed to transfer TabletInsertionEvent {} (committer key={}, commit
id={}).";
+ public static final String FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_BATCH =
+ "Failed to transfer TabletInsertionEvent batch. Total failed events: {},
related pipe "
+ + "names: {}";
public static final String FAILED_TO_TRANSFER_TSFILE_BATCH =
"Failed to transfer tsfile batch ({}).";
public static final String FAILED_TO_TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY =
"Failed to transfer tsfile event {} asynchronously.";
+ public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
+ "Failed to transfer TsFileInsertionEvent {} (committer key {}, commit id
{}).";
+ public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_BATCHED_TABLE_EVENTS =
+ "Failed to transfer TsFileInsertionEvent {} (batched
TableInsertionEvents).";
public static final String FAILED_TO_UPDATE_LEADER_CACHE_FOR_DEVICE =
"Failed to update leader cache for device {} with endpoint {}:{}.";
public static final String FAILED_TO_WRITE = "Failed to write ";
@@ -995,6 +1019,17 @@ public final class DataNodePipeMessages {
public static final String TIOTCONSENSUSV2TRANSFERRESP_IS_NULL =
"TIoTConsensusV2TransferResp is null";
public static final String TPIPETRANSFERRESP_IS_NULL = "TPipeTransferResp is
null";
+ public static final String
OPC_UA_SINK_MODEL_MUST_BE_CLIENT_SERVER_WHEN_OUTER_OR_WITH_QUALITY =
+ "When the OPC UA sink points to an outer server or sets 'with-quality'
to true, the %s or "
+ + "%s must be %s.";
+ public static final String
WITH_QUALITY_MEASUREMENT_MUST_BE_VALUE_OR_QUALITY_NAME =
+ "When the 'with-quality' mode is enabled, the measurement must be either
\"value-name\" or "
+ + "\"quality-name\"";
+ public static final String SESSION_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT =
+ "Session {}: Failed to check authority for statement {}, username = {},
response = {}.";
+ public static final String TRANSFER_REQUEST_BODY_TOO_LARGE_WILL_BE_SLICED =
+ "The body size of the request is too large. The request will be sliced.
Origin req: {}-{}. "
+ + "Request body size: {}, threshold: {}";
public static final String
TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY_WAS_INTERRUPTED =
"Transfer tsfile event {} asynchronously was interrupted.";
public static final String UNABLE_TO_CREATE_SECURITY_DIR = "unable to create
security dir: ";
@@ -1036,6 +1071,8 @@ public final class DataNodePipeMessages {
"Database name is unexpectedly null for LoadTsFileStatement: {}. Skip
data type conversion.";
public static final String DATABASE_NAME_IS_UNEXPECTEDLY_NULL_FOR_STATEMENT =
"Database name is unexpectedly null for statement: {}. Skip data type
conversion.";
+ public static final String
DATABASE_NAME_IS_UNEXPECTEDLY_NULL_SKIP_DATA_TYPE_CONVERSION =
+ "Pipe: Database name is unexpectedly null. Skip data type conversion.";
public static final String
DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL =
"Data type conversion for LoadTsFileStatement {} is successful.";
public static final String
DATA_TYPE_MISMATCH_DETECTED_TSSTATUS_FOR_LOADTSFILESTATEMENT =
@@ -1049,6 +1086,11 @@ public final class DataNodePipeMessages {
"Failed to convert data type for LoadTsFileStatement: {}.";
public static final String FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE =
"Failed to execute statement after data type conversion.";
+ public static final String
+
FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE_CONVERSION_WITH_EXCEPTION_TYPE =
+ "Pipe: Failed to execute statement after data type conversion.
Exception type: {}.";
+ public static final String
FAILED_TO_PARSE_ROW_VALUE_DURING_DATA_TYPE_CONVERSION =
+ "Pipe: Failed to parse row value during data type conversion. Registered
type {}.";
public static final String FAILED_TO_HANDLE_CONFIG_CLIENT_ID_EXIT =
"Failed to handle config client (id = {}) exit";
public static final String
FAIL_TO_CREATE_IOTCONSENSUSV2_RECEIVER_FILE_FOLDERS =
@@ -1235,6 +1277,10 @@ public final class DataNodePipeMessages {
public static final String PIPE_AIR_GAP_RECEIVER_TSSTATUS_IS_ENCOUNTERED =
"Pipe air gap receiver {}: TSStatus {} is encountered at the air gap
receiver, will ignore.";
public static final String PIPE_DATA_TRANSPORT_ERROR = "Pipe data transport
error, {}";
+ public static final String PIPE_INSERTING_ROW_CASTING_TYPE_FROM =
+ "Pipe: Inserting row. Casting type from {} to {}.";
+ public static final String PIPE_INSERTING_TABLET_CASTING_TYPE_FROM =
+ "Pipe: Inserting tablet. Casting type from {} to {}.";
public static final String PIPE_INSERTING_TABLET_TO_CASTING_TYPE_FROM =
"Pipe: Inserting tablet to {}.{}. Casting type from {} to {}.";
public static final String RECEIVERS_EXECUTOR_IS_CLOSED = "Receivers-{}'
executor is closed.";
@@ -1247,6 +1293,17 @@ public final class DataNodePipeMessages {
"Receiver id = {}: Unknown PipeRequestType, response status = {}.";
public static final String
RECEIVER_ID_UNSUPPORTED_STATEMENT_TYPE_FOR_REDIRECTION =
"Receiver id = {}: Unsupported statement type {} for redirection.";
+ public static final String
RECEIVER_ID_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT =
+ "Receiver id = {}: Failed to check authority for statement {}, username
= {}, response = {}.";
+ public static final String
RECEIVER_ID_FAILURE_STATUS_WHILE_EXECUTING_STATEMENT =
+ "Receiver id = {}: Failure status encountered while executing statement
{}: {}";
+ public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
+ "Receiver id = {}: Exception encountered while executing statement {}: ";
+ public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
+ "Receiver id = {}, statement = {}, exception = {}, message = {}";
+ public static final String UNKNOWN_PIPEREQUESTTYPE = "Unknown
PipeRequestType %s.";
+ public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
+ "Exception %s encountered while handling request %s.";
public static final String RECEIVER_IS_READY = "Receiver-{} is ready";
public static final String RECEIVER_TEMPORARILY_OUT_OF_MEMORY_FORMAT =
"Temporarily out of memory when %s. Requested memory: %d bytes, used
memory: %d bytes, "
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 3a51486d847..b34fb373ddb 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -100,6 +100,8 @@ public final class DataNodePipeMessages {
"从 processor 收集事件时发生错误";
public static final String
EXCEPTION_IN_PIPE_EVENT_PROCESSING_IGNORED_BECAUSE =
"pipe event processing 中发生异常，因为 pipe 已被删除，忽略该异常。{}";
+ public static final String
TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING =
+ "Pipe 事件处理时暂时内存不足，将等待内存释放。消息：{}";
public static final String
EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_CONNECTOR_SUBTASK =
"关闭 pipe connector 子任务 {} 时发生异常，根因：{}";
public static final String
EXCEPTION_OCCURRED_WHEN_CLOSING_PIPE_PROCESSOR_SUBTASK =
@@ -492,6 +494,9 @@ public final class DataNodePipeMessages {
"等待 processor 停止时被中断";
public static final String
INTERRUPTED_WHEN_WAITING_FOR_PARSING_PRIVILEGE_FOR_TSFILE =
"等待解析 TsFile %s 的权限信息时被中断。";
+ public static final String INTERRUPTED_WHEN_WAITING_FOR_CLOSING_TSFILE =
+ "等待 TsFile %s 关闭时被中断。";
+ public static final String PARSE_TSFILE_ERROR_BECAUSE = "解析 TsFile %s
失败。原因：%s";
public static final String PARSE_TSFILE_WHEN_CHECKING_PRIVILEGE_ERROR =
"检查权限时解析 TsFile %s 失败。原因：%s";
public static final String READ_TSFILE_ERROR = "读取 TsFile %s 失败。";
@@ -675,6 +680,13 @@ public final class DataNodePipeMessages {
"传输文件失败后调整超时时间失败。";
public static final String FAILED_TO_BORROW_CLIENT_FOR_CACHED_LEADER =
"为 cached leader 借用 client {}:{} 失败。";
+ public static final String HANDSHAKE_ERROR_WITH_RECEIVER =
+ "与接收端 {}:{} 握手失败，状态码：{}，消息：{}。";
+ public static final String HANDSHAKE_ERROR_WITH_RECEIVER_1 =
+ "与接收端 {}:{} 握手失败。";
+ public static final String HANDSHAKE_ERROR_BY_HANDSHAKE_V2_RETRY_WITH_V1 =
+ "使用 PipeTransferHandshakeV2Req 与接收端 {}:{} 握手失败，改用
PipeTransferHandshakeV1Req "
+ + "重试握手。";
public static final String FAILED_TO_BUILD_AND_STARTUP_OPCUASERVER =
"构建并启动 OpcUaServer 失败";
public static final String
FAILED_TO_CLOSE_ASYNCPIPEDATATRANSFERSERVICECLIENTMANAGER_FOR_RECEIVER_ATTRIBUTE
=
@@ -718,11 +730,21 @@ public final class DataNodePipeMessages {
"成功创建 node 后传输 dataValue 失败";
public static final String
FAILED_TO_TRANSFER_PIPEDELETENODEEVENT_COMMITTER_KEY_REPLICATE =
"传输 PipeDeleteNodeEvent {} (committer key={}, replicate index={}) 失败。";
+ public static final String FAILED_TO_TRANSFER_SLICE_RETRY_WHOLE_TRANSFER =
+ "传输 slice 失败。原始请求：{}-{}。将重试完整传输。";
public static final String
FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_REPLICATE =
"传输 TabletInsertionEvent {} (committer key={}, replicate index={}) 失败。";
+ public static final String
FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
+ "传输 TabletInsertionEvent {}（committer key={}，commit id={}）失败。";
+ public static final String FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_BATCH =
+ "传输 TabletInsertionEvent 批次失败。失败事件总数：{}，相关 pipe 名称：{}";
public static final String FAILED_TO_TRANSFER_TSFILE_BATCH = "传输 tsfile
batch ({}) 失败。";
public static final String FAILED_TO_TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY =
"传输 tsfile event {} asynchronously 失败。";
+ public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID =
+ "传输 TsFileInsertionEvent {}（committer key {}，commit id {}）失败。";
+ public static final String
FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_BATCHED_TABLE_EVENTS =
+ "传输 TsFileInsertionEvent {}（批量 TableInsertionEvent）失败。";
public static final String FAILED_TO_UPDATE_LEADER_CACHE_FOR_DEVICE =
"更新 leader cache for device {} with endpoint {}:{} 失败。";
public static final String FAILED_TO_WRITE = "写入失败 ";
@@ -937,6 +959,14 @@ public final class DataNodePipeMessages {
"TIoTConsensusV2BatchTransferResp 为空";
public static final String TIOTCONSENSUSV2TRANSFERRESP_IS_NULL =
"TIoTConsensusV2TransferResp 为空";
public static final String TPIPETRANSFERRESP_IS_NULL = "TPipeTransferResp
为空";
+ public static final String
OPC_UA_SINK_MODEL_MUST_BE_CLIENT_SERVER_WHEN_OUTER_OR_WITH_QUALITY =
+ "当 OPC UA sink 指向外部 server 或将 'with-quality' 设置为 true 时，%s 或 %s 必须为 %s。";
+ public static final String
WITH_QUALITY_MEASUREMENT_MUST_BE_VALUE_OR_QUALITY_NAME =
+ "启用 'with-quality' 模式时，measurement 必须是 \"value-name\" 或
\"quality-name\"。";
+ public static final String SESSION_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT =
+ "Session {}: 检查 statement {} 权限失败，username = {}，response = {}。";
+ public static final String TRANSFER_REQUEST_BODY_TOO_LARGE_WILL_BE_SLICED =
+ "请求体过大，将切分请求。原始请求：{}-{}。请求体大小：{}，阈值：{}";
public static final String
TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY_WAS_INTERRUPTED =
"异步传输 tsfile event {} 被中断。";
public static final String UNABLE_TO_CREATE_SECURITY_DIR = "无法创建 security
dir: ";
@@ -974,6 +1004,8 @@ public final class DataNodePipeMessages {
"LoadTsFileStatement：{} 的数据库名称为空，跳过数据类型转换。";
public static final String DATABASE_NAME_IS_UNEXPECTEDLY_NULL_FOR_STATEMENT =
"statement：{} 的数据库名称为空，跳过数据类型转换。";
+ public static final String
DATABASE_NAME_IS_UNEXPECTEDLY_NULL_SKIP_DATA_TYPE_CONVERSION =
+ "Pipe：数据库名称为空，跳过数据类型转换。";
public static final String
DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL =
"LoadTsFileStatement {} 的数据类型转换成功。";
public static final String
DATA_TYPE_MISMATCH_DETECTED_TSSTATUS_FOR_LOADTSFILESTATEMENT =
@@ -986,6 +1018,11 @@ public final class DataNodePipeMessages {
"转换 data type for LoadTsFileStatement: {} 失败。";
public static final String FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE =
"execute statement after data type conversion 失败。";
+ public static final String
+
FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE_CONVERSION_WITH_EXCEPTION_TYPE =
+ "Pipe：执行数据类型转换后的 statement 失败。异常类型：{}。";
+ public static final String
FAILED_TO_PARSE_ROW_VALUE_DURING_DATA_TYPE_CONVERSION =
+ "Pipe：数据类型转换时解析 row value 失败。注册类型：{}。";
public static final String FAILED_TO_HANDLE_CONFIG_CLIENT_ID_EXIT =
"处理 config client (id = {}) exit 失败";
public static final String
FAIL_TO_CREATE_IOTCONSENSUSV2_RECEIVER_FILE_FOLDERS =
@@ -1164,6 +1201,10 @@ public final class DataNodePipeMessages {
public static final String PIPE_AIR_GAP_RECEIVER_TSSTATUS_IS_ENCOUNTERED =
"Pipe air gap receiver {}：在 air gap receiver 处遇到 TSStatus {}，将忽略。";
public static final String PIPE_DATA_TRANSPORT_ERROR = "Pipe 数据传输错误，{}";
+ public static final String PIPE_INSERTING_ROW_CASTING_TYPE_FROM =
+ "Pipe：写入 row。将类型从 {} 转换为 {}。";
+ public static final String PIPE_INSERTING_TABLET_CASTING_TYPE_FROM =
+ "Pipe：写入 tablet。将类型从 {} 转换为 {}。";
public static final String PIPE_INSERTING_TABLET_TO_CASTING_TYPE_FROM =
"Pipe：向 {}.{} 写入 tablet。将类型从 {} 转换为 {}。";
public static final String RECEIVERS_EXECUTOR_IS_CLOSED = "Receivers-{} 的
executor 已关闭。";
@@ -1175,6 +1216,17 @@ public final class DataNodePipeMessages {
"Receiver id = {}：未知的 PipeRequestType，response status = {}。";
public static final String
RECEIVER_ID_UNSUPPORTED_STATEMENT_TYPE_FOR_REDIRECTION =
"Receiver id = {}：不支持的 statement type {} 用于 redirection。";
+ public static final String
RECEIVER_ID_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT =
+ "Receiver id = {}: 检查 statement {} 权限失败，username = {}，response = {}。";
+ public static final String
RECEIVER_ID_FAILURE_STATUS_WHILE_EXECUTING_STATEMENT =
+ "Receiver id = {}: 执行 statement {} 时遇到失败状态：{}";
+ public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
+ "Receiver id = {}: 执行 statement {} 时遇到异常：";
+ public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
+ "Receiver id = {}，statement = {}，exception = {}，message = {}";
+ public static final String UNKNOWN_PIPEREQUESTTYPE = "未知 PipeRequestType
%s。";
+ public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
+ "遇到异常 %s，处理请求 %s 时。";
public static final String RECEIVER_IS_READY = "Receiver-{} 已就绪";
public static final String RECEIVER_TEMPORARILY_OUT_OF_MEMORY_FORMAT =
"执行 %s 时暂时内存不足。请求内存：%d bytes，已用内存：%d bytes，可用内存：%d bytes，"
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index 1003c8a0ccb..8c26726bfb2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -249,14 +249,14 @@ public class PipeProcessorSubtask extends
PipeReportableSubtask {
} catch (final PipeRuntimeOutOfMemoryCriticalException e) {
PipeLogger.log(
LOGGER::info,
- "Temporarily out of memory in pipe event processing, will wait for
the memory to release. Message: %s",
+
DataNodePipeMessages.TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING,
e.getMessage());
return false;
} catch (final Exception e) {
if (ExceptionUtils.getRootCause(e) instanceof
PipeRuntimeOutOfMemoryCriticalException) {
PipeLogger.log(
LOGGER::info,
- "Temporarily out of memory in pipe event processing, will wait for
the memory to release. Message: %s",
+
DataNodePipeMessages.TEMPORARILY_OUT_OF_MEMORY_IN_PIPE_EVENT_PROCESSING,
e.getMessage());
return false;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
index 77ea67fdd97..757411df9b9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/PipeTsFileInsertionEvent.java
@@ -855,9 +855,12 @@ public class PipeTsFileInsertionEvent extends
PipeInsertionEvent
final String errorMsg =
e instanceof InterruptedException
? String.format(
- "Interrupted when waiting for closing TsFile %s.",
resource.getTsFilePath())
+
DataNodePipeMessages.INTERRUPTED_WHEN_WAITING_FOR_CLOSING_TSFILE,
+ resource.getTsFilePath())
: String.format(
- "Parse TsFile %s error. Because: %s",
resource.getTsFilePath(), e.getMessage());
+ DataNodePipeMessages.PARSE_TSFILE_ERROR_BECAUSE,
+ resource.getTsFilePath(),
+ e.getMessage());
if (e instanceof PipeRuntimeOutOfMemoryCriticalException) {
PipeLogger.log(LOGGER::warn, errorMsg);
} else {
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 46c00a095c8..3048ad13084 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
@@ -464,7 +464,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
final TSStatus status =
RpcUtils.getStatus(
TSStatusCode.PIPE_TYPE_ERROR,
- String.format("Unknown PipeRequestType %s.", rawRequestType));
+ String.format(DataNodePipeMessages.UNKNOWN_PIPEREQUESTTYPE,
rawRequestType));
LOGGER.warn(
DataNodePipeMessages.RECEIVER_ID_UNKNOWN_PIPEREQUESTTYPE_RESPONSE_STATUS,
receiverId.get(),
@@ -472,8 +472,11 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
return new TPipeTransferResp(status);
} catch (final Exception e) {
final String error =
- String.format("Exception %s encountered while handling request %s.",
e.getMessage(), req);
- PipeLogger.log(LOGGER::warn, e, "Receiver id = %s: %s",
receiverId.get(), error);
+ String.format(
+
DataNodePipeMessages.EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST,
+ e.getMessage(),
+ req);
+ PipeLogger.log(LOGGER::warn, e, DataNodePipeMessages.RECEIVER_ID,
receiverId.get(), error);
return new TPipeTransferResp(RpcUtils.getStatus(TSStatusCode.PIPE_ERROR,
error));
}
}
@@ -828,7 +831,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Receiver id = %s: Failed to check authority for statement %s,
username = %s, response = %s.",
+
DataNodePipeMessages.RECEIVER_ID_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT,
receiverId.get(),
StatementType.ALTER_LOGICAL_VIEW.name(),
username,
@@ -1057,7 +1060,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
if (code != TSStatusCode.OUT_OF_TTL.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Receiver id = %s: Failure status encountered while executing
statement %s: %s",
+
DataNodePipeMessages.RECEIVER_ID_FAILURE_STATUS_WHILE_EXECUTING_STATEMENT,
receiverId.get(),
statement.getPipeLoggingString(),
result);
@@ -1080,7 +1083,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
PipeLogger.log(
LOGGER::warn,
e,
- "Receiver id = %s: Exception encountered while executing statement
%s: ",
+ DataNodePipeMessages.RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT,
receiverId.get(),
Objects.isNull(statement) ? null : statement.getPipeLoggingString());
}
@@ -1091,7 +1094,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
// Use the reducer cache as a gate. The actual stack trace is logged only
when it passes.
return PipePeriodicalLogReducer.log(
message -> {},
- "Receiver id = %s, statement = %s, exception = %s, message = %s",
+ DataNodePipeMessages.RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE,
receiverId,
Objects.isNull(statement) ? null : statement.getPipeLoggingString(),
e.getClass().getName(),
@@ -1149,7 +1152,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Receiver id = %s: Failed to check authority for statement %s,
username = %s, response = %s.",
+
DataNodePipeMessages.RECEIVER_ID_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT,
receiverId.get(),
statement.getType().name(),
username,
@@ -1500,7 +1503,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
|| result.getCode() ==
TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode())) {
PipeLogger.log(
LOGGER::warn,
- "Receiver id = %s: Failure status encountered while executing
statement %s: %s",
+
DataNodePipeMessages.RECEIVER_ID_FAILURE_STATUS_WHILE_EXECUTING_STATEMENT,
receiverId.get(),
statement,
result);
@@ -1510,7 +1513,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
PipeLogger.log(
LOGGER::warn,
e,
- "Receiver id = %s: Exception encountered while executing statement
%s: ",
+ DataNodePipeMessages.RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT,
receiverId.get(),
statement);
return new
TSStatus(TSStatusCode.PIPE_TRANSFER_EXECUTE_STATEMENT_ERROR.getStatusCode())
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
index 5f70750864b..99edef690b5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.PathNotExistException;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.receiver.transform.converter.ValueConverter;
import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
@@ -92,13 +93,13 @@ public class PipeConvertedInsertRowStatement extends
InsertRowStatement {
@Override
protected boolean checkAndCastDataType(int columnIndex, TSDataType dataType)
{
- PipeLogger.log(
- LOGGER::info,
- "Pipe: Inserting row to %s.%s. Casting type from %s to %s.",
- devicePath,
- measurements[columnIndex],
- dataTypes[columnIndex],
- dataType);
+ if (LOGGER.isInfoEnabled()) {
+ PipeLogger.log(
+ LOGGER::info,
+ DataNodePipeMessages.PIPE_INSERTING_ROW_CASTING_TYPE_FROM,
+ dataTypes[columnIndex],
+ dataType);
+ }
values[columnIndex] =
ValueConverter.convert(dataTypes[columnIndex], dataType,
values[columnIndex]);
dataTypes[columnIndex] = dataType;
@@ -138,15 +139,12 @@ public class PipeConvertedInsertRowStatement extends
InsertRowStatement {
try {
values[i] = ValueConverter.parse(values[i].toString(), dataTypes[i]);
} catch (Exception e) {
- PipeLogger.log(
- LOGGER::warn,
- "data type of %s.%s is not consistent, "
- + "registered type %s, inserting timestamp %s, value %s",
- devicePath,
- measurements[i],
- dataTypes[i],
- time,
- values[i]);
+ if (LOGGER.isWarnEnabled()) {
+ PipeLogger.log(
+ LOGGER::warn,
+
DataNodePipeMessages.FAILED_TO_PARSE_ROW_VALUE_DURING_DATA_TYPE_CONVERSION,
+ dataTypes[i]);
+ }
if
(!IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert()) {
throw e;
} else {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
index 24f8a793c1c..f7e6f0be1e6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertTabletStatement.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.pipe.receiver.transform.statement;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.receiver.transform.converter.ArrayConverter;
import
org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
@@ -98,12 +99,13 @@ public class PipeConvertedInsertTabletStatement extends
InsertTabletStatement {
@Override
protected boolean checkAndCastDataType(int columnIndex, TSDataType dataType)
{
- LOGGER.info(
- DataNodePipeMessages.PIPE_INSERTING_TABLET_TO_CASTING_TYPE_FROM,
- devicePath,
- measurements[columnIndex],
- dataTypes[columnIndex],
- dataType);
+ if (LOGGER.isInfoEnabled()) {
+ PipeLogger.log(
+ LOGGER::info,
+ DataNodePipeMessages.PIPE_INSERTING_TABLET_CASTING_TYPE_FROM,
+ dataTypes[columnIndex],
+ dataType);
+ }
columns[columnIndex] =
ArrayConverter.convert(dataTypes[columnIndex], dataType,
columns[columnIndex]);
dataTypes[columnIndex] = dataType;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTableStatementDataTypeConvertExecutionVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTableStatementDataTypeConvertExecutionVisitor.java
index 908b0055afe..5a80da6f6e6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTableStatementDataTypeConvertExecutionVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTableStatementDataTypeConvertExecutionVisitor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.receiver.visitor;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import
org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.table.TsFileInsertionEventTableParser;
@@ -77,14 +78,23 @@ public class
PipeTableStatementDataTypeConvertExecutionVisitor
private Optional<TSStatus> tryExecute(final Statement statement, final
String databaseName) {
try {
if (Objects.isNull(databaseName)) {
- LOGGER.warn(
-
DataNodePipeMessages.DATABASE_NAME_IS_UNEXPECTEDLY_NULL_FOR_STATEMENT,
statement);
+ if (LOGGER.isWarnEnabled()) {
+ PipeLogger.log(
+ LOGGER::warn,
+
DataNodePipeMessages.DATABASE_NAME_IS_UNEXPECTEDLY_NULL_SKIP_DATA_TYPE_CONVERSION);
+ }
return Optional.empty();
}
return Optional.of(statementExecutor.execute(statement, databaseName));
} catch (final Exception e) {
-
LOGGER.warn(DataNodePipeMessages.FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE,
e);
+ if (LOGGER.isWarnEnabled()) {
+ PipeLogger.log(
+ LOGGER::warn,
+ DataNodePipeMessages
+
.FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE_CONVERSION_WITH_EXCEPTION_TYPE,
+ e.getClass().getName());
+ }
return Optional.empty();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTreeStatementDataTypeConvertExecutionVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTreeStatementDataTypeConvertExecutionVisitor.java
index f8c6eadc5f5..489ec2ede22 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTreeStatementDataTypeConvertExecutionVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipeTreeStatementDataTypeConvertExecutionVisitor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.pipe.receiver.visitor;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.pipe.config.PipeConfig;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import
org.apache.iotdb.db.pipe.event.common.tsfile.parser.scan.TsFileInsertionEventScanParser;
import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver;
@@ -75,7 +76,13 @@ public class PipeTreeStatementDataTypeConvertExecutionVisitor
try {
return Optional.of(statementExecutor.execute(statement));
} catch (final Exception e) {
-
LOGGER.warn(DataNodePipeMessages.FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE,
e);
+ if (LOGGER.isWarnEnabled()) {
+ PipeLogger.log(
+ LOGGER::warn,
+ DataNodePipeMessages
+
.FAILED_TO_EXECUTE_STATEMENT_AFTER_DATA_TYPE_CONVERSION_WITH_EXCEPTION_TYPE,
+ e.getClass().getName());
+ }
return Optional.empty();
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
index 381206258f1..460c3c5f846 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
@@ -209,7 +209,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
endPoint.getPort(),
e),
e,
- "Failed to borrow client %s:%s for cached leader.",
+ DataNodePipeMessages.FAILED_TO_BORROW_CLIENT_FOR_CACHED_LEADER,
endPoint.getIp(),
endPoint.getPort());
}
@@ -245,7 +245,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
if (response.getStatus().getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Handshake error with receiver %s:%s, code: %s, message:
%s.",
+ DataNodePipeMessages.HANDSHAKE_ERROR_WITH_RECEIVER,
targetNodeUrl.getIp(),
targetNodeUrl.getPort(),
response.getStatus().getCode(),
@@ -279,7 +279,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
PipeLogger.log(
LOGGER::warn,
e,
- "Handshake error with receiver %s:%s.",
+ DataNodePipeMessages.HANDSHAKE_ERROR_WITH_RECEIVER_1,
targetNodeUrl.getIp(),
targetNodeUrl.getPort());
exception.set(e);
@@ -332,8 +332,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
&& resp.get().getStatus().getCode() ==
TSStatusCode.PIPE_TYPE_ERROR.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Handshake error by PipeTransferHandshakeV2Req with receiver %s:%s
"
- + "retry to handshake by PipeTransferHandshakeV1Req.",
+ DataNodePipeMessages.HANDSHAKE_ERROR_BY_HANDSHAKE_V2_RETRY_WITH_V1,
targetNodeUrl.getIp(),
targetNodeUrl.getPort());
@@ -374,7 +373,7 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
targetNodeUrl.getPort(),
e),
e,
- "Failed to close client %s:%s after handshake failure when the
manager is closed.",
+
DataNodePipeMessages.FAILED_TO_CLOSE_CLIENT_AFTER_HANDSHAKE_FAILURE,
targetNodeUrl.getIp(),
targetNodeUrl.getPort());
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java
index b4766a2dae4..05e14ed08ae 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java
@@ -233,7 +233,7 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_TRY_TO_REMOVE_EVENT_AFTER,
consensusGroupId,
event),
- "IoTConsensusV2-ConsensusGroup-%s: try to remove event-%s after
iotConsensusV2AsyncConnector being closed. Ignore it.",
+
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_TRY_TO_REMOVE_EVENT_AFTER,
consensusGroupId,
event);
return;
@@ -254,7 +254,7 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
consensusGroupId,
event,
transferBuffer.size()),
- "IoTConsensusV2-ConsensusGroup-%s: event-%s not found in
transferBuffer, skip removing. queue size = %s",
+
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_EVENT_NOT_FOUND_IN_TRANSFERBUFFER,
consensusGroupId,
event,
transferBuffer.size());
@@ -540,7 +540,7 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
.IOTCONSENSUSV2_CONSENSUSGROUP_RETRYEVENTQUEUE_IS_NOT_EMPTY_AFTER,
consensusGroupId,
retryEventQueue.size()),
- "IoTConsensusV2-ConsensusGroup-%s: retryEventQueue is not empty
after 20 seconds. retryQueue size: %s",
+
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_RETRYEVENTQUEUE_IS_NOT_EMPTY_AFTER,
consensusGroupId,
retryEventQueue.size());
return;
@@ -563,7 +563,7 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
retryInterval,
peekedEvent.getReplicateIndexForIoTV2(),
peekedEvent),
- "IoTConsensusV2-ConsensusGroup-%s: retry with interval %s for
index %s %s",
+
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_RETRY_WITH_INTERVAL_FOR_INDEX,
consensusGroupId,
retryInterval,
peekedEvent.getReplicateIndexForIoTV2(),
@@ -586,7 +586,8 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
DataNodePipeMessages
.IOTCONSENSUSV2ASYNCCONNECTOR_DOES_NOT_SUPPORT_TRANSFER_GENERIC_EVENT,
peekedEvent),
- "IoTConsensusV2AsyncConnector does not support transfer
generic event: %s.",
+ DataNodePipeMessages
+
.IOTCONSENSUSV2ASYNCCONNECTOR_DOES_NOT_SUPPORT_TRANSFER_GENERIC_EVENT,
peekedEvent);
}
},
@@ -666,7 +667,8 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
consensusGroupId,
event,
event.getReplicateIndexForIoTV2()),
- "IoTConsensusV2-ConsensusGroup-%s: Event %s replicate index %s
transfer failed, will be added to retry queue.",
+ DataNodePipeMessages
+
.IOTCONSENSUSV2_CONSENSUSGROUP_EVENT_REPLICATE_INDEX_TRANSFER_FAILED_1,
consensusGroupId,
event,
event.getReplicateIndexForIoTV2());
@@ -679,7 +681,7 @@ public class IoTConsensusV2AsyncSink extends IoTDBSink
implements ConsensusPipeS
consensusGroupId,
event,
event.getReplicateIndexForIoTV2()),
- "IoTConsensusV2-ConsensusGroup-%s: Event %s replicate index %s
transfer failed, added to retry queue failed, this event will be ignored.",
+
DataNodePipeMessages.IOTCONSENSUSV2_CONSENSUSGROUP_EVENT_REPLICATE_INDEX_TRANSFER_FAILED,
consensusGroupId,
event,
event.getReplicateIndexForIoTV2());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java
index 36dfdffbf42..bcb850ea45a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java
@@ -112,7 +112,7 @@ public class IoTConsensusV2DeleteEventHandler
event.getReplicateIndexForIoTV2(),
e),
e,
- "Failed to transfer PipeDeleteNodeEvent %s (committer key=%s,
replicate index=%s).",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_PIPEDELETENODEEVENT_COMMITTER_KEY_REPLICATE,
event.coreReportMessage(),
event.getCommitterKey(),
event.getReplicateIndexForIoTV2());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java
index b107b7fe064..54d8b3e6319 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java
@@ -135,7 +135,7 @@ public class IoTConsensusV2TabletBatchEventHandler
pipeNames,
exception),
exception,
- "IoTConsensusV2: Failed to transfer TabletInsertionEvent batch. Total
failed events: %s, related pipe names: %s",
+
DataNodePipeMessages.IOTCONSENSUSV2_FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_BATCH_TOTAL,
events.size(),
pipeNames);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java
index a74e334d5e5..3e1efd17088 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java
@@ -124,7 +124,7 @@ public abstract class
IoTConsensusV2TabletInsertionEventHandler<
event.getReplicateIndexForIoTV2(),
exception),
exception,
- "Failed to transfer TabletInsertionEvent %s (committer key=%s,
replicate index=%s).",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_REPLICATE,
event.coreReportMessage(),
event.getCommitterKey(),
event.getReplicateIndexForIoTV2());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
index 52815e645bf..8ebff374531 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java
@@ -320,7 +320,7 @@ public class IoTConsensusV2TsFileInsertionEventHandler
event.getReplicateIndexForIoTV2(),
exception),
exception,
- "IoTConsensusV2-%s: Failed to transfer TsFileInsertionEvent %s
(committer key %s, replicate index %s).",
+
DataNodePipeMessages.IOTCONSENSUSV2_FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY,
consensusPipeName,
tsFile,
event.getCommitterKey(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
index 9ad0d81ddbe..ea32d6263aa 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
@@ -190,7 +190,8 @@ public class OpcUaSink implements PipeConnector {
validator.validate(
CONNECTOR_OPC_UA_MODEL_CLIENT_SERVER_VALUE::equals,
String.format(
- "When the OPC UA sink points to an outer server or sets
'with-quality' to true, the %s or %s must be %s.",
+ DataNodePipeMessages
+
.OPC_UA_SINK_MODEL_MUST_BE_CLIENT_SERVER_WHEN_OUTER_OR_WITH_QUALITY,
CONNECTOR_OPC_UA_MODEL_KEY,
SINK_OPC_UA_MODEL_KEY,
CONNECTOR_OPC_UA_MODEL_CLIENT_SERVER_VALUE),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
index 9aea22ab3e3..62e489b2e1f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
@@ -152,7 +152,7 @@ public class IoTDBOpcUaClient {
if (Objects.nonNull(sink.getValueName()) &&
!sink.getValueName().equals(name)) {
PipeLogger.log(
LOGGER::warn,
- "When the 'with-quality' mode is enabled, the measurement must be
either \"value-name\" or \"quality-name\"");
+
DataNodePipeMessages.WITH_QUALITY_MEASUREMENT_MUST_BE_VALUE_OR_QUALITY_NAME);
continue;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
index 4b7bac5a1d2..f95e8ed245b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -292,7 +292,7 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
if (Objects.nonNull(sink.getValueName()) &&
!sink.getValueName().equals(name)) {
PipeLogger.log(
LOGGER::warn,
- "When the 'with-quality' mode is enabled, the measurement must be
either \"value-name\" or \"quality-name\"");
+
DataNodePipeMessages.WITH_QUALITY_MEASUREMENT_MUST_BE_VALUE_OR_QUALITY_NAME);
continue;
}
final UaVariableNode measurementNode;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
index 250ebeabb39..213e43ab0b8 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java
@@ -281,7 +281,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
ignored ->
LOGGER.warn(DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH,
dbTsFilePairs, e),
e,
- "Failed to transfer tsfile batch (%s).",
+ DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_BATCH,
dbTsFilePairs);
if (eventsHadBeenAddedToRetryQueue.compareAndSet(false, true)) {
addFailureEventsToRetryQueue(events, e);
@@ -489,7 +489,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
pipeTransferTsFileHandler.getTsFile(),
e),
e,
- "Transfer tsfile event %s asynchronously was interrupted.",
+
DataNodePipeMessages.TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY_WAS_INTERRUPTED,
pipeTransferTsFileHandler.getTsFile());
}
@@ -501,7 +501,7 @@ public class IoTDBDataRegionAsyncSink extends IoTDBSink
implements PipeSinkWithS
pipeTransferTsFileHandler.getTsFile(),
e),
e,
- "Failed to transfer tsfile event %s asynchronously.",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILE_EVENT_ASYNCHRONOUSLY,
pipeTransferTsFileHandler.getTsFile());
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java
index c58a828625b..d0907c45abe 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletBatchEventHandler.java
@@ -118,7 +118,7 @@ public class PipeTransferTabletBatchEventHandler extends
PipeTransferTrackableHa
PipeLogger.log(
LOGGER::warn,
exception,
- "Failed to transfer TabletInsertionEvent batch. Total failed events:
%s, related pipe names: %s",
+ DataNodePipeMessages.FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_BATCH,
events.size(),
events.stream().map(EnrichedEvent::getPipeName).collect(Collectors.toSet()));
} finally {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java
index 1500be27894..94ca8ee5fe6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTabletInsertionEventHandler.java
@@ -93,7 +93,7 @@ public abstract class PipeTransferTabletInsertionEventHandler
extends PipeTransf
PipeLogger.log(
LOGGER::warn,
exception,
- "Failed to transfer TabletInsertionEvent %s (committer key=%s,
commit id=%s).",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_TABLETINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID,
event.coreReportMessage(),
event.getCommitterKey(),
event.getCommitId());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
index 7ec25753ea7..53967d94a6b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java
@@ -161,7 +161,7 @@ public abstract class PipeTransferTrackableHandler
PipeLogger.log(
LOGGER::warn,
- "The body size of the request is too large. The request will be
sliced. Origin req: %s-%s. Request body size: %s, threshold: %s",
+ DataNodePipeMessages.TRANSFER_REQUEST_BODY_TOO_LARGE_WILL_BE_SLICED,
req.getVersion(),
req.getType(),
req.body.limit(),
@@ -270,7 +270,7 @@ public abstract class PipeTransferTrackableHandler
PipeLogger.log(
LOGGER::warn,
exception,
- "Failed to transfer slice. Origin req: %s-%s. Retry the whole
transfer.",
+ DataNodePipeMessages.FAILED_TO_TRANSFER_SLICE_RETRY_WHOLE_TRANSFER,
originalReq.getVersion(),
originalReq.getType());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
index 700f241eb1f..4c25d03de53 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java
@@ -156,7 +156,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL,
sink.isClosed() ? "CLOSED" : "NOT CLOSED",
tsFile),
- "Client has been returned to the pool. Connector is %s. TsFile: %s.",
+ DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL,
sink.isClosed() ? "CLOSED" : "NOT CLOSED",
tsFile);
return;
@@ -375,7 +375,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
PipeLogger.log(
LOGGER::warn,
exception,
- "Failed to transfer TsFileInsertionEvent %s (committer key %s,
commit id %s).",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_COMMITTER_KEY_COMMIT_ID,
tsFile,
events.stream().map(EnrichedEvent::getCommitterKey).collect(Collectors.toList()),
events.stream().map(EnrichedEvent::getCommitIds).collect(Collectors.toList()));
@@ -383,7 +383,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
PipeLogger.log(
LOGGER::warn,
exception,
- "Failed to transfer TsFileInsertionEvent %s (batched
TableInsertionEvents).",
+
DataNodePipeMessages.FAILED_TO_TRANSFER_TSFILEINSERTIONEVENT_BATCHED_TABLE_EVENTS,
tsFile);
}
} catch (final Exception e) {
@@ -460,7 +460,7 @@ public class PipeTransferTsFileHandler extends
PipeTransferTrackableHandler {
DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL,
sink.isClosed() ? "CLOSED" : "NOT CLOSED",
tsFile),
- "Client has been returned to the pool. Connector is %s. TsFile: %s.",
+ DataNodePipeMessages.CLIENT_HAS_BEEN_RETURNED_TO_THE_POOL,
sink.isClosed() ? "CLOSED" : "NOT CLOSED",
tsFile);
return;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java
index 215c43c73d8..1d60c52eeac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java
@@ -856,7 +856,7 @@ public class WriteBackSink implements PipeConnector {
if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
PipeLogger.log(
LOGGER::warn,
- "Session {}: Failed to check authority for statement {}, username =
{}, response = {}.",
+ DataNodePipeMessages.SESSION_FAILED_TO_CHECK_AUTHORITY_FOR_STATEMENT,
session.getClientAddress() + ":" + session.getClientPort(),
statement.getType().name(),
session.getUsername(),
