From 41b4ece877383d958ab2f7343d97ffe3162312b5 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:42:19 +0800 Subject: [PATCH 1/2] Reduce repeated failure logs --- .../confignode/i18n/ManagerMessages.java | 13 ++++ .../confignode/i18n/ManagerMessages.java | 13 ++++ .../rpc/DataNodeTSStatusRPCHandler.java | 4 ++ .../confignode/conf/ConfigNodeDescriptor.java | 4 +- .../manager/partition/PartitionManager.java | 34 +++++++--- .../runtime/PipeConfigNodeRuntimeAgent.java | 10 +-- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 - .../iotdb/db/i18n/DataNodePipeMessages.java | 2 - .../apache/iotdb/db/conf/IoTDBDescriptor.java | 4 +- .../runtime/PipeDataNodeRuntimeAgent.java | 10 +-- .../thrift/IoTDBDataNodeReceiver.java | 4 +- .../thrift/impl/DataNodeRegionManager.java | 8 ++- .../schemaregion/SchemaRegionLoader.java | 8 ++- .../iotdb/commons/i18n/CommonMessages.java | 4 ++ .../iotdb/commons/i18n/PipeMessages.java | 5 +- .../iotdb/commons/i18n/CommonMessages.java | 4 ++ .../iotdb/commons/i18n/PipeMessages.java | 5 +- .../iotdb/commons/conf/CommonConfig.java | 47 +++++++++----- .../iotdb/commons/disk/FolderManager.java | 5 +- .../LoggerPeriodicalLogReducer.java} | 65 ++++++++++++------- .../commons/pipe/config/PipeDescriptor.java | 16 +++-- .../pipe/receiver/IoTDBFileReceiver.java | 16 +++-- .../commons/pipe/resource/log/PipeLogger.java | 22 +++---- .../LoggerPeriodicalLogReducerTest.java} | 21 +++--- 24 files changed, 214 insertions(+), 112 deletions(-) rename iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/{pipe/resource/log/PipePeriodicalLogReducer.java => log/LoggerPeriodicalLogReducer.java} (55%) rename iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/{pipe/resource/log/PipePeriodicalLogReducerTest.java => log/LoggerPeriodicalLogReducerTest.java} (67%) 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 082fcd9439bcd..e40a0fa7a1a78 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 = 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 072c2b728a2d3..4db8aca823f86 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 = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java index 629e7fc393f90..3597293b4b164 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.rpc.RpcUtils; @@ -91,6 +92,9 @@ public void onError(Exception e) { } private void logFailure(final String format, final Object... args) { + if (!LoggerPeriodicalLogReducer.shouldLog(format, args)) { + return; + } if (requestType == CnToDnAsyncRequestType.SUBSCRIPTION_PUSH_RUNTIME) { LOGGER.warn(format, args); } else { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java index 290e7c7f455cc..ef56f9c6125f2 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java @@ -25,8 +25,8 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.exception.BadNodeUrlException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.pipe.config.PipeDescriptor; -import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer; import org.apache.iotdb.commons.schema.SchemaConstant; import org.apache.iotdb.commons.utils.NodeUrlUtils; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; @@ -959,7 +959,7 @@ private void loadHotModifiedProcedureConfig(TrimProperties properties) throws IO private void loadPipeHotModifiedProp(TrimProperties properties) throws IOException { PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties, true); - PipePeriodicalLogReducer.update(); + LoggerPeriodicalLogReducer.update(); } public static ConfigNodeDescriptor getInstance() { 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 1cf416bedd7fb..7efdae278db3c 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 @@ -33,6 +33,7 @@ import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.enums.RepairDataPartitionTableProgressState; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.partition.DataPartitionTable; import org.apache.iotdb.commons.partition.SchemaPartitionTable; import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor; @@ -72,6 +73,7 @@ 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 +142,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(); @@ -307,7 +309,12 @@ public SchemaPartitionResp getOrCreateSchemaPartition(final GetOrCreateSchemaPar return resp; } - LOGGER.error(ManagerMessages.CREATE_SCHEMAPARTITION_FAILED_BECAUSE, e); + if (LoggerPeriodicalLogReducer.shouldLog( + ManagerMessages.CREATE_SCHEMAPARTITION_FAILED_BECAUSE + + e.getClass().getName() + + String.valueOf(e.getMessage()))) { + LOGGER.error(ManagerMessages.CREATE_SCHEMAPARTITION_FAILED_BECAUSE, e); + } resp.setStatus( new TSStatus(TSStatusCode.NO_AVAILABLE_REGION_GROUP.getStatusCode()) .setMessage(e.getMessage())); @@ -347,8 +354,11 @@ public SchemaPartitionResp getOrCreateSchemaPartition(final GetOrCreateSchemaPar 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 +521,11 @@ public DataPartitionResp getOrCreateDataPartition(final GetOrCreateDataPartition 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 +542,7 @@ public TSStatus dataPartitionTableIntegrityCheck() { || !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 +568,8 @@ public TShowRepairDataPartitionTableProgressResp showRepairDataPartitionTablePro 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 +1091,7 @@ public Optional generateTConsensusGroupIdByRegionId(final int } 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/agent/runtime/PipeConfigNodeRuntimeAgent.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java index fb0a89e71cabb..0599a215994e6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java @@ -21,13 +21,13 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor; import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalPhantomReferenceCleaner; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; -import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer; import org.apache.iotdb.commons.service.IService; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.confignode.i18n.ManagerMessages; @@ -59,7 +59,7 @@ public class PipeConfigNodeRuntimeAgent implements IService { @Override public synchronized void start() { PipeConfig.getInstance().printAllConfigs(); - initPipePeriodicalLogReducer(); + initLoggerPeriodicalLogReducer(); // PipeTasks will not be started here and will be started by "HandleLeaderChange" // procedure when the consensus layer notify leader ready @@ -96,10 +96,10 @@ public synchronized void stop() { LOGGER.info(ManagerMessages.PIPERUNTIMECONFIGNODEAGENT_STOPPED); } - private void initPipePeriodicalLogReducer() { - PipePeriodicalLogReducer.setMemoryResizeFunction( + private void initLoggerPeriodicalLogReducer() { + LoggerPeriodicalLogReducer.setMemoryResizeFunction( PipeConfigNodeResourceManager.memory()::resizeLogReducerMemory); - PipeLogger.setLogger(PipePeriodicalLogReducer::log); + PipeLogger.setLogger(LoggerPeriodicalLogReducer::log); } public boolean isShutdown() { 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 bcdab68abfd17..69ab04662e6ad 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 @@ -1313,8 +1313,6 @@ public final class DataNodePipeMessages { "Interrupted while waiting for the lock."; public static final String IS_RELEASED_AFTER_THREAD_INTERRUPTION = "{} is released after thread interruption."; - public static final String PIPEPERIODICALLOGREDUCER_IS_ALLOCATED_TO_BYTES = - "PipePeriodicalLogReducer is allocated to {} bytes."; public static final String PIPETSFILERESOURCE_CACHED_DEVICEISALIGNEDMAP_FOR_TSFILE = "PipeTsFileResource: Cached deviceIsAlignedMap for tsfile {}."; public static final String PIPETSFILERESOURCE_CACHED_OBJECTS_FOR_TSFILE = 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 3a51486d8474f..b88456698c113 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 @@ -1238,8 +1238,6 @@ public final class DataNodePipeMessages { public static final String INTERRUPTED_WHILE_WAITING_FOR_THE_LOCK = "等待锁时被中断。"; public static final String IS_RELEASED_AFTER_THREAD_INTERRUPTION = "{} 在线程中断后已被释放。"; - public static final String PIPEPERIODICALLOGREDUCER_IS_ALLOCATED_TO_BYTES = - "PipePeriodicalLogReducer 已分配 {} 字节。"; public static final String PIPETSFILERESOURCE_CACHED_DEVICEISALIGNEDMAP_FOR_TSFILE = "PipeTsFileResource：已为 tsfile {} 缓存 deviceIsAlignedMap。"; public static final String PIPETSFILERESOURCE_CACHED_OBJECTS_FOR_TSFILE = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 0830b7465399b..e3862bc0e9b99 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -26,9 +26,9 @@ import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.commons.conf.TrimProperties; import org.apache.iotdb.commons.exception.BadNodeUrlException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.memory.MemoryManager; import org.apache.iotdb.commons.pipe.config.PipeDescriptor; -import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer; import org.apache.iotdb.commons.schema.SchemaConstant; import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.utils.JVMCommonUtils; @@ -2749,7 +2749,7 @@ private void loadLoadTsFileHotModifiedProp(TrimProperties properties) throws IOE private void loadPipeHotModifiedProp(TrimProperties properties) throws IOException { PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties, true); - PipePeriodicalLogReducer.update(); + LoggerPeriodicalLogReducer.update(); } @SuppressWarnings("squid:S3518") // "proportionSum" can't be zero diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java index ed7f07d6ade69..1a589d5970d3d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/runtime/PipeDataNodeRuntimeAgent.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.exception.StartupException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.pipe.agent.runtime.PipePeriodicalJobExecutor; @@ -35,7 +36,6 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; -import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer; import org.apache.iotdb.commons.service.IService; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.utils.TestOnly; @@ -95,23 +95,23 @@ public synchronized void preparePipeResources( IoTDBTreePattern.setDevicePathGetter(PipeDataNodeRuntimeAgent::getPath); IoTDBTreePattern.setMeasurementPathGetter(PipeDataNodeRuntimeAgent::getPath); - initPipePeriodicalLogReducer(); + initLoggerPeriodicalLogReducer(); } - private void initPipePeriodicalLogReducer() { + private void initLoggerPeriodicalLogReducer() { if (pipeLogReducerMemoryBlock == null) { pipeLogReducerMemoryBlock = PipeDataNodeResourceManager.memory() .tryAllocate(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes()); } - PipePeriodicalLogReducer.setMemoryResizeFunction( + LoggerPeriodicalLogReducer.setMemoryResizeFunction( targetSizeInBytes -> { PipeDataNodeResourceManager.memory() .resize(pipeLogReducerMemoryBlock, Math.max(0, targetSizeInBytes), false); return pipeLogReducerMemoryBlock.getMemoryUsageInBytes(); }); - PipeLogger.setLogger(PipePeriodicalLogReducer::log); + PipeLogger.setLogger(LoggerPeriodicalLogReducer::log); } private static MeasurementPath getPath(final IDeviceID device, final String measurement) 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 46c00a095c841..d0d32c8a018a1 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 @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.exception.IoTDBException; import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; @@ -38,7 +39,6 @@ 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.resource.log.PipePeriodicalLogReducer; 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; @@ -1089,7 +1089,7 @@ private void logStatementExceptionIfNecessary(final Statement statement, final E static boolean shouldLogStatementException( final long receiverId, final Statement statement, final Exception e) { // Use the reducer cache as a gate. The actual stack trace is logged only when it passes. - return PipePeriodicalLogReducer.log( + return LoggerPeriodicalLogReducer.log( message -> {}, "Receiver id = %s, statement = %s, exception = %s, message = %s", receiverId, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java index 4d427fde9c191..9704ebc7c051a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.consensus.common.Peer; import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException; @@ -127,7 +128,12 @@ public TSStatus createSchemaRegion( tsStatus = new TSStatus(TSStatusCode.ILLEGAL_PATH.getStatusCode()); tsStatus.setMessage(DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED_ILLEGAL_PATH_MSG); } catch (final MetadataException e2) { - LOGGER.error(DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED, storageGroup, e2.getMessage()); + if (LoggerPeriodicalLogReducer.shouldLog( + DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED_FMT, + e2.getClass().getName() + String.valueOf(e2.getMessage()))) { + LOGGER.error( + DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED, storageGroup, e2.getMessage()); + } tsStatus = new TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode()); tsStatus.setMessage( String.format(DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED_FMT, e2.getMessage())); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java index b4f06e2e21118..365f380830af8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionLoader.java @@ -115,7 +115,13 @@ public ISchemaRegion createSchemaRegion(ISchemaRegionParams schemaRegionParams) throws MetadataException { try { return currentConstructor.newInstance(schemaRegionParams); - } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) { + } catch (InvocationTargetException e) { + if (e.getTargetException() instanceof MetadataException) { + throw (MetadataException) e.getTargetException(); + } + logger.warn(e.getMessage(), e); + throw new MetadataException(e); + } catch (InstantiationException | IllegalAccessException e) { logger.warn(e.getMessage(), e); throw new MetadataException(e); } diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index ed63849b13256..ae97d09ddc252 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -69,6 +69,10 @@ public final class CommonMessages { // --- subscription --- public static final String CONFIG_PRINT = "{}: {}"; + // --- log --- + public static final String LOG_LOGGERPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_C8373CF5 = + "LoggerPeriodicalLogReducer is allocated to {} bytes."; + // --- partition --- public static final String DATABASE_NOT_EXISTS_AND_AUTO_CREATE_DISABLED = "Database %s not exists and failed to create automatically because enable_auto_create_schema is FALSE."; diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java index afb977f85a204..eb38e3524a2ec 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -187,7 +187,7 @@ public final class PipeMessages { public static final String CONFIG_PIPE_RECEIVER_LOAD_CONVERSION_ENABLED = "PipeReceiverLoadConversionEnabled: {}"; public static final String CONFIG_PIPE_PERIODICAL_LOG_MIN_INTERVAL_SECONDS = - "PipePeriodicalLogMinIntervalSeconds: {}"; + "LoggerPeriodicalLogMinIntervalSeconds: {}"; public static final String CONFIG_PIPE_RETRY_LOCALLY_FOR_PARALLEL_OR_USER_CONFLICT = "PipeRetryLocallyForParallelOrUserConflict: {}"; public static final String CONFIG_PIPE_META_REPORT_MAX_LOG_NUM_PER_ROUND = @@ -199,7 +199,7 @@ public final class PipeMessages { public static final String CONFIG_PIPE_TSFILE_PIN_MAX_LOG_INTERVAL_ROUNDS = "PipeTsFilePinMaxLogIntervalRounds: {}"; public static final String CONFIG_PIPE_LOGGER_CACHE_MAX_SIZE_IN_BYTES = - "PipeLoggerCacheMaxSizeInBytes: {}"; + "LoggerCacheMaxSizeInBytes: {}"; public static final String CONFIG_PIPE_MEMORY_MANAGEMENT_ENABLED = "PipeMemoryManagementEnabled: {}"; public static final String CONFIG_PIPE_MEMORY_ALLOCATE_MAX_RETRIES = @@ -947,6 +947,5 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_COMPLETED_SUCCESSFULLY_E076E3B2 = "DataPartitionTable generation completed successfully"; public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable generation failed: "; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "Unknown task status: "; - public static final String MESSAGE_PIPEPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_54E0E369 = "PipePeriodicalLogReducer is allocated to {} bytes."; } diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index ce41f1f3bfc8a..1c0e2ee212588 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -68,6 +68,10 @@ public final class CommonMessages { // --- subscription --- public static final String CONFIG_PRINT = "{}：{}"; + // --- log --- + public static final String LOG_LOGGERPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_C8373CF5 = + "LoggerPeriodicalLogReducer \u5df2\u5206\u914d {} \u5b57\u8282\u3002"; + // --- partition --- public static final String DATABASE_NOT_EXISTS_AND_AUTO_CREATE_DISABLED = "Database %s 不存在，且由于 enable_auto_create_schema 为 FALSE，无法自动创建。"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java index 87f1601a7544b..6914f4b13793b 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/PipeMessages.java @@ -186,7 +186,7 @@ public final class PipeMessages { public static final String CONFIG_PIPE_RECEIVER_LOAD_CONVERSION_ENABLED = "PipeReceiverLoadConversionEnabled: {}"; public static final String CONFIG_PIPE_PERIODICAL_LOG_MIN_INTERVAL_SECONDS = - "PipePeriodicalLogMinIntervalSeconds: {}"; + "LoggerPeriodicalLogMinIntervalSeconds: {}"; public static final String CONFIG_PIPE_RETRY_LOCALLY_FOR_PARALLEL_OR_USER_CONFLICT = "PipeRetryLocallyForParallelOrUserConflict: {}"; public static final String CONFIG_PIPE_META_REPORT_MAX_LOG_NUM_PER_ROUND = @@ -198,7 +198,7 @@ public final class PipeMessages { public static final String CONFIG_PIPE_TSFILE_PIN_MAX_LOG_INTERVAL_ROUNDS = "PipeTsFilePinMaxLogIntervalRounds: {}"; public static final String CONFIG_PIPE_LOGGER_CACHE_MAX_SIZE_IN_BYTES = - "PipeLoggerCacheMaxSizeInBytes: {}"; + "LoggerCacheMaxSizeInBytes: {}"; public static final String CONFIG_PIPE_MEMORY_MANAGEMENT_ENABLED = "PipeMemoryManagementEnabled: {}"; public static final String CONFIG_PIPE_MEMORY_ALLOCATE_MAX_RETRIES = @@ -915,6 +915,5 @@ private PipeMessages() {} public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_COMPLETED_SUCCESSFULLY_E076E3B2 = "DataPartitionTable 生成已成功完成"; public static final String MESSAGE_DATAPARTITIONTABLE_GENERATION_FAILED_D85CD23A = "DataPartitionTable 生成失败："; public static final String MESSAGE_UNKNOWN_TASK_STATUS_E05D98F0 = "未知任务状态："; - public static final String MESSAGE_PIPEPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_54E0E369 = "PipePeriodicalLogReducer 已分配 {} 字节。"; } 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 8b98da6bc3172..c811ae9f5f904 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 @@ -329,8 +329,8 @@ public class CommonConfig { 64 * 1024 * 1024, (int) Math.min(Runtime.getRuntime().maxMemory() / 64, Integer.MAX_VALUE)); private boolean pipeReceiverLoadConversionEnabled = false; - private volatile long pipePeriodicalLogMinIntervalSeconds = 60; - private volatile long pipeLoggerCacheMaxSizeInBytes = 16 * MB; + private volatile long loggerPeriodicalLogMinIntervalSeconds = 60; + private volatile long loggerCacheMaxSizeInBytes = 64 * MB; private volatile double pipeMetaReportMaxLogNumPerRound = 0.1; private volatile int pipeMetaReportMaxLogIntervalRounds = 360; @@ -1875,33 +1875,48 @@ public void setPipeReceiverLoadConversionEnabled(boolean pipeReceiverLoadConvers pipeReceiverLoadConversionEnabled); } + public long getLoggerPeriodicalLogMinIntervalSeconds() { + return loggerPeriodicalLogMinIntervalSeconds; + } + + public void setLoggerPeriodicalLogMinIntervalSeconds(long loggerPeriodicalLogMinIntervalSeconds) { + if (this.loggerPeriodicalLogMinIntervalSeconds == loggerPeriodicalLogMinIntervalSeconds) { + return; + } + this.loggerPeriodicalLogMinIntervalSeconds = loggerPeriodicalLogMinIntervalSeconds; + logger.info( + ConfigMessages.CONFIG_SET_TO, + "loggerPeriodicalLogMinIntervalSeconds", + loggerPeriodicalLogMinIntervalSeconds); + } + public long getPipePeriodicalLogMinIntervalSeconds() { - return pipePeriodicalLogMinIntervalSeconds; + return getLoggerPeriodicalLogMinIntervalSeconds(); } public void setPipePeriodicalLogMinIntervalSeconds(long pipePeriodicalLogMinIntervalSeconds) { - if (this.pipePeriodicalLogMinIntervalSeconds == pipePeriodicalLogMinIntervalSeconds) { + setLoggerPeriodicalLogMinIntervalSeconds(pipePeriodicalLogMinIntervalSeconds); + } + + public long getLoggerCacheMaxSizeInBytes() { + return loggerCacheMaxSizeInBytes; + } + + public void setLoggerCacheMaxSizeInBytes(long loggerCacheMaxSizeInBytes) { + if (this.loggerCacheMaxSizeInBytes == loggerCacheMaxSizeInBytes) { return; } - this.pipePeriodicalLogMinIntervalSeconds = pipePeriodicalLogMinIntervalSeconds; + this.loggerCacheMaxSizeInBytes = loggerCacheMaxSizeInBytes; logger.info( - ConfigMessages.LOG_PIPEPERIODICALLOGMININTERVALSECONDS_SET_ARG_5535C79E, - pipePeriodicalLogMinIntervalSeconds); + ConfigMessages.CONFIG_SET_TO, "loggerCacheMaxSizeInBytes", loggerCacheMaxSizeInBytes); } public long getPipeLoggerCacheMaxSizeInBytes() { - return pipeLoggerCacheMaxSizeInBytes; + return getLoggerCacheMaxSizeInBytes(); } public void setPipeLoggerCacheMaxSizeInBytes(long pipeLoggerCacheMaxSizeInBytes) { - if (this.pipeLoggerCacheMaxSizeInBytes == pipeLoggerCacheMaxSizeInBytes) { - return; - } - this.pipeLoggerCacheMaxSizeInBytes = pipeLoggerCacheMaxSizeInBytes; - logger.info( - ConfigMessages.CONFIG_SET_TO, - "pipeLoggerCacheMaxSizeInBytes", - pipeLoggerCacheMaxSizeInBytes); + setLoggerCacheMaxSizeInBytes(pipeLoggerCacheMaxSizeInBytes); } public int getPipeReceiverReqDecompressedMaxLengthInBytes() { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java index 08e2b7872efff..a7707077f84dd 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.disk.strategy.SequenceStrategy; import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException; import org.apache.iotdb.commons.i18n.UtilMessages; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.utils.JVMCommonUtils; import org.slf4j.Logger; @@ -124,7 +125,9 @@ private boolean hasFolderWithAvailableDiskSpace() { private void changeToReadOnlyIfDiskFull(DiskSpaceInsufficientException e) { if (!hasFolderWithAvailableDiskSpace()) { - logger.error(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY, e); + if (LoggerPeriodicalLogReducer.shouldLog(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY)) { + logger.error(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY, e); + } CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly); CommonDescriptor.getInstance().getConfig().setStatusReason(NodeStatus.DISK_FULL); } else { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java similarity index 55% rename from iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java rename to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java index 2f4242857aa3d..4f863e2cc40cb 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducer.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducer.java @@ -17,23 +17,24 @@ * under the License. */ -package org.apache.iotdb.commons.pipe.resource.log; +package org.apache.iotdb.commons.log; -import org.apache.iotdb.commons.i18n.PipeMessages; -import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.i18n.CommonMessages; import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; import org.apache.tsfile.utils.RamUsageEstimator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.helpers.MessageFormatter; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.LongUnaryOperator; -public class PipePeriodicalLogReducer { - private static final Logger LOGGER = LoggerFactory.getLogger(PipePeriodicalLogReducer.class); +public class LoggerPeriodicalLogReducer { + private static final Logger LOGGER = LoggerFactory.getLogger(LoggerPeriodicalLogReducer.class); private static final LongUnaryOperator DEFAULT_MEMORY_RESIZE_FUNCTION = sizeInBytes -> sizeInBytes; @@ -42,10 +43,9 @@ public class PipePeriodicalLogReducer { protected static final Cache LOGGER_CACHE = Caffeine.newBuilder() - .expireAfterWrite( - PipeConfig.getInstance().getPipePeriodicalLogMinIntervalSeconds(), TimeUnit.SECONDS) - .weigher(PipePeriodicalLogReducer::estimateSize) - .maximumWeight(PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes()) + .expireAfterWrite(getLogMinIntervalSeconds(), TimeUnit.SECONDS) + .weigher(LoggerPeriodicalLogReducer::estimateSize) + .maximumWeight(getLoggerCacheMaxSizeInBytes()) .build(); private static int estimateSize(final String key, final String value) { @@ -55,28 +55,33 @@ private static int estimateSize(final String key, final String value) { public static boolean log( final Consumer loggerFunction, final String rawMessage, final Object... formatter) { - final String loggerMessage = PipeLogger.formatMessage(rawMessage, formatter); - if (!LOGGER_CACHE.asMap().containsKey(loggerMessage)) { - LOGGER_CACHE.put(loggerMessage, loggerMessage); + final String loggerMessage = formatMessage(rawMessage, formatter); + if (shouldLog(loggerMessage)) { loggerFunction.accept(loggerMessage); return true; } return false; } + public static boolean shouldLog(final String loggerMessage) { + return LOGGER_CACHE.asMap().putIfAbsent(loggerMessage, loggerMessage) == null; + } + + public static boolean shouldLog(final String rawMessage, final Object... formatter) { + return shouldLog(formatMessage(rawMessage, formatter)); + } + public static synchronized void setMemoryResizeFunction( final LongUnaryOperator memoryResizeFunction) { - PipePeriodicalLogReducer.memoryResizeFunction = + LoggerPeriodicalLogReducer.memoryResizeFunction = memoryResizeFunction == null ? DEFAULT_MEMORY_RESIZE_FUNCTION : memoryResizeFunction; update(); } public static synchronized void update() { - final long maxWeight = - memoryResizeFunction.applyAsLong( - PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes()); + final long maxWeight = memoryResizeFunction.applyAsLong(getLoggerCacheMaxSizeInBytes()); LOGGER.info( - PipeMessages.MESSAGE_PIPEPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_54E0E369, + CommonMessages.LOG_LOGGERPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_C8373CF5, maxWeight); update(maxWeight); } @@ -85,15 +90,29 @@ public static synchronized void update(final long maxWeight) { LOGGER_CACHE .policy() .expireAfterWrite() - .ifPresent( - time -> - time.setExpiresAfter( - PipeConfig.getInstance().getPipePeriodicalLogMinIntervalSeconds(), - TimeUnit.SECONDS)); + .ifPresent(time -> time.setExpiresAfter(getLogMinIntervalSeconds(), TimeUnit.SECONDS)); LOGGER_CACHE.policy().eviction().ifPresent(eviction -> eviction.setMaximum(maxWeight)); } - private PipePeriodicalLogReducer() { + public static String formatMessage(final String rawMessage, final Object... formatter) { + if (formatter == null || formatter.length == 0) { + return rawMessage; + } + if (rawMessage.contains("{}")) { + return MessageFormatter.arrayFormat(rawMessage, formatter).getMessage(); + } + return String.format(rawMessage, formatter); + } + + private static long getLogMinIntervalSeconds() { + return CommonDescriptor.getInstance().getConfig().getLoggerPeriodicalLogMinIntervalSeconds(); + } + + private static long getLoggerCacheMaxSizeInBytes() { + return CommonDescriptor.getInstance().getConfig().getLoggerCacheMaxSizeInBytes(); + } + + private LoggerPeriodicalLogReducer() { // static } } 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 94181010520c3..780d6b3252a07 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 @@ -468,16 +468,20 @@ public static void loadPipeInternalConfig(CommonConfig config, TrimProperties pr properties.getProperty( "pipe_receiver_load_conversion_enabled", String.valueOf(config.isPipeReceiverLoadConversionEnabled())))); - config.setPipePeriodicalLogMinIntervalSeconds( + config.setLoggerPeriodicalLogMinIntervalSeconds( Long.parseLong( properties.getProperty( - "pipe_periodical_log_min_interval_seconds", - String.valueOf(config.getPipePeriodicalLogMinIntervalSeconds())))); - config.setPipeLoggerCacheMaxSizeInBytes( + "logger_periodical_log_min_interval_seconds", + properties.getProperty( + "pipe_periodical_log_min_interval_seconds", + String.valueOf(config.getLoggerPeriodicalLogMinIntervalSeconds()))))); + config.setLoggerCacheMaxSizeInBytes( Long.parseLong( properties.getProperty( - "pipe_logger_cache_max_size_in_bytes", - String.valueOf(config.getPipeLoggerCacheMaxSizeInBytes())))); + "logger_cache_max_size_in_bytes", + properties.getProperty( + "pipe_logger_cache_max_size_in_bytes", + String.valueOf(config.getLoggerCacheMaxSizeInBytes()))))); config.setPipeMemoryAllocateMaxRetries( Integer.parseInt( 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 c39b0d93c6fe4..4078d529539f9 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException; import org.apache.iotdb.commons.i18n.PipeMessages; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; @@ -181,13 +182,20 @@ protected TPipeTransferResp handleTransferHandshakeV1(final PipeTransferHandshak try { receiverFileBaseDir = getReceiverFileBaseDir(); if (Objects.isNull(receiverFileBaseDir)) { - PipeLogger.log( - LOGGER::warn, PipeMessages.RECEIVER_FAILED_INIT_FOLDER_FULL, receiverId.get()); + if (LoggerPeriodicalLogReducer.shouldLog(PipeMessages.RECEIVER_FAILED_INIT_FOLDER_FULL)) { + PipeLogger.log( + LOGGER::warn, PipeMessages.RECEIVER_FAILED_INIT_FOLDER_FULL, receiverId.get()); + } return new TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT)); } } catch (Exception e) { - PipeLogger.log( - LOGGER::warn, e, PipeMessages.RECEIVER_FAILED_CREATE_FOLDER_FULL, receiverId.get()); + if (LoggerPeriodicalLogReducer.shouldLog( + PipeMessages.RECEIVER_FAILED_CREATE_FOLDER_FULL + + e.getClass().getName() + + e.getMessage())) { + PipeLogger.log( + LOGGER::warn, e, PipeMessages.RECEIVER_FAILED_CREATE_FOLDER_FULL, receiverId.get()); + } return new TPipeTransferResp(StatusUtils.getStatus(TSStatusCode.DISK_SPACE_INSUFFICIENT)); } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java index 97ff4afba6342..720999aae9718 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/resource/log/PipeLogger.java @@ -19,7 +19,7 @@ package org.apache.iotdb.commons.pipe.resource.log; -import org.slf4j.helpers.MessageFormatter; +import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer; import java.io.ByteArrayOutputStream; import java.io.PrintStream; @@ -28,11 +28,12 @@ public class PipeLogger { private static PipePeriodicalLogger logger = (loggerFunction, rawMessage, formatter) -> - loggerFunction.accept(formatMessage(rawMessage, formatter)); + loggerFunction.accept(LoggerPeriodicalLogReducer.formatMessage(rawMessage, formatter)); public static void log( final Consumer loggerFunction, final String rawMessage, final Object... formatter) { - logger.log(loggerFunction, "%s", formatMessage(rawMessage, formatter)); + logger.log( + loggerFunction, "%s", LoggerPeriodicalLogReducer.formatMessage(rawMessage, formatter)); } public static void log( @@ -42,7 +43,10 @@ public static void log( final Object... formatter) { final ByteArrayOutputStream out = new ByteArrayOutputStream(); throwable.printStackTrace(new PrintStream(out)); - logger.log(loggerFunction, "%s", formatMessage(rawMessage, formatter) + "\n" + out); + logger.log( + loggerFunction, + "%s", + LoggerPeriodicalLogReducer.formatMessage(rawMessage, formatter) + "\n" + out); } public static void setLogger(final PipePeriodicalLogger logger) { @@ -53,16 +57,6 @@ private PipeLogger() { // static } - static String formatMessage(final String rawMessage, final Object... formatter) { - if (formatter == null || formatter.length == 0) { - return rawMessage; - } - if (rawMessage.contains("{}")) { - return MessageFormatter.arrayFormat(rawMessage, formatter).getMessage(); - } - return String.format(rawMessage, formatter); - } - @FunctionalInterface public interface PipePeriodicalLogger { void log(final Consumer loggerFunction, final String rawMessage, final Object... args); diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java similarity index 67% rename from iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java rename to iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java index d390cbb9fddb7..a0a6df14e9c0d 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/resource/log/PipePeriodicalLogReducerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/log/LoggerPeriodicalLogReducerTest.java @@ -17,9 +17,9 @@ * under the License. */ -package org.apache.iotdb.commons.pipe.resource.log; +package org.apache.iotdb.commons.log; -import org.apache.iotdb.commons.pipe.config.PipeConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.junit.After; import org.junit.Assert; @@ -28,20 +28,20 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; -public class PipePeriodicalLogReducerTest { +public class LoggerPeriodicalLogReducerTest { @After public void tearDown() { - PipePeriodicalLogReducer.setMemoryResizeFunction(null); + LoggerPeriodicalLogReducer.setMemoryResizeFunction(null); } @Test public void testLogReducesDuplicateMessages() { final AtomicInteger logCount = new AtomicInteger(0); - final String message = "PipePeriodicalLogReducerTest-" + System.nanoTime(); + final String message = "LoggerPeriodicalLogReducerTest-" + System.nanoTime(); - Assert.assertTrue(PipePeriodicalLogReducer.log(log -> logCount.incrementAndGet(), message)); - Assert.assertFalse(PipePeriodicalLogReducer.log(log -> logCount.incrementAndGet(), message)); + Assert.assertTrue(LoggerPeriodicalLogReducer.log(log -> logCount.incrementAndGet(), message)); + Assert.assertFalse(LoggerPeriodicalLogReducer.log(log -> logCount.incrementAndGet(), message)); Assert.assertEquals(1, logCount.get()); } @@ -50,16 +50,17 @@ public void testUpdateUsesMemoryResizeFunction() { final AtomicLong requestedSizeInBytes = new AtomicLong(-1); final long allocatedSizeInBytes = 1024; - PipePeriodicalLogReducer.setMemoryResizeFunction( + LoggerPeriodicalLogReducer.setMemoryResizeFunction( sizeInBytes -> { requestedSizeInBytes.set(sizeInBytes); return allocatedSizeInBytes; }); Assert.assertEquals( - PipeConfig.getInstance().getPipeLoggerCacheMaxSizeInBytes(), requestedSizeInBytes.get()); + CommonDescriptor.getInstance().getConfig().getLoggerCacheMaxSizeInBytes(), + requestedSizeInBytes.get()); Assert.assertEquals( allocatedSizeInBytes, - PipePeriodicalLogReducer.LOGGER_CACHE.policy().eviction().get().getMaximum()); + LoggerPeriodicalLogReducer.LOGGER_CACHE.policy().eviction().get().getMaximum()); } } From 1cf504cc49755eac1044ed8f37b7712676c400ec Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 13 Jul 2026 10:40:54 +0800 Subject: [PATCH 2/2] Fix Chinese log message encoding --- .../i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 1c0e2ee212588..71c45ccaf3fc0 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -70,7 +70,7 @@ public final class CommonMessages { // --- log --- public static final String LOG_LOGGERPERIODICALLOGREDUCER_IS_ALLOCATED_TO_ARG_BYTES_C8373CF5 = - "LoggerPeriodicalLogReducer \u5df2\u5206\u914d {} \u5b57\u8282\u3002"; + "LoggerPeriodicalLogReducer 已分配 {} 字节。"; // --- partition --- public static final String DATABASE_NOT_EXISTS_AND_AUTO_CREATE_DISABLED =