This is an automated email from the ASF dual-hosted git repository. rong pushed a commit to branch leader-transfer-trigger-info in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 810676a2463941eaade78c27a81ad564913e00de Author: Steve Yurong Su <[email protected]> AuthorDate: Tue Mar 26 11:50:14 2024 +0800 Log: add logs for notifyLeaderReady, notifyLeaderChanged and procedure workers --- .../statemachine/ConfigRegionStateMachine.java | 13 +++++++++++- .../confignode/procedure/ProcedureExecutor.java | 9 +++++++-- .../schemaregion/SchemaRegionStateMachine.java | 23 ++++++++++++++++++++++ 3 files changed, 42 insertions(+), 3 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java index 375628100e9..76e88207b8f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java @@ -232,13 +232,19 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev // Shutdown leader related service for config pipe PipeConfigNodeAgent.runtime().notifyLeaderUnavailable(); + + LOGGER.info( + "Current node [nodeId:{}, ip:port: {}] is not longer the leader, " + + "all services on old leader are unavailable now.", + currentNodeId, + currentNodeTEndPoint); } } @Override public void notifyLeaderReady() { LOGGER.info( - "Current node [nodeId: {}, ip:port: {}] becomes Leader and is ready to work", + "Current node [nodeId: {}, ip:port: {}] becomes config region leader", ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(), currentNodeTEndPoint); @@ -277,6 +283,11 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev // To adapt old version, we check cluster ID after state machine has been fully recovered. // Do check async because sync will be slow and block every other things. threadPool.submit(() -> configManager.getClusterManager().checkClusterId()); + + LOGGER.info( + "Current node [nodeId: {}, ip:port: {}] as config region leader is ready to work", + ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(), + currentNodeTEndPoint); } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java index 0cd0b2ee422..6719324933d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java @@ -279,6 +279,7 @@ public class ProcedureExecutor<Env> { for (WorkerThread workerThread : workerThreads) { workerThread.start(); } + LOG.info("{} procedure workers are started.", workerThreads.size()); } public void startCompletedCleaner(long cleanTimeInterval, long cleanEvictTTL) { @@ -749,10 +750,14 @@ public class ProcedureExecutor<Env> { } catch (Throwable throwable) { if (this.activeProcedure.get() != null) { - LOG.warn("Worker terminated {}", this.activeProcedure.get(), throwable); + LOG.warn( + "Procedure Worker {} terminated {}", + getName(), + this.activeProcedure.get(), + throwable); } } finally { - LOG.debug("Worker terminated."); + LOG.info("Procedure worker {} terminated.", getName()); } workerThreads.remove(this); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java index 1a70bb8ad5d..87404225e3f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/schemaregion/SchemaRegionStateMachine.java @@ -65,15 +65,38 @@ public class SchemaRegionStateMachine extends BaseStateMachine { public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) { if (schemaRegion.getSchemaRegionId().equals(groupId) && newLeaderId != IoTDBDescriptor.getInstance().getConfig().getDataNodeId()) { + logger.info( + "Current node [nodeId: {}] is not longer the schema region leader [regionId: {}], " + + "the new leader is [nodeId:{}]", + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), + schemaRegion.getSchemaRegionId(), + newLeaderId); + // Shutdown leader related service for schema pipe PipeAgent.runtime().notifySchemaLeaderUnavailable(schemaRegion.getSchemaRegionId()); + + logger.info( + "Current node [nodeId: {}] is not longer the schema region leader [regionId: {}], " + + "all services on old leader are unavailable now.", + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), + schemaRegion.getSchemaRegionId()); } } @Override public void notifyLeaderReady() { + logger.info( + "Current node [nodeId: {}] becomes schema region leader [regionId: {}]", + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), + schemaRegion.getSchemaRegionId()); + // Activate leader related service for schema pipe PipeAgent.runtime().notifySchemaLeaderReady(schemaRegion.getSchemaRegionId()); + + logger.info( + "Current node [nodeId: {}] as schema region leader [regionId: {}] is ready to work", + IoTDBDescriptor.getInstance().getConfig().getDataNodeId(), + schemaRegion.getSchemaRegionId()); } @Override
