[ISSUE #7300] jRaft-Controller Implemention

WORKSPACE

@@ -106,6 +106,8 @@ maven_install(
         "com.adobe.testing:s3mock-junit4:2.11.0",
         "io.github.aliyunmq:rocketmq-grpc-netty-codec-haproxy:1.0.0",
         "io.github.aliyunmq:rocketmq-rocksdb:1.0.3",
+        "com.alipay.sofa:jraft-core:1.3.11",
+        "com.alipay.sofa:hessian:3.3.6",
     ],
     fetch_sources = True,
     repositories = [

common/src/main/java/org/apache/rocketmq/common/ControllerConfig.java

@@ -24,7 +24,9 @@ public class ControllerConfig {
 
     private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
     private String configStorePath = System.getProperty("user.home") + File.separator + "controller" + File.separator + "controller.properties";
-
+    public static final String DLEDGER_CONTROLLER = "DLedger";
+    public static final String JRAFT_CONTROLLER = "jRaft";
+    private String controllerType = DLEDGER_CONTROLLER;
     /**
      * Interval of periodic scanning for non-active broker;
      * Unit: millisecond
@@ -45,7 +47,7 @@ public class ControllerConfig {
     private String controllerDLegerPeers;
     private String controllerDLegerSelfId;
     private int mappedFileSize = 1024 * 1024 * 1024;
-    private String controllerStorePath = System.getProperty("user.home") + File.separator + "DledgerController";
+    private String controllerStorePath = "";
 
     /**
      * Whether the controller can elect a master which is not in the syncStateSet.
@@ -67,6 +69,13 @@ public class ControllerConfig {
      */
     private long scanInactiveMasterInterval = 5 * 1000;
 
+    private int jRaftElectionTimeoutMs = 1000;
+    private int jRaftSnapshotIntervalSecs = 3600;
+    private String jRaftGroupId = "jRaft-Controller";
+    private String jRaftServerId = "localhost:9880";
+    private String jRaftInitConf = "localhost:9880,localhost:9881,localhost:9882";
+    private String jRaftControllerRPCAddr = "localhost:9770,localhost:9771,localhost:9772";
+
     private MetricsExporterType metricsExporterType = MetricsExporterType.DISABLE;
 
     private String metricsGrpcExporterTarget = "";
@@ -156,6 +165,9 @@ public void setMappedFileSize(int mappedFileSize) {
     }
 
     public String getControllerStorePath() {
+        if (controllerStorePath.isEmpty()) {
+            return System.getProperty("user.home") + File.separator + controllerType + "Controller";
+        }
         return controllerStorePath;
     }
 
@@ -201,6 +213,10 @@ public String getDLedgerAddress() {
             .map(x -> x.split("-")[1]).findFirst().get();
     }
 
+    public String getJRaftAddress() {
+        return jRaftServerId;
+    }
+
     public MetricsExporterType getMetricsExporterType() {
         return metricsExporterType;
     }
@@ -280,4 +296,60 @@ public boolean isMetricsInDelta() {
     public void setMetricsInDelta(boolean metricsInDelta) {
         this.metricsInDelta = metricsInDelta;
     }
+
+    public String getControllerType() {
+        return controllerType;
+    }
+
+    public void setControllerType(String controllerType) {
+        this.controllerType = controllerType;
+    }
+
+    public int getjRaftElectionTimeoutMs() {
+        return jRaftElectionTimeoutMs;
+    }
+
+    public void setjRaftElectionTimeoutMs(int jRaftElectionTimeoutMs) {
+        this.jRaftElectionTimeoutMs = jRaftElectionTimeoutMs;
+    }
+
+    public int getjRaftSnapshotIntervalSecs() {
+        return jRaftSnapshotIntervalSecs;
+    }
+
+    public void setjRaftSnapshotIntervalSecs(int jRaftSnapshotIntervalSecs) {
+        this.jRaftSnapshotIntervalSecs = jRaftSnapshotIntervalSecs;
+    }
+
+    public String getjRaftGroupId() {
+        return jRaftGroupId;
+    }
+
+    public void setjRaftGroupId(String jRaftGroupId) {
+        this.jRaftGroupId = jRaftGroupId;
+    }
+
+    public String getjRaftServerId() {
+        return jRaftServerId;
+    }
+
+    public void setjRaftServerId(String jRaftServerId) {
+        this.jRaftServerId = jRaftServerId;
+    }
+
+    public String getjRaftInitConf() {
+        return jRaftInitConf;
+    }
+
+    public void setjRaftInitConf(String jRaftInitConf) {
+        this.jRaftInitConf = jRaftInitConf;
+    }
+
+    public String getjRaftControllerRPCAddr() {
+        return jRaftControllerRPCAddr;
+    }
+
+    public void setjRaftControllerRPCAddr(String jRaftControllerRPCAddr) {
+        this.jRaftControllerRPCAddr = jRaftControllerRPCAddr;
+    }
 }

common/src/main/java/org/apache/rocketmq/common/Pair.java

@@ -16,7 +16,9 @@
  */
 package org.apache.rocketmq.common;
 
-public class Pair<T1, T2> {
+import java.io.Serializable;
+
+public class Pair<T1, T2> implements Serializable {
     private T1 object1;
     private T2 object2;
 

controller/BUILD.bazel

@@ -50,6 +50,8 @@ java_library(
         "@maven//:io_opentelemetry_opentelemetry_sdk_metrics",
         "@maven//:io_opentelemetry_opentelemetry_exporter_logging",
         "@maven//:org_slf4j_jul_to_slf4j",
+        "@maven//:com_alipay_sofa_jraft_core",
+        "@maven//:com_alipay_sofa_hessian",
     ],
 )
 
@@ -68,7 +70,7 @@ java_library(
         "@maven//:org_apache_commons_commons_lang3", 
         "@maven//:io_netty_netty_all",    
         "@maven//:com_google_guava_guava",  
-        "@maven//:com_alibaba_fastjson",         
+        "@maven//:com_alibaba_fastjson",     
     ],
     resources = glob(["src/test/resources/certs/*.pem"]) + glob(["src/test/resources/certs/*.key"])
 )

controller/pom.xml

@@ -15,7 +15,8 @@
   limitations under the License.
   -->
 
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>rocketmq-all</artifactId>
         <groupId>org.apache.rocketmq</groupId>
@@ -62,5 +63,24 @@
             <groupId>org.slf4j</groupId>
             <artifactId>jul-to-slf4j</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.alipay.sofa</groupId>
+            <artifactId>jraft-core</artifactId>
+            <version>1.3.11</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.ow2.asm</groupId>
+                    <artifactId>asm</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.google.protobuf</groupId>
+                    <artifactId>protobuf-java</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java-util</artifactId>
+        </dependency>
     </dependencies>
 </project>

controller/src/main/java/org/apache/rocketmq/controller/BrokerHeartbeatManager.java

@@ -17,17 +17,32 @@
 package org.apache.rocketmq.controller;
 
 import io.netty.channel.Channel;
-import java.util.Map;
+import org.apache.rocketmq.common.ControllerConfig;
 import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
 import org.apache.rocketmq.controller.impl.heartbeat.BrokerLiveInfo;
+import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
+import org.apache.rocketmq.controller.impl.heartbeat.RaftBrokerHeartBeatManager;
+
+import java.util.Map;
 
 public interface BrokerHeartbeatManager {
+    public static final long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 10;
+
+    public static BrokerHeartbeatManager newBrokerHeartbeatManager(ControllerConfig controllerConfig) {
+        if (controllerConfig.getControllerType().equals(ControllerConfig.JRAFT_CONTROLLER)) {
+            return new RaftBrokerHeartBeatManager(controllerConfig);
+        } else {
+            return new DefaultBrokerHeartbeatManager(controllerConfig);
+        }
+    }
 
     /**
      * initialize the resources
+     *
      * @return
      */
     void initialize();
+
     /**
      * Broker new heartbeat.
      */
@@ -67,6 +82,7 @@ void onBrokerHeartbeat(final String clusterName, final String brokerName, final
 
     /**
      * Count the number of active brokers in each broker-set of each cluster
+     *
      * @return active brokers count
      */
     Map<String/*cluster*/, Map<String/*broker-set*/, Integer/*active broker num*/>> getActiveBrokersNum();

controller/src/main/java/org/apache/rocketmq/controller/Controller.java

@@ -17,21 +17,21 @@
 
 package org.apache.rocketmq.controller;
 
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
 import org.apache.rocketmq.controller.helper.BrokerLifecycleListener;
 import org.apache.rocketmq.remoting.RemotingServer;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.body.SyncStateSet;
 import org.apache.rocketmq.remoting.protocol.header.controller.AlterSyncStateSetRequestHeader;
-import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.GetReplicaInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.controller.admin.CleanControllerBrokerDataRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.register.ApplyBrokerIdRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.register.GetNextBrokerIdRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.controller.register.RegisterBrokerToControllerRequestHeader;
 
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
 /**
  * The api for controller
  */
@@ -113,6 +113,7 @@ CompletableFuture<RemotingCommand> alterSyncStateSet(
 
     /**
      * Add broker's lifecycle listener
+     *
      * @param listener listener
      */
     void registerBrokerLifecycleListener(final BrokerLifecycleListener listener);
@@ -124,7 +125,6 @@ CompletableFuture<RemotingCommand> alterSyncStateSet(
 
     /**
      * Clean controller broker data
-     *
      */
     CompletableFuture<RemotingCommand> cleanBrokerData(final CleanControllerBrokerDataRequestHeader requestHeader);
 }