New features of cluster scalability and multi-raft

cluster/src/assembly/resources/conf/iotdb-cluster.properties

@@ -72,6 +72,9 @@ seed_nodes=127.0.0.1:9003
 # number of replications for one partition
 default_replica_num=1
 
+# sub raft num for multi-raft
+multi_raft_factor=1
+
 # cluster name to identify different clusters
 # all node's cluster_name in one cluster are the same
 # cluster_name=default
@@ -169,6 +172,10 @@ default_replica_num=1
 # This default value is 1000
 # max_read_log_lag=1000
 
+# When a follower tries to sync log with the leader, sync will fail if the log Lag exceeds max_sync_log_lag.
+# This default value is 100000
+# max_sync_log_lag=100000
+
 # Max number of clients in a ClientPool of a member for one node.
 # max_client_pernode_permember_number=1000
 

cluster/src/assembly/resources/sbin/add-node.bat

@@ -19,7 +19,7 @@
 
 @echo off
 echo ````````````````````````
-echo Starting IoTDB
+echo Starting IoTDB (Cluster Mode)
 echo ````````````````````````
 
 PATH %PATH%;%JAVA_HOME%\bin\

cluster/src/assembly/resources/sbin/add-node.sh

@@ -20,7 +20,7 @@
 
 
 echo ---------------------
-echo Starting IoTDB
+echo "Starting IoTDB (Cluster Mode)"
 echo ---------------------
 
 if [ -z "${IOTDB_HOME}" ]; then

cluster/src/assembly/resources/sbin/remove-node.bat

@@ -0,0 +1,117 @@
+@REM
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM     http://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM
+
+@echo off
+echo ````````````````````````
+echo Starting to remove a node (Cluster Mode)
+echo ````````````````````````
+
+PATH %PATH%;%JAVA_HOME%\bin\
+set "FULL_VERSION="
+set "MAJOR_VERSION="
+set "MINOR_VERSION="
+
+
+for /f tokens^=2-5^ delims^=.-_+^" %%j in ('java -fullversion 2^>^&1') do (
+	set "FULL_VERSION=%%j-%%k-%%l-%%m"
+	IF "%%j" == "1" (
+	    set "MAJOR_VERSION=%%k"
+	    set "MINOR_VERSION=%%l"
+	) else (
+	    set "MAJOR_VERSION=%%j"
+	    set "MINOR_VERSION=%%k"
+	)
+)
+
+set JAVA_VERSION=%MAJOR_VERSION%
+
+IF NOT %JAVA_VERSION% == 8 (
+	IF NOT %JAVA_VERSION% == 11 (
+		echo IoTDB only supports jdk8 or jdk11, please check your java version.
+		goto finally
+	)
+)
+
+if "%OS%" == "Windows_NT" setlocal
+
+pushd %~dp0..
+if NOT DEFINED IOTDB_HOME set IOTDB_HOME=%cd%
+popd
+
+set IOTDB_CONF=%IOTDB_HOME%\conf
+set IOTDB_LOGS=%IOTDB_HOME%\logs
+
+@setlocal ENABLEDELAYEDEXPANSION ENABLEEXTENSIONS
+set CONF_PARAMS=-r
+set is_conf_path=false
+for %%i in (%*) do (
+	IF "%%i" == "-c" (
+		set is_conf_path=true
+	) ELSE IF "!is_conf_path!" == "true" (
+		set is_conf_path=false
+		set IOTDB_CONF=%%i
+	) ELSE (
+		set CONF_PARAMS=!CONF_PARAMS! %%i
+	)
+)
+
+if NOT DEFINED MAIN_CLASS set MAIN_CLASS=org.apache.iotdb.cluster.ClusterMain
+if NOT DEFINED JAVA_HOME goto :err
+
+@REM -----------------------------------------------------------------------------
+@REM JVM Opts we'll use in legacy run or installation
+set JAVA_OPTS=-ea^
+ -Dlogback.configurationFile="%IOTDB_CONF%\logback.xml"^
+ -DIOTDB_HOME="%IOTDB_HOME%"^
+ -DTSFILE_HOME="%IOTDB_HOME%"^
+ -DCLUSTER_CONF="%IOTDB_CONF%"^
+ -DIOTDB_CONF="%IOTDB_CONF%"
+
+@REM ***** CLASSPATH library setting *****
+@REM Ensure that any user defined CLASSPATH variables are not used on startup
+set CLASSPATH="%IOTDB_HOME%\lib"
+
+@REM For each jar in the IOTDB_HOME lib directory call append to build the CLASSPATH variable.
+set CLASSPATH=%CLASSPATH%;"%IOTDB_HOME%\lib\*"
+set CLASSPATH=%CLASSPATH%;iotdb.ClusterMain
+goto okClasspath
+
+:append
+set CLASSPATH=%CLASSPATH%;%1
+goto :eof
+
+@REM -----------------------------------------------------------------------------
+:okClasspath
+
+rem echo CLASSPATH: %CLASSPATH%
+
+"%JAVA_HOME%\bin\java" %JAVA_OPTS% %IOTDB_HEAP_OPTS% -cp %CLASSPATH% %IOTDB_JMX_OPTS% %MAIN_CLASS% %CONF_PARAMS%
+goto finally
+
+:err
+echo JAVA_HOME environment variable must be set!
+pause
+
+
+@REM -----------------------------------------------------------------------------
+:finally
+
+pause
+
+ENDLOCAL

cluster/src/assembly/resources/sbin/remove-node.sh

@@ -0,0 +1,88 @@
+#!/bin/bash
+#
+# 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.
+#
+
+
+echo ---------------------
+echo "Starting to remove a node(Cluster Mode)"
+echo ---------------------
+
+if [ -z "${IOTDB_HOME}" ]; then
+  export IOTDB_HOME="`dirname "$0"`/.."
+fi
+
+IOTDB_CONF=${IOTDB_HOME}/conf
+
+is_conf_path=false
+for arg do
+    shift
+    if [ "$arg" == "-c" ]; then
+        is_conf_path=true
+        continue
+    fi
+
+    if [ $is_conf_path == true ]; then
+        IOTDB_CONF=$arg
+        is_conf_path=false
+        continue
+    fi
+    set -- "$@" "$arg"
+done
+
+CONF_PARAMS="-r "$*
+
+if [ -n "$JAVA_HOME" ]; then
+    for java in "$JAVA_HOME"/bin/amd64/java "$JAVA_HOME"/bin/java; do
+        if [ -x "$java" ]; then
+            JAVA="$java"
+            break
+        fi
+    done
+else
+    JAVA=java
+fi
+
+if [ -z $JAVA ] ; then
+    echo Unable to find java executable. Check JAVA_HOME and PATH environment variables.  > /dev/stderr
+    exit 1;
+fi
+
+CLASSPATH=""
+for f in ${IOTDB_HOME}/lib/*.jar; do
+  CLASSPATH=${CLASSPATH}":"$f
+done
+classname=org.apache.iotdb.cluster.ClusterMain
+
+launch_service()
+{
+	class="$1"
+	iotdb_parms="-Dlogback.configurationFile=${IOTDB_CONF}/logback.xml"
+	iotdb_parms="$iotdb_parms -DIOTDB_HOME=${IOTDB_HOME}"
+	iotdb_parms="$iotdb_parms -DTSFILE_HOME=${IOTDB_HOME}"
+	iotdb_parms="$iotdb_parms -DIOTDB_CONF=${IOTDB_CONF}"
+	iotdb_parms="$iotdb_parms -DCLUSTER_CONF=${IOTDB_CONF}"
+	iotdb_parms="$iotdb_parms -Dname=iotdb\.IoTDB"
+	exec "$JAVA" $iotdb_parms $IOTDB_JMX_OPTS -cp "$CLASSPATH" "$class" $CONF_PARAMS
+	return $?
+}
+
+# Start up the service
+launch_service "$classname"
+
+exit $?

cluster/src/assembly/resources/sbin/start-node.bat

@@ -19,7 +19,7 @@
 
 @echo off
 echo ````````````````````````
-echo Starting IoTDB
+echo Starting IoTDB (Cluster Mode)
 echo ````````````````````````
 
 PATH %PATH%;%JAVA_HOME%\bin\

cluster/src/main/java/org/apache/iotdb/cluster/ClusterMain.java

@@ -51,6 +51,8 @@
 import java.util.HashSet;
 import java.util.Set;
 
+import static org.apache.iotdb.cluster.utils.ClusterUtils.UNKNOWN_CLIENT_IP;
+
 public class ClusterMain {
 
   private static final Logger logger = LoggerFactory.getLogger(ClusterMain.class);
@@ -99,8 +101,8 @@ public static void main(String[] args) {
     }
 
     String mode = args[0];
-
     logger.info("Running mode {}", mode);
+
     if (MODE_START.equals(mode)) {
       try {
         metaServer = new MetaClusterServer();
@@ -121,13 +123,19 @@ public static void main(String[] args) {
       }
     } else if (MODE_ADD.equals(mode)) {
       try {
+        long startTime = System.currentTimeMillis();
         metaServer = new MetaClusterServer();
         preStartCustomize();
         metaServer.start();
         metaServer.joinCluster();
         // Currently, we do not register ClusterInfoService as a JMX Bean,
         // so we use startService() rather than start()
         ClusterInfoServer.getInstance().startService();
+
+        logger.info(
+            "Adding this node {} to cluster costs {} ms",
+            metaServer.getMember().getThisNode(),
+            (System.currentTimeMillis() - startTime));
       } catch (TTransportException
           | StartupException
           | QueryProcessException
@@ -221,7 +229,7 @@ private static void doRemoveNode(String[] args) throws IOException {
     TProtocolFactory factory =
         config.isRpcThriftCompressionEnabled() ? new TCompactProtocol.Factory() : new Factory();
     Node nodeToRemove = new Node();
-    nodeToRemove.setInternalIp(ip).setMetaPort(metaPort);
+    nodeToRemove.setInternalIp(ip).setMetaPort(metaPort).setClientIp(UNKNOWN_CLIENT_IP);
     // try sending the request to each seed node
     for (String url : config.getSeedNodeUrls()) {
       Node node = ClusterUtils.parseNode(url);
@@ -230,6 +238,7 @@ private static void doRemoveNode(String[] args) throws IOException {
       }
       AsyncMetaClient client = new AsyncMetaClient(factory, new TAsyncClientManager(), node, null);
       Long response = null;
+      long startTime = System.currentTimeMillis();
       try {
         logger.info("Start removing node {} with the help of node {}", nodeToRemove, node);
         response = SyncClientAdaptor.removeNode(client, nodeToRemove);
@@ -240,19 +249,25 @@ private static void doRemoveNode(String[] args) throws IOException {
         logger.warn("Cannot send remove node request through {}, try next node", node);
       }
       if (response != null) {
-        handleNodeRemovalResp(response, nodeToRemove);
+        handleNodeRemovalResp(response, nodeToRemove, startTime);
         return;
       }
     }
   }
 
-  private static void handleNodeRemovalResp(Long response, Node nodeToRemove) {
+  private static void handleNodeRemovalResp(Long response, Node nodeToRemove, long startTime) {
     if (response == Response.RESPONSE_AGREE) {
-      logger.info("Node {} is successfully removed", nodeToRemove);
+      logger.info(
+          "Node {} is successfully removed, cost {}ms",
+          nodeToRemove,
+          (System.currentTimeMillis() - startTime));
     } else if (response == Response.RESPONSE_CLUSTER_TOO_SMALL) {
       logger.error("Cluster size is too small, cannot remove any node");
     } else if (response == Response.RESPONSE_REJECT) {
       logger.error("Node {} is not found in the cluster, please check", nodeToRemove);
+    } else if (response == Response.RESPONSE_DATA_MIGRATION_NOT_FINISH) {
+      logger.warn(
+          "The data migration of the previous membership change operation is not finished. Please try again later");
     } else {
       logger.error("Unexpected response {}", response);
     }

cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java

@@ -126,9 +126,7 @@ private AsyncClient waitForClient(Deque<AsyncClient> clientStack, ClusterNode cl
         this.wait(waitClientTimeutMS);
         if (clientStack.isEmpty() && System.currentTimeMillis() - waitStart >= waitClientTimeutMS) {
           logger.warn(
-              "Cannot get an available client after {}ms, create a new one.",
-              waitClientTimeutMS,
-              asyncClientFactory);
+              "Cannot get an available client after {}ms, create a new one.", waitClientTimeutMS);
           AsyncClient asyncClient = asyncClientFactory.getAsyncClient(clusterNode, this);
           nodeClientNumMap.computeIfPresent(clusterNode, (n, oldValue) -> oldValue + 1);
           return asyncClient;