upgrade jraft to 1.3.5.Alpha1

Author: dzdx

pom.xml

@@ -59,7 +59,7 @@
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <bolt.version>1.5.2</bolt.version>
+        <bolt.version>1.6.2</bolt.version>
         <commons.lang.version>2.6</commons.lang.version>
         <google.guava.version>18.0</google.guava.version>
         <sofa.common.tools.version>1.0.12</sofa.common.tools.version>
@@ -75,7 +75,7 @@
         <lookout.version>1.5.2</lookout.version>
         <mockito.version>1.10.19</mockito.version>
         <powermock.version>1.6.6</powermock.version>
-        <jraft.version>1.2.7.beta1</jraft.version>
+        <jraft.version>1.3.5.Alpha1</jraft.version>
         <metrics.version>4.0.2</metrics.version>
         <commons-io.version>2.4</commons-io.version>
         <jetty.version>[9.4.17.v20190418,9.4.19.v20190610]</jetty.version>

server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/remoting/RaftExchanger.java

@@ -38,7 +38,7 @@
 import com.alipay.sofa.jraft.core.NodeImpl;
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.option.CliOptions;
-import com.alipay.sofa.jraft.rpc.impl.AbstractBoltClientService;
+import com.alipay.sofa.jraft.rpc.impl.AbstractClientService;
 import com.alipay.sofa.registry.common.model.metaserver.MetaNode;
 import com.alipay.sofa.registry.common.model.store.URL;
 import com.alipay.sofa.registry.jraft.bootstrap.RaftClient;
@@ -233,8 +233,7 @@ public void startRaftClient() {
                 if (raftServer != null && raftServer.getNode() != null) {
                     //TODO this cannot be invoke,because RaftAnnotationBeanPostProcessor.getProxy will start first
                     raftClient = new RaftClient(getGroup(), serverConf,
-                        (AbstractBoltClientService) (((NodeImpl) raftServer.getNode())
-                            .getRpcService()));
+                        (AbstractClientService) (((NodeImpl) raftServer.getNode()).getRpcService()));
                 } else {
                     raftClient = new RaftClient(getGroup(), serverConf);
                 }

server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/remoting/connection/DataConnectionHandler.java

@@ -20,18 +20,16 @@
 import com.alipay.sofa.registry.net.NetUtil;
 import com.alipay.sofa.registry.remoting.Channel;
 import com.alipay.sofa.registry.remoting.RemotingException;
-import com.alipay.sofa.registry.server.meta.executor.ExecutorManager;
 import com.alipay.sofa.registry.server.meta.remoting.handler.AbstractServerHandler;
-import org.springframework.beans.factory.annotation.Autowired;
 
 import java.net.InetSocketAddress;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
 
 /**
  * Handle data node's connect request
+ *
  * @author shangyu.wh
  * @version $Id: DataConnectionHandler.java, v 0.1 2018-01-24 16:04 shangyu.wh Exp $
  */

server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/bootstrap/RaftClient.java

@@ -24,8 +24,10 @@
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.option.CliOptions;
 import com.alipay.sofa.jraft.rpc.CliClientService;
-import com.alipay.sofa.jraft.rpc.impl.AbstractBoltClientService;
-import com.alipay.sofa.jraft.rpc.impl.cli.BoltCliClientService;
+import com.alipay.sofa.jraft.rpc.ClientService;
+import com.alipay.sofa.jraft.rpc.impl.AbstractClientService;
+import com.alipay.sofa.jraft.rpc.impl.BoltRpcClient;
+import com.alipay.sofa.jraft.rpc.impl.cli.CliClientServiceImpl;
 import com.alipay.sofa.registry.jraft.command.ProcessRequest;
 import com.alipay.sofa.registry.jraft.command.ProcessResponse;
 import com.alipay.sofa.registry.jraft.handler.NotifyLeaderChangeHandler;
@@ -39,25 +41,24 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
- *
  * @author shangyu.wh
  * @version $Id: RaftClient.java, v 0.1 2018-05-16 11:40 shangyu.wh Exp $
  */
 public class RaftClient {
 
-    private static final Logger  LOGGER  = LoggerFactory.getLogger(RaftClient.class);
+    private static final Logger LOGGER  = LoggerFactory.getLogger(RaftClient.class);
 
-    private BoltCliClientService cliClientService;
-    private RpcClient            rpcClient;
-    private CliOptions           cliOptions;
-    private String               groupId;
-    private Configuration        conf;
+    private CliClientService    cliClientService;
+    private RpcClient           rpcClient;
+    private CliOptions          cliOptions;
+    private String              groupId;
+    private Configuration       conf;
 
-    private AtomicBoolean        started = new AtomicBoolean(false);
+    private AtomicBoolean       started = new AtomicBoolean(false);
 
     /**
      * @param groupId
-     * @param confStr  Example: 127.0.0.1:8081,127.0.0.1:8082,127.0.0.1:8083
+     * @param confStr Example: 127.0.0.1:8081,127.0.0.1:8082,127.0.0.1:8083
      */
     public RaftClient(String groupId, String confStr) {
 
@@ -67,23 +68,23 @@ public RaftClient(String groupId, String confStr) {
             throw new IllegalArgumentException("Fail to parse conf:" + confStr);
         }
         cliOptions = new CliOptions();
-        cliClientService = new BoltCliClientService();
+        cliClientService = new CliClientServiceImpl();
     }
 
     /**
      * @param groupId
      * @param confStr
      * @param cliClientService
      */
-    public RaftClient(String groupId, String confStr, AbstractBoltClientService cliClientService) {
+    public RaftClient(String groupId, String confStr, AbstractClientService cliClientService) {
 
         this.groupId = groupId;
         conf = new Configuration();
         if (!conf.parse(confStr)) {
             throw new IllegalArgumentException("Fail to parse conf:" + confStr);
         }
         cliOptions = new CliOptions();
-        this.cliClientService = (BoltCliClientService) cliClientService;
+        this.cliClientService = (CliClientService) cliClientService;
     }
 
     /**
@@ -96,7 +97,9 @@ public void start() {
 
             cliClientService.init(cliOptions);
 
-            rpcClient = cliClientService.getRpcClient();
+            BoltRpcClient jraftRpcClient = ((BoltRpcClient) ((AbstractClientService) cliClientService)
+                .getRpcClient());
+            rpcClient = jraftRpcClient.getRpcClient();
 
             RaftClientConnectionHandler raftClientConnectionHandler = new RaftClientConnectionHandler(
                 this);
@@ -131,6 +134,7 @@ public void shutdown() {
 
     /**
      * repick leader
+     *
      * @return
      */
     public PeerId refreshLeader() {
@@ -166,6 +170,7 @@ public static PeerId refreshLeader(CliClientService cliClientService, String gro
 
     /**
      * get leader
+     *
      * @return
      */
     public PeerId getLeader() {
@@ -178,6 +183,7 @@ public PeerId getLeader() {
 
     /**
      * raft client send request
+     *
      * @param request
      * @return
      */

server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/bootstrap/RaftServer.java

@@ -16,24 +16,20 @@
  */
 package com.alipay.sofa.registry.jraft.bootstrap;
 
-import com.alipay.remoting.ProtocolCode;
-import com.alipay.remoting.ProtocolManager;
 import com.alipay.remoting.rpc.RpcClient;
-import com.alipay.remoting.rpc.RpcServer;
-import com.alipay.remoting.rpc.protocol.RpcProtocol;
-import com.alipay.remoting.rpc.protocol.RpcProtocolV2;
 import com.alipay.sofa.jraft.Node;
 import com.alipay.sofa.jraft.RaftGroupService;
 import com.alipay.sofa.jraft.conf.Configuration;
 import com.alipay.sofa.jraft.core.NodeImpl;
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.option.NodeOptions;
 import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
-import com.alipay.sofa.jraft.rpc.impl.AbstractBoltClientService;
+import com.alipay.sofa.jraft.rpc.RpcServer;
+import com.alipay.sofa.jraft.rpc.impl.AbstractClientService;
+import com.alipay.sofa.jraft.rpc.impl.BoltRpcClient;
+import com.alipay.sofa.jraft.rpc.impl.BoltRpcServer;
 import com.alipay.sofa.jraft.storage.impl.RocksDBLogStorage;
 import com.alipay.sofa.jraft.util.StorageOptionsFactory;
-import com.alipay.sofa.jraft.util.ThreadPoolMetricSet;
-import com.alipay.sofa.jraft.util.ThreadPoolUtil;
 import com.alipay.sofa.registry.common.model.store.URL;
 import com.alipay.sofa.registry.jraft.command.NotifyLeaderChange;
 import com.alipay.sofa.registry.jraft.handler.NotifyLeaderChangeHandler;
@@ -50,8 +46,6 @@
 import com.alipay.sofa.registry.remoting.bolt.BoltServer;
 import com.alipay.sofa.registry.remoting.bolt.SyncUserProcessorAdapter;
 import com.alipay.sofa.registry.util.FileUtils;
-import com.alipay.sofa.registry.util.NamedThreadFactory;
-import com.codahale.metrics.MetricRegistry;
 import org.rocksdb.BlockBasedTableConfig;
 import org.rocksdb.BloomFilter;
 import org.rocksdb.IndexType;
@@ -123,6 +117,7 @@ public RaftServer(String dataPath, String groupId, String serverIdStr, String in
      * @throws IOException
      */
     public void start(RaftServerConfig raftServerConfig) throws IOException {
+
         FileUtils.forceMkdir(new File(dataPath));
 
         serverHandlers.add(new RaftServerHandler(this, raftServerExecutor));
@@ -133,8 +128,7 @@ public void start(RaftServerConfig raftServerConfig) throws IOException {
 
         boltServer.initServer();
 
-        RpcServer rpcServer = boltServer.getRpcServer();
-
+        RpcServer rpcServer = new BoltRpcServer(boltServer.getRpcServer());
         RaftRpcServerFactory.addRaftRequestProcessors(rpcServer, raftExecutor, raftExecutor);
 
         this.fsm = ServiceStateMachine.getInstance();
@@ -153,8 +147,8 @@ public void start(RaftServerConfig raftServerConfig) throws IOException {
                 node.getNodeMetrics().getMetricRegistry(), raftServerConfig.getMetricsLogger());
         }
 
-        RpcClient raftClient = ((AbstractBoltClientService) (((NodeImpl) node).getRpcService()))
-            .getRpcClient();
+        RpcClient raftClient = ((BoltRpcClient) ((AbstractClientService) (((NodeImpl) node)
+            .getRpcService())).getRpcClient()).getRpcClient();
 
         NotifyLeaderChangeHandler notifyLeaderChangeHandler = new NotifyLeaderChangeHandler(
             groupId, null);