Skip to content

Limit repeated warning logs #17889

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 11 commits into
base: master
Choose a base branch
from
Open

Limit repeated warning logs #17889

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
f966af0
Limit repeated warning logs
Caideyipi Jun 9, 2026
8e25bb7
Throttle disk and WAL warning logs
Caideyipi Jun 10, 2026
a1281d1
Throttle repeated disk full logs
Caideyipi Jun 10, 2026
86b2341
Fix Logback dependency analysis warnings
Caideyipi Jun 10, 2026
b0f0bd5
Throttle repeated failure logs
Caideyipi Jun 10, 2026
95a5e79
Fix datanode unit test failures
Caideyipi Jun 11, 2026
b33be53
Merge remote-tracking branch 'origin/master' into fix-log-throttling
Caideyipi Jun 23, 2026
c589639
spotless
Caideyipi Jun 24, 2026
5d8f3ef
Merge origin/master into fix-log-throttling
Caideyipi Jun 25, 2026
5084be1
Move failure log synchronization into state
Caideyipi Jul 3, 2026
903f2ff
Extract metrics failure log state
Caideyipi Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions ...ain/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -679,9 +682,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);
}
}
}
}
Expand Down
37 changes: 30 additions & 7 deletions ...rc/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionMetaSyncer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -49,6 +50,7 @@ public class SubscriptionMetaSyncer {
SubscriptionConfig.getInstance().getSubscriptionMetaSyncerSyncIntervalMinutes();

private final ConfigManager configManager;
private final LogThrottler syncFailureLogThrottler = new LogThrottler();

private Future<?> metaSyncFuture;

Expand Down Expand Up @@ -84,37 +86,53 @@ 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("Failed to sync commit progress. Result status: {}.", commitProgressSyncStatus);
if (syncFailureLogThrottler.shouldLog(
"commit-progress-sync", getStatusFailureSignature(commitProgressSyncStatus))) {
LOGGER.warn("Failed to sync commit progress. Result status: {}.", commitProgressSyncStatus);
}
return;
}

syncFailureLogThrottler.reset();
LOGGER.info(
ManagerMessages.AFTER_THIS_SUCCESSFUL_SYNC_IF_SUBSCRIPTIONINFO_IS_EMPTY_DURING_THIS);
isLastSubscriptionSyncSuccessful = true;
Expand All @@ -124,7 +142,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();
}
}
8 changes: 8 additions & 0 deletions iotdb-core/datanode/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
37 changes: 30 additions & 7 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/ClusterAuthorityFetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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));
}
Expand All @@ -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 {
Expand All @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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()));
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -629,8 +637,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));
Expand Down Expand Up @@ -660,8 +669,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));
Expand Down Expand Up @@ -715,6 +725,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);
Expand Down
Loading
Loading