Limit repeated warning logs

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java

@@ -29,6 +29,7 @@
 import org.apache.iotdb.commons.consensus.ConsensusGroupId;
 import org.apache.iotdb.commons.file.SystemFileFactory;
 import org.apache.iotdb.commons.request.IConsensusRequest;
+import org.apache.iotdb.commons.utils.LogThrottler;
 import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
@@ -112,6 +113,8 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
   /** Variables for {@link ConfigNode} Simple Consensus. */
   private LogWriter simpleLogWriter;
 
+  private final LogThrottler simpleConsensusWalFlushFailureLogThrottler = new LogThrottler();
+
   private File simpleLogFile;
   private int startIndex;
   private int endIndex;
@@ -681,9 +684,14 @@ private void flushWALForSimpleConsensus() {
     if (simpleLogWriter != null) {
       try {
         simpleLogWriter.force();
+        simpleConsensusWalFlushFailureLogThrottler.reset();
       } catch (IOException e) {
-        LOGGER.error(
-            ConfigNodeMessages.CAN_T_FORCE_LOGWRITER_FOR_CONFIGNODE_FLUSHWALFORSIMPLECONSENSUS, e);
+        if (simpleConsensusWalFlushFailureLogThrottler.shouldLog(
+            LogThrottler.getFailureSignature(e))) {
+          LOGGER.error(
+              ConfigNodeMessages.CAN_T_FORCE_LOGWRITER_FOR_CONFIGNODE_FLUSHWALFORSIMPLECONSENSUS,
+              e);
+        }
       }
     }
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionMetaSyncer.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.commons.concurrent.ThreadName;
 import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
 import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import org.apache.iotdb.commons.utils.LogThrottler;
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
 import org.apache.iotdb.confignode.manager.ConfigManager;
 import org.apache.iotdb.confignode.manager.ProcedureManager;
@@ -49,6 +50,7 @@ public class SubscriptionMetaSyncer {
       SubscriptionConfig.getInstance().getSubscriptionMetaSyncerSyncIntervalMinutes();
 
   private final ConfigManager configManager;
+  private final LogThrottler syncFailureLogThrottler = new LogThrottler();
 
   private Future<?> metaSyncFuture;
 
@@ -84,39 +86,55 @@ private synchronized void sync() {
     isLastSubscriptionSyncSuccessful = false;
 
     if (configManager.getSubscriptionManager().getSubscriptionCoordinator().isLocked()) {
-      LOGGER.warn(
-          ManagerMessages.SUBSCRIPTIONCOORDINATORLOCK_IS_HELD_BY_ANOTHER_THREAD_SKIP_THIS_ROUND_OF);
+      if (syncFailureLogThrottler.shouldLog("coordinator-lock-held", "locked")) {
+        LOGGER.warn(
+            ManagerMessages
+                .SUBSCRIPTIONCOORDINATORLOCK_IS_HELD_BY_ANOTHER_THREAD_SKIP_THIS_ROUND_OF);
+      }
       return;
     }
+    syncFailureLogThrottler.reset("coordinator-lock-held");
 
     final ProcedureManager procedureManager = configManager.getProcedureManager();
 
     // sync topic meta firstly
     // TODO: consider drop the topic which is subscribed by consumers
     final TSStatus topicMetaSyncStatus = procedureManager.topicMetaSync();
     if (topicMetaSyncStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      LOGGER.warn(ManagerMessages.FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS, topicMetaSyncStatus);
+      if (syncFailureLogThrottler.shouldLog(
+          "topic-meta-sync", getStatusFailureSignature(topicMetaSyncStatus))) {
+        LOGGER.warn(ManagerMessages.FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS, topicMetaSyncStatus);
+      }
       return;
     }
+    syncFailureLogThrottler.reset("topic-meta-sync");
 
     // sync consumer meta if syncing topic meta successfully
     final TSStatus consumerGroupMetaSyncStatus = procedureManager.consumerGroupMetaSync();
     if (consumerGroupMetaSyncStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      LOGGER.warn(
-          ManagerMessages.FAILED_TO_SYNC_CONSUMER_GROUP_META_RESULT_STATUS,
-          consumerGroupMetaSyncStatus);
+      if (syncFailureLogThrottler.shouldLog(
+          "consumer-group-meta-sync", getStatusFailureSignature(consumerGroupMetaSyncStatus))) {
+        LOGGER.warn(
+            ManagerMessages.FAILED_TO_SYNC_CONSUMER_GROUP_META_RESULT_STATUS,
+            consumerGroupMetaSyncStatus);
+      }
       return;
     }
+    syncFailureLogThrottler.reset("consumer-group-meta-sync");
 
     // sync commit progress if syncing consumer group meta successfully
     final TSStatus commitProgressSyncStatus = procedureManager.commitProgressSync();
     if (commitProgressSyncStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-      LOGGER.warn(
-          ManagerMessages.LOG_FAILED_SYNC_COMMIT_PROGRESS_RESULT_STATUS_ARG_A9E46E80,
-          commitProgressSyncStatus);
+      if (syncFailureLogThrottler.shouldLog(
+          "commit-progress-sync", getStatusFailureSignature(commitProgressSyncStatus))) {
+        LOGGER.warn(
+            ManagerMessages.LOG_FAILED_SYNC_COMMIT_PROGRESS_RESULT_STATUS_ARG_A9E46E80,
+            commitProgressSyncStatus);
+      }
       return;
     }
 
+    syncFailureLogThrottler.reset();
     LOGGER.info(
         ManagerMessages.AFTER_THIS_SUCCESSFUL_SYNC_IF_SUBSCRIPTIONINFO_IS_EMPTY_DURING_THIS);
     isLastSubscriptionSyncSuccessful = true;
@@ -126,7 +144,12 @@ public synchronized void stop() {
     if (metaSyncFuture != null) {
       metaSyncFuture.cancel(false);
       metaSyncFuture = null;
+      syncFailureLogThrottler.reset();
       LOGGER.info(ManagerMessages.SUBSCRIPTIONMETASYNCER_IS_STOPPED_SUCCESSFULLY);
     }
   }
+
+  private static String getStatusFailureSignature(TSStatus status) {
+    return status.getCode() + ":" + status.getMessage();
+  }
 }

iotdb-core/datanode/pom.xml

@@ -485,6 +485,14 @@
                         <ignoredDependency>org.apache.iotdb:isession</ignoredDependency>
                         <ignoredDependency>at.yawk.lz4:lz4-java</ignoredDependency>
                     </ignoredDependencies>
+                    <ignoredUsedUndeclaredDependencies>
+                        <ignoredUsedUndeclaredDependency>ch.qos.logback:logback-classic</ignoredUsedUndeclaredDependency>
+                        <ignoredUsedUndeclaredDependency>ch.qos.logback:logback-core</ignoredUsedUndeclaredDependency>
+                    </ignoredUsedUndeclaredDependencies>
+                    <ignoredNonTestScopedDependencies>
+                        <ignoredNonTestScopedDependency>ch.qos.logback:logback-classic</ignoredNonTestScopedDependency>
+                        <ignoredNonTestScopedDependency>ch.qos.logback:logback-core</ignoredNonTestScopedDependency>
+                    </ignoredNonTestScopedDependencies>
                 </configuration>
             </plugin>
             <plugin>

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java

@@ -72,6 +72,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiFunction;
 
 import static org.apache.iotdb.commons.auth.utils.AuthUtils.constructAuthorityScope;
@@ -89,6 +90,8 @@ public class ClusterAuthorityFetcher implements IAuthorityFetcher {
       ConfigNodeClientManager.getInstance();
 
   private static final String CONNECTERROR = "Failed to connect to config node.";
+  private static final long CONNECT_ERROR_LOG_INTERVAL_MS = 60_000L;
+  private static final AtomicLong LAST_CONNECT_ERROR_LOG_TIME = new AtomicLong(0L);
 
   public ClusterAuthorityFetcher(IAuthorCache iAuthorCache) {
     this.iAuthorCache = iAuthorCache;
@@ -370,8 +373,9 @@ private PathPatternTree fetchAuthizedPatternTree(String username, PrivilegeType
     try (ConfigNodeClient configNodeClient =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
       authizedPatternTree = configNodeClient.fetchAuthizedPatternTree(req);
+      resetConfigNodeConnectionErrorLogTime();
     } catch (ClientManagerException | TException e) {
-      LOGGER.error(CONNECTERROR);
+      logConfigNodeConnectionError();
       authizedPatternTree.setStatus(
           RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, CONNECTERROR));
     }
@@ -397,6 +401,7 @@ private SettableFuture<ConfigTaskResult> operatePermissionInternal(
                   statementToAuthorizerReq((RelationalAuthorStatement) plan))
               : configNodeClient.operatePermission(
                   statementToAuthorizerReq((AuthorStatement) plan));
+      resetConfigNodeConnectionErrorLogTime();
       if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
         future.setException(new IoTDBException(tsStatus));
       } else {
@@ -406,7 +411,7 @@ private SettableFuture<ConfigTaskResult> operatePermissionInternal(
     } catch (AuthException e) {
       future.setException(e);
     } catch (ClientManagerException | TException e) {
-      LOGGER.error(CONNECTERROR);
+      logConfigNodeConnectionError();
       future.setException(e);
     }
     return future;
@@ -471,6 +476,7 @@ private SettableFuture<ConfigTaskResult> queryPermissionInternal(
               ? configNodeClient.queryRPermission(
                   statementToAuthorizerReq((RelationalAuthorStatement) plan))
               : configNodeClient.queryPermission(statementToAuthorizerReq((AuthorStatement) plan));
+      resetConfigNodeConnectionErrorLogTime();
       if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != authorizerResp.getStatus().getCode()) {
         future.setException(
             new IoTDBException(
@@ -481,7 +487,7 @@ private SettableFuture<ConfigTaskResult> queryPermissionInternal(
     } catch (AuthException e) {
       future.setException(e);
     } catch (ClientManagerException | TException e) {
-      LOGGER.error(CONNECTERROR);
+      logConfigNodeConnectionError();
       authorizerResp.setStatus(
           RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, CONNECTERROR));
       future.setException(new IoTDBException(authorizerResp.getStatus()));
@@ -561,8 +567,9 @@ public TSStatus checkUser(
           CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
         // Send request to some API server
         status = configNodeClient.login(req);
+        resetConfigNodeConnectionErrorLogTime();
       } catch (ClientManagerException | TException e) {
-        LOGGER.error(CONNECTERROR);
+        logConfigNodeConnectionError();
         status = new TPermissionInfoResp();
         status.setStatus(RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, CONNECTERROR));
       } finally {
@@ -593,8 +600,9 @@ public User getUser(String userName, final boolean force) {
           CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
         // Send request to some API server
         permissionInfoResp = configNodeClient.getUser(userName);
+        resetConfigNodeConnectionErrorLogTime();
       } catch (ClientManagerException | TException e) {
-        LOGGER.error(CONNECTERROR);
+        logConfigNodeConnectionError();
       }
       if (permissionInfoResp != null
           && permissionInfoResp.getStatus().getCode()
@@ -631,8 +639,9 @@ private TPermissionInfoResp checkPrivilegeFromConfigNode(TCheckUserPrivilegesReq
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
       // Send request to some API server
       permissionInfoResp = configNodeClient.checkUserPrivileges(req);
+      resetConfigNodeConnectionErrorLogTime();
     } catch (ClientManagerException | TException e) {
-      LOGGER.error(CONNECTERROR);
+      logConfigNodeConnectionError();
       permissionInfoResp = new TPermissionInfoResp();
       permissionInfoResp.setStatus(
           RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, CONNECTERROR));
@@ -662,8 +671,9 @@ private boolean checkRoleFromConfigNode(String username, String rolename) {
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
       // Send request to some API server
       permissionInfoResp = configNodeClient.checkRoleOfUser(req);
+      resetConfigNodeConnectionErrorLogTime();
     } catch (ClientManagerException | TException e) {
-      LOGGER.error(CONNECTERROR);
+      logConfigNodeConnectionError();
       permissionInfoResp = new TPermissionInfoResp();
       permissionInfoResp.setStatus(
           RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, CONNECTERROR));
@@ -717,6 +727,19 @@ public User cacheUser(TPermissionInfoResp tPermissionInfoResp) {
     return user;
   }
 
+  static void logConfigNodeConnectionError() {
+    long now = System.currentTimeMillis();
+    long lastLogTime = LAST_CONNECT_ERROR_LOG_TIME.get();
+    if ((lastLogTime == 0 || now - lastLogTime >= CONNECT_ERROR_LOG_INTERVAL_MS)
+        && LAST_CONNECT_ERROR_LOG_TIME.compareAndSet(lastLogTime, now)) {
+      LOGGER.error(CONNECTERROR);
+    }
+  }
+
+  static void resetConfigNodeConnectionErrorLogTime() {
+    LAST_CONNECT_ERROR_LOG_TIME.set(0L);
+  }
+
   /** Cache role. */
   public Role cacheRole(String roleName, TPermissionInfoResp tPermissionInfoResp) {
     TRoleResp resp = tPermissionInfoResp.getRoleInfo().get(roleName);