Improve ConfigNode removing apache#15189

Author: liyuheng55555

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java

@@ -114,7 +114,7 @@ public class ConfigNodeConfig {
   private int dataRegionPerDataNode = 0;
 
   /** each dataNode automatically has the number of CPU cores / 2 regions. */
-  private double dataRegionPerDataNodeProportion = 0.5;
+  private final double dataRegionPerDataNodeProportion = 0.5;
 
   /** RegionGroup allocate policy. */
   private RegionBalancer.RegionGroupAllocatePolicy regionGroupAllocatePolicy =
@@ -193,9 +193,6 @@ public class ConfigNodeConfig {
   /** Acceptable pause duration for Phi accrual failure detector */
   private long failureDetectorPhiAcceptablePauseInMs = 10000;
 
-  /** The unknown DataNode detect interval in milliseconds. */
-  private long unknownDataNodeDetectInterval = heartbeatIntervalInMs;
-
   /** The policy of cluster RegionGroups' leader distribution. */
   private String leaderDistributionPolicy = AbstractLeaderBalancer.CFD_POLICY;
 
@@ -660,14 +657,6 @@ public void setHeartbeatIntervalInMs(long heartbeatIntervalInMs) {
     this.heartbeatIntervalInMs = heartbeatIntervalInMs;
   }
 
-  public long getUnknownDataNodeDetectInterval() {
-    return unknownDataNodeDetectInterval;
-  }
-
-  public void setUnknownDataNodeDetectInterval(long unknownDataNodeDetectInterval) {
-    this.unknownDataNodeDetectInterval = unknownDataNodeDetectInterval;
-  }
-
   public String getLeaderDistributionPolicy() {
     return leaderDistributionPolicy;
   }

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

@@ -547,7 +547,7 @@ public TSStatus reportDataNodeShutdown(TDataNodeLocation dataNodeLocation) {
               dataNodeLocation.getDataNodeId(),
               new NodeHeartbeatSample(NodeStatus.Unknown));
       LOGGER.info(
-          "[ShutdownHook] The DataNode-{} will be shutdown soon, mark it as Unknown",
+          "The DataNode-{} will be shutdown soon, mark it as Unknown",
           dataNodeLocation.getDataNodeId());
     }
     return status;
@@ -1194,6 +1194,11 @@ private void printNewCreatedDataPartition(
   }
 
   protected TSStatus confirmLeader() {
+    if (NodeStatus.Removing == CommonDescriptor.getInstance().getConfig().getNodeStatus()) {
+      TSStatus status = new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
+      status.setMessage("ConfigNode is Removing");
+      return status;
+    }
     // Make sure the consensus layer has been initialized
     if (getConsensusManager() == null) {
       return new TSStatus(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode())
@@ -1511,7 +1516,7 @@ public TSStatus reportConfigNodeShutdown(TConfigNodeLocation configNodeLocation)
               configNodeLocation.getConfigNodeId(),
               new NodeHeartbeatSample(NodeStatus.Unknown));
       LOGGER.info(
-          "[ShutdownHook] The ConfigNode-{} will be shutdown soon, mark it as Unknown",
+          "The ConfigNode-{} will be shutdown soon, mark it as Unknown",
           configNodeLocation.getConfigNodeId());
     }
     return status;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java

@@ -37,7 +37,7 @@ public class StatisticsService {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(StatisticsService.class);
 
-  private static final long HEARTBEAT_INTERVAL =
+  public static final long STATISTICS_UPDATE_INTERVAL =
       ConfigNodeDescriptor.getInstance().getConf().getHeartbeatIntervalInMs();
 
   private final LoadCache loadCache;
@@ -63,7 +63,7 @@ public void startLoadStatisticsService() {
                 loadStatisticsExecutor,
                 this::updateLoadStatistics,
                 0,
-                HEARTBEAT_INTERVAL,
+                STATISTICS_UPDATE_INTERVAL,
                 TimeUnit.MILLISECONDS);
         LOGGER.info("LoadStatistics service is started successfully.");
       }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java

@@ -69,6 +69,7 @@
 import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
 import org.apache.iotdb.confignode.manager.load.LoadManager;
 import org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache;
+import org.apache.iotdb.confignode.manager.load.service.StatisticsService;
 import org.apache.iotdb.confignode.manager.partition.PartitionManager;
 import org.apache.iotdb.confignode.manager.partition.PartitionMetrics;
 import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
@@ -110,6 +111,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
@@ -801,13 +803,34 @@ public void applyConfigNode(
   public TSStatus checkConfigNodeBeforeRemove(RemoveConfigNodePlan removeConfigNodePlan) {
     removeConfigNodeLock.lock();
     try {
-      // Check OnlineConfigNodes number
-      if (filterConfigNodeThroughStatus(NodeStatus.Running).size() <= 1) {
+      // Check ConfigNodes number
+      if (getRegisteredConfigNodes().size() <= 1) {
         return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
             .setMessage(
                 "Remove ConfigNode failed because there is only one ConfigNode in current Cluster.");
       }
 
+      // Check OnlineConfigNodes number
+      final long deadline =
+          System.nanoTime()
+              + TimeUnit.MILLISECONDS.toNanos(
+                  (CONF.getHeartbeatIntervalInMs() + StatisticsService.STATISTICS_UPDATE_INTERVAL)
+                      * 3);
+      while (filterConfigNodeThroughStatus(NodeStatus.Running).size() <= 1) {
+        if (System.nanoTime() > deadline) {
+          return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
+              .setMessage(
+                  "Remove ConfigNode failed because there is no other ConfigNode in Running status in current Cluster.");
+        }
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
+              .setMessage("Remove ConfigNode failed due to thread interruption.");
+        }
+      }
+
       // Check whether the registeredConfigNodes contain the ConfigNode to be removed.
       if (!getRegisteredConfigNodes().contains(removeConfigNodePlan.getConfigNodeLocation())) {
         return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java

@@ -24,6 +24,8 @@
 import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
 import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.conf.SystemPropertiesUtils;
@@ -446,6 +448,11 @@ public TSStatus applyConfigNode(ApplyConfigNodePlan applyConfigNodePlan) {
    * @return {@link TSStatusCode#REMOVE_CONFIGNODE_ERROR} if remove online ConfigNode failed.
    */
   public TSStatus removeConfigNode(RemoveConfigNodePlan removeConfigNodePlan) {
+    if (removeConfigNodePlan.getConfigNodeLocation().getConfigNodeId()
+        == ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId()) {
+      // set myself to Removing status
+      CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.Removing);
+    }
     TSStatus status = new TSStatus();
     configNodeInfoReadWriteLock.writeLock().lock();
     versionInfoReadWriteLock.writeLock().lock();

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java

@@ -859,7 +859,10 @@ public TSStatus removeConfigNode(TConfigNodeLocation configNodeLocation) throws
     RemoveConfigNodePlan removeConfigNodePlan = new RemoveConfigNodePlan(configNodeLocation);
     TSStatus status = configManager.removeConfigNode(removeConfigNodePlan);
     // Print log to record the ConfigNode that performs the RemoveConfigNodeRequest
-    LOGGER.info("Execute RemoveConfigNodeRequest {} with result {}", configNodeLocation, status);
+    LOGGER.info(
+        "The result of submitting RemoveConfigNode job is {}. RemoveConfigNodeRequest: {}",
+        status,
+        configNodeLocation);
 
     return status;
   }