Reduce repeated failure logs

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 =

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 =

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 {

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() {

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<TConsensusGroupId> 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();

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() {

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 =

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 =

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

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)

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,

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()));