Implement Cluster ID

integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterRestartIT.java

@@ -24,15 +24,13 @@
 import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.confignode.it.utils.ConfigNodeTestUtils;
 import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
-import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.EnvUtils;
 import org.apache.iotdb.it.env.cluster.config.MppBaseConfig;
 import org.apache.iotdb.it.env.cluster.config.MppCommonConfig;
 import org.apache.iotdb.it.env.cluster.config.MppJVMConfig;
 import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
-import org.apache.iotdb.it.framework.IoTDBTestLogger;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.ClusterIT;
 
@@ -43,19 +41,19 @@
 import org.junit.experimental.categories.Category;
 import org.junit.runner.RunWith;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
+
 @RunWith(IoTDBTestRunner.class)
 @Category({ClusterIT.class})
 public class IoTDBClusterRestartIT {
-  private static final Logger logger = IoTDBTestLogger.logger;
-
-  private static final String ratisConsensusProtocolClass =
-      "org.apache.iotdb.consensus.ratis.RatisConsensus";
+  private static final Logger logger = LoggerFactory.getLogger(IoTDBClusterRestartIT.class);
 
   private static final int testReplicationFactor = 2;
 
@@ -66,10 +64,9 @@ public void setUp() {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
-        .setConfigNodeConsensusProtocolClass(ratisConsensusProtocolClass)
-        .setSchemaRegionConsensusProtocolClass(ratisConsensusProtocolClass)
-        .setDataRegionConsensusProtocolClass(ratisConsensusProtocolClass)
-        .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+        .setConfigNodeConsensusProtocolClass(RATIS_CONSENSUS)
+        .setSchemaRegionConsensusProtocolClass(RATIS_CONSENSUS)
+        .setDataRegionConsensusProtocolClass(RATIS_CONSENSUS)
         .setSchemaReplicationFactor(testReplicationFactor)
         .setDataReplicationFactor(testReplicationFactor);
 

integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterStartIT.java

@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.it.cluster;
+
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.confignode.rpc.thrift.TGetClusterIdResp;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({ClusterIT.class})
+public class IoTDBClusterStartIT {
+  private static final Logger logger = LoggerFactory.getLogger(IoTDBClusterStartIT.class);
+
+  private static final int testConfigNodeNum = 3, testDataNodeNum = 0;
+
+  @Before
+  public void setUp() {
+    EnvFactory.getEnv()
+        .getConfig()
+        .getCommonConfig()
+        .setConfigNodeConsensusProtocolClass(RATIS_CONSENSUS)
+        .setSchemaRegionConsensusProtocolClass(RATIS_CONSENSUS)
+        .setDataRegionConsensusProtocolClass(RATIS_CONSENSUS);
+
+    EnvFactory.getEnv().initClusterEnvironment(testConfigNodeNum, testDataNodeNum);
+  }
+
+  @After
+  public void tearDown() {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void clusterIdTest() throws ClientManagerException, IOException, InterruptedException {
+    final long maxTestTime = TimeUnit.SECONDS.toMillis(30);
+    final long testInterval = TimeUnit.SECONDS.toMillis(1);
+    try (SyncConfigNodeIServiceClient client =
+        (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+      long startTime = System.currentTimeMillis();
+      while (System.currentTimeMillis() - startTime < maxTestTime) {
+        try {
+          TGetClusterIdResp resp = client.getClusterId();
+          if (TSStatusCode.SUCCESS_STATUS.getStatusCode() == resp.getStatus().getCode()) {
+            Assert.assertNotNull(resp.getClusterId());
+            Assert.assertNotEquals("", resp.getClusterId());
+            return;
+          }
+        } catch (TException e) {
+          logger.error("TException:", e);
+        }
+        Thread.sleep(testInterval);
+      }
+      String errorMessage = String.format("Cluster ID failed to generate in %d ms.", maxTestTime);
+      Assert.fail(errorMessage);
+    }
+  }
+}

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -154,6 +154,7 @@ public enum TSStatusCode {
   REMOVE_DATANODE_ERROR(1006),
   CAN_NOT_CONNECT_DATANODE(1007),
   TRANSFER_LEADER_ERROR(1008),
+  GET_CLUSTER_ID_ERROR(1009),
 
   // Sync, Load TsFile
   LOAD_FILE_ERROR(1100),

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IdentifierParser.g4

@@ -57,6 +57,7 @@ keyWords
     | CHILD
     | CLEAR
     | CLUSTER
+    | CLUSTERID
     | CONCAT
     | CONDITION
     | CONFIGNODES

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -59,7 +59,7 @@ ddlStatement
     // CQ
     | createContinuousQuery | dropContinuousQuery | showContinuousQueries
     // Cluster
-    | showVariables | showCluster | showRegions | showDataNodes | showConfigNodes
+    | showVariables | showCluster | showRegions | showDataNodes | showConfigNodes | showClusterId
     | getRegionId | getTimeSlotList | countTimeSlotList | getSeriesSlotList | migrateRegion
     // Quota
     | setSpaceQuota | showSpaceQuota | setThrottleQuota | showThrottleQuota
@@ -475,6 +475,11 @@ showConfigNodes
     : SHOW CONFIGNODES
     ;
 
+// ---- Show Cluster Id
+showClusterId
+    : SHOW CLUSTERID
+    ;
+
 // ---- Get Region Id
 getRegionId
     : SHOW (DATA|SCHEMA) REGIONID WHERE (DATABASE operator_eq database=prefixPath

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -141,6 +141,10 @@ CLUSTER
     : C L U S T E R
     ;
 
+CLUSTERID
+    : C L U S T E R I D
+    ;
+
 CONCAT
     : C O N C A T
     ;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -51,6 +51,7 @@
 import org.apache.iotdb.confignode.consensus.request.read.trigger.GetTriggerTablePlan;
 import org.apache.iotdb.confignode.consensus.request.write.confignode.ApplyConfigNodePlan;
 import org.apache.iotdb.confignode.consensus.request.write.confignode.RemoveConfigNodePlan;
+import org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateClusterIdPlan;
 import org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateVersionInfoPlan;
 import org.apache.iotdb.confignode.consensus.request.write.cq.ActiveCQPlan;
 import org.apache.iotdb.confignode.consensus.request.write.cq.AddCQPlan;
@@ -295,6 +296,9 @@ public static ConfigPhysicalPlan create(ByteBuffer buffer) throws IOException {
         case UpdateVersionInfo:
           plan = new UpdateVersionInfoPlan();
           break;
+        case UpdateClusterId:
+          plan = new UpdateClusterIdPlan();
+          break;
         case CreateFunction:
           plan = new CreateFunctionPlan();
           break;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -28,6 +28,7 @@ public enum ConfigPhysicalPlanType {
   ApplyConfigNode((short) 0),
   RemoveConfigNode((short) 1),
   UpdateVersionInfo((short) 2),
+  UpdateClusterId((short) 3),
 
   /** DataNode. */
   RegisterDataNode((short) 100),

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/confignode/UpdateClusterIdPlan.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.consensus.request.write.confignode;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+public class UpdateClusterIdPlan extends ConfigPhysicalPlan {
+  private String clusterId;
+
+  public UpdateClusterIdPlan() {
+    super(ConfigPhysicalPlanType.UpdateClusterId);
+  }
+
+  public UpdateClusterIdPlan(String clusterId) {
+    this();
+    this.clusterId = clusterId;
+  }
+
+  public String getClusterId() {
+    return this.clusterId;
+  }
+
+  @Override
+  protected void serializeImpl(DataOutputStream stream) throws IOException {
+    ReadWriteIOUtils.write(getType().getPlanType(), stream);
+    ReadWriteIOUtils.write(clusterId, stream);
+  }
+
+  @Override
+  protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+    clusterId = ReadWriteIOUtils.readString(buffer);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    if (!getType().equals(((UpdateClusterIdPlan) o).getType())) {
+      return false;
+    }
+    return Objects.equals(clusterId, ((UpdateClusterIdPlan) o).clusterId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(clusterId);
+  }
+}

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

@@ -222,7 +222,7 @@ public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
   @Override
   public void notifyLeaderReady() {
     LOGGER.info(
-        "Current node [nodeId: {}, ip:port: {}] becomes Leader",
+        "Current node [nodeId: {}, ip:port: {}] becomes Leader and is ready to work",
         ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
         currentNodeTEndPoint);
 
@@ -248,6 +248,10 @@ public void notifyLeaderReady() {
         () -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeMetaSync());
     threadPool.submit(
         () -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeHeartbeat());
+
+    // 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());
   }
 
   @Override

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.manager;
+
+import org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateClusterIdPlan;
+import org.apache.iotdb.confignode.persistence.ClusterInfo;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+
+public class ClusterManager {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ClusterManager.class);
+
+  private final IManager configManager;
+  private final ClusterInfo clusterInfo;
+
+  private static final String CONSENSUS_WRITE_ERROR =
+      "Failed in the write API executing the consensus layer due to: ";
+
+  public ClusterManager(IManager configManager, ClusterInfo clusterInfo) {
+    this.configManager = configManager;
+    this.clusterInfo = clusterInfo;
+  }
+
+  public void checkClusterId() {
+    if (clusterInfo.getClusterId() != null) {
+      LOGGER.info("clusterID: {}", clusterInfo.getClusterId());
+      return;
+    }
+    generateClusterId();
+  }
+
+  public String getClusterId() {
+    return clusterInfo.getClusterId();
+  }
+
+  private void generateClusterId() {
+    String clusterId = String.valueOf(UUID.randomUUID());
+    UpdateClusterIdPlan updateClusterIdPlan = new UpdateClusterIdPlan(clusterId);
+    try {
+      configManager.getConsensusManager().write(updateClusterIdPlan);
+    } catch (ConsensusException e) {
+      LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+    }
+  }
+}