[ISSUE #7300] jRaft-Controller Implemention

WORKSPACE

@@ -107,6 +107,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

@@ -21,10 +21,11 @@
 import org.apache.rocketmq.common.metrics.MetricsExporterType;
 
 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 +46,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 = System.getProperty("user.home") + File.separator + controllerType + "Controller";
 
     /**
      * Whether the controller can elect a master which is not in the syncStateSet.
@@ -67,6 +68,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 = "";
@@ -201,6 +209,10 @@ public String getDLedgerAddress() {
             .map(x -> x.split("-")[1]).findFirst().get();
     }
 
+    public String getJRaftAddress() {
+        return jRaftServerId;
+    }
+
     public MetricsExporterType getMetricsExporterType() {
         return metricsExporterType;
     }
@@ -280,4 +292,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

@@ -51,6 +51,8 @@ java_library(
         "@maven//:io_opentelemetry_opentelemetry_exporter_logging",
         "@maven//:io_opentelemetry_opentelemetry_exporter_logging_otlp",
         "@maven//:org_slf4j_jul_to_slf4j",
+        "@maven//:com_alipay_sofa_jraft_core",
+        "@maven//:com_alipay_sofa_hessian",
     ],
 )
 
@@ -69,7 +71,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/ControllerManager.java

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.controller;
 
+import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
@@ -26,17 +27,16 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
-
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.common.ControllerConfig;
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.common.constant.LoggerName;
-
 import org.apache.rocketmq.common.utils.ThreadUtils;
 import org.apache.rocketmq.controller.elect.impl.DefaultElectPolicy;
 import org.apache.rocketmq.controller.impl.DLedgerController;
-import org.apache.rocketmq.controller.impl.heartbeat.DefaultBrokerHeartbeatManager;
+import org.apache.rocketmq.controller.impl.JRaftController;
+import org.apache.rocketmq.controller.impl.heartbeat.RaftBrokerHeartBeatManager;
 import org.apache.rocketmq.controller.metrics.ControllerMetricsManager;
 import org.apache.rocketmq.controller.processor.ControllerRequestProcessor;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
@@ -68,12 +68,10 @@ public class ControllerManager {
     private final Configuration configuration;
     private final RemotingClient remotingClient;
     private Controller controller;
-    private BrokerHeartbeatManager heartbeatManager;
+    private final BrokerHeartbeatManager heartbeatManager;
     private ExecutorService controllerRequestExecutor;
     private BlockingQueue<Runnable> controllerRequestThreadPoolQueue;
-
-    private NotifyService notifyService;
-
+    private final NotifyService notifyService;
     private ControllerMetricsManager controllerMetricsManager;
 
     public ControllerManager(ControllerConfig controllerConfig, NettyServerConfig nettyServerConfig,
@@ -85,7 +83,7 @@ public ControllerManager(ControllerConfig controllerConfig, NettyServerConfig ne
         this.configuration = new Configuration(log, this.controllerConfig, this.nettyServerConfig);
         this.configuration.setStorePathFromConfig(this.controllerConfig, "configStorePath");
         this.remotingClient = new NettyRemotingClient(nettyClientConfig);
-        this.heartbeatManager = new DefaultBrokerHeartbeatManager(this.controllerConfig);
+        this.heartbeatManager = BrokerHeartbeatManager.newBrokerHeartbeatManager(controllerConfig);
         this.notifyService = new NotifyService();
     }
 
@@ -100,15 +98,31 @@ public boolean initialize() {
             new ThreadFactoryImpl("ControllerRequestExecutorThread_"));
 
         this.notifyService.initialize();
-        if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerPeers())) {
-            throw new IllegalArgumentException("Attribute value controllerDLegerPeers of ControllerConfig is null or empty");
-        }
-        if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerSelfId())) {
-            throw new IllegalArgumentException("Attribute value controllerDLegerSelfId of ControllerConfig is null or empty");
+
+        if (controllerConfig.getControllerType().equals(ControllerConfig.JRAFT_CONTROLLER)) {
+            if (StringUtils.isEmpty(this.controllerConfig.getjRaftInitConf())) {
+                throw new IllegalArgumentException("Attribute value jRaftInitConf of ControllerConfig is null or empty");
+            }
+            if (StringUtils.isEmpty(this.controllerConfig.getjRaftServerId())) {
+                throw new IllegalArgumentException("Attribute value jRaftServerId of ControllerConfig is null or empty");
+            }
+            try {
+                this.controller = new JRaftController(controllerConfig, this.brokerHousekeepingService);
+                ((RaftBrokerHeartBeatManager) this.heartbeatManager).setController((JRaftController) this.controller);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        } else {
+            if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerPeers())) {
+                throw new IllegalArgumentException("Attribute value controllerDLegerPeers of ControllerConfig is null or empty");
+            }
+            if (StringUtils.isEmpty(this.controllerConfig.getControllerDLegerSelfId())) {
+                throw new IllegalArgumentException("Attribute value controllerDLegerSelfId of ControllerConfig is null or empty");
+            }
+            this.controller = new DLedgerController(this.controllerConfig, this.heartbeatManager::isBrokerActive,
+                this.nettyServerConfig, this.nettyClientConfig, this.brokerHousekeepingService,
+                new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo));
         }
-        this.controller = new DLedgerController(this.controllerConfig, this.heartbeatManager::isBrokerActive,
-            this.nettyServerConfig, this.nettyClientConfig, this.brokerHousekeepingService,
-            new DefaultElectPolicy(this.heartbeatManager::isBrokerActive, this.heartbeatManager::getBrokerLiveInfo));
 
         // Initialize the basic resources
         this.heartbeatManager.initialize();
@@ -126,10 +140,12 @@ public boolean initialize() {
      * something else.
      *
      * @param clusterName The cluster name of this inactive broker
-     * @param brokerName The inactive broker name
-     * @param brokerId The inactive broker id, null means that the election forced to be triggered
+     * @param brokerName  The inactive broker name
+     * @param brokerId    The inactive broker id, null means that the election forced to be triggered
      */
     private void onBrokerInactive(String clusterName, String brokerName, Long brokerId) {
+        log.info("Controller Manager received broker inactive event, clusterName: {}, brokerName: {}, brokerId: {}",
+            clusterName, brokerName, brokerId);
         if (controller.isLeaderState()) {
             if (brokerId == null) {
                 // Means that force triggering election for this broker-set
@@ -188,20 +204,21 @@ public void notifyBrokerRoleChanged(final RoleChangeNotifyEntry entry) {
             // Inform all active brokers
             final Map<Long, String> brokerAddrs = memberGroup.getBrokerAddrs();
             brokerAddrs.entrySet().stream().filter(x -> this.heartbeatManager.isBrokerActive(clusterName, brokerName, x.getKey()))
-                    .forEach(x -> this.notifyService.notifyBroker(x.getValue(), entry));
+                .forEach(x -> this.notifyService.notifyBroker(x.getValue(), entry));
         }
     }
 
     /**
      * Notify broker that there are roles-changing in controller
+     *
      * @param brokerAddr target broker's address to notify
-     * @param entry role change entry
+     * @param entry      role change entry
      */
     public void doNotifyBrokerRoleChanged(final String brokerAddr, final RoleChangeNotifyEntry entry) {
         if (StringUtils.isNoneEmpty(brokerAddr)) {
             log.info("Try notify broker {} that role changed, RoleChangeNotifyEntry:{}", brokerAddr, entry);
             final NotifyBrokerRoleChangedRequestHeader requestHeader = new NotifyBrokerRoleChangedRequestHeader(entry.getMasterAddress(), entry.getMasterBrokerId(),
-                    entry.getMasterEpoch(), entry.getSyncStateSetEpoch());
+                entry.getMasterEpoch(), entry.getSyncStateSetEpoch());
             final RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.NOTIFY_BROKER_ROLE_CHANGED, requestHeader);
             request.setBody(new SyncStateSet(entry.getSyncStateSet(), entry.getSyncStateSetEpoch()).encode());
             try {
@@ -214,7 +231,7 @@ public void doNotifyBrokerRoleChanged(final String brokerAddr, final RoleChangeN
 
     public void registerProcessor() {
         final ControllerRequestProcessor controllerRequestProcessor = new ControllerRequestProcessor(this);
-        final RemotingServer controllerRemotingServer = this.controller.getRemotingServer();
+        RemotingServer controllerRemotingServer = this.controller.getRemotingServer();
         assert controllerRemotingServer != null;
         controllerRemotingServer.registerProcessor(RequestCode.CONTROLLER_ALTER_SYNC_STATE_SET, controllerRequestProcessor, this.controllerRequestExecutor);
         controllerRemotingServer.registerProcessor(RequestCode.CONTROLLER_ELECT_MASTER, controllerRequestProcessor, this.controllerRequestExecutor);
@@ -231,8 +248,8 @@ public void registerProcessor() {
     }
 
     public void start() {
-        this.heartbeatManager.start();
         this.controller.startup();
+        this.heartbeatManager.start();
         this.remotingClient.start();
     }
 
@@ -335,7 +352,9 @@ public int hashCode() {
 
             @Override
             public boolean equals(Object obj) {
-                if (this == obj) return true;
+                if (this == obj) {
+                    return true;
+                }
                 if (!(obj instanceof NotifyTask)) {
                     return false;
                 }