Log entry checksum validation

jraft-core/src/main/java/com/alipay/sofa/jraft/JRaftUtils.java

@@ -20,7 +20,6 @@
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.lang.StringUtils;
 
@@ -29,6 +28,7 @@
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.option.BootstrapOptions;
 import com.alipay.sofa.jraft.util.Endpoint;
+import com.alipay.sofa.jraft.util.NamedThreadFactory;
 import com.alipay.sofa.jraft.util.ThreadPoolUtil;
 
 /**
@@ -49,7 +49,7 @@ private JRaftUtils() {
      * @param opts options of bootstrap
      * @return true if bootstrap success
      */
-    public static boolean bootstrap(BootstrapOptions opts) throws InterruptedException {
+    public static boolean bootstrap(final BootstrapOptions opts) throws InterruptedException {
         final NodeImpl node = new NodeImpl();
         final boolean ret = node.bootstrap(opts);
         node.shutdown();
@@ -64,7 +64,7 @@ public static boolean bootstrap(BootstrapOptions opts) throws InterruptedExcepti
      * @param number thread number
      * @return a new {@link ThreadPoolExecutor} instance
      */
-    public static Executor createExecutor(final String prefix, int number) {
+    public static Executor createExecutor(final String prefix, final int number) {
         if (number <= 0) {
             return null;
         }
@@ -81,24 +81,14 @@ public static Executor createExecutor(final String prefix, int number) {
      * @since 0.0.3
      */
     public static ThreadFactory createThreadFactory(final String prefixName) {
-        return new ThreadFactory() {
-            private final AtomicInteger c = new AtomicInteger(0);
-
-            @Override
-            public Thread newThread(Runnable r) {
-                final Thread t = new Thread(r);
-                t.setName(prefixName + c.getAndIncrement());
-                t.setDaemon(true);
-                return t;
-            }
-        };
+        return new NamedThreadFactory(prefixName);
     }
 
     /**
      * Create a configuration from a string in the form of "host1:port1[:idx],host2:port2[:idx]......",
      * returns a empty configuration when string is blank.
      */
-    public static Configuration getConfiguration(String s) {
+    public static Configuration getConfiguration(final String s) {
         final Configuration conf = new Configuration();
         if (StringUtils.isBlank(s)) {
             return conf;
@@ -113,7 +103,7 @@ public static Configuration getConfiguration(String s) {
      * Create a peer from a string in the form of "host:port[:idx]",
      * returns a empty peer when string is blank.
      */
-    public static PeerId getPeerId(String s) {
+    public static PeerId getPeerId(final String s) {
         final PeerId peer = new PeerId();
         if (StringUtils.isBlank(s)) {
             return peer;
@@ -128,11 +118,11 @@ public static PeerId getPeerId(String s) {
      * Create a Endpoint instance from  a string in the form of "host:port",
      * returns null when string is blank.
      */
-    public static Endpoint getEndPoint(String s) {
+    public static Endpoint getEndPoint(final String s) {
         if (StringUtils.isBlank(s)) {
             return null;
         }
-        final String[] tmps = s.split(":");
+        final String[] tmps = StringUtils.split(s, ':');
         if (tmps.length != 2) {
             throw new IllegalArgumentException("Invalid endpoint string: " + s);
         }

jraft-core/src/main/java/com/alipay/sofa/jraft/RaftGroupService.java

@@ -64,7 +64,7 @@ public class RaftGroupService {
     /**
      * If we want to share the rpcServer instance, then we can't stop it when shutdown.
      */
-    private boolean             sharedRpcServer;
+    private final boolean       sharedRpcServer;
 
     /**
      * The raft group id
@@ -75,22 +75,19 @@ public class RaftGroupService {
      */
     private Node                node;
 
-    public RaftGroupService() {
-        this(null, null, null);
-    }
-
-    public RaftGroupService(String groupId, PeerId serverId, NodeOptions nodeOptions) {
+    public RaftGroupService(final String groupId, final PeerId serverId, final NodeOptions nodeOptions) {
         this(groupId, serverId, nodeOptions, RaftRpcServerFactory.createRaftRpcServer(serverId.getEndpoint(),
             JRaftUtils.createExecutor("RAFT-RPC-executor-", nodeOptions.getRaftRpcThreadPoolSize()),
             JRaftUtils.createExecutor("CLI-RPC-executor-", nodeOptions.getCliRpcThreadPoolSize())));
     }
 
-    public RaftGroupService(String groupId, PeerId serverId, NodeOptions nodeOptions, RpcServer rpcServer) {
+    public RaftGroupService(final String groupId, final PeerId serverId, final NodeOptions nodeOptions,
+                            final RpcServer rpcServer) {
         this(groupId, serverId, nodeOptions, rpcServer, false);
     }
 
-    public RaftGroupService(String groupId, PeerId serverId, NodeOptions nodeOptions, RpcServer rpcServer,
-                            boolean sharedRpcServer) {
+    public RaftGroupService(final String groupId, final PeerId serverId, final NodeOptions nodeOptions,
+                            final RpcServer rpcServer, final boolean sharedRpcServer) {
         super();
         this.groupId = groupId;
         this.serverId = serverId;
@@ -115,21 +112,21 @@ public synchronized Node start() {
      *
      * @param startRpcServer whether to start RPC server.
      */
-    public synchronized Node start(boolean startRpcServer) {
+    public synchronized Node start(final boolean startRpcServer) {
         if (this.started) {
             return this.node;
         }
         if (this.serverId == null || this.serverId.getEndpoint() == null
             || this.serverId.getEndpoint().equals(new Endpoint(Utils.IP_ANY, 0))) {
-            throw new IllegalArgumentException("Blank serverId:" + serverId);
+            throw new IllegalArgumentException("Blank serverId:" + this.serverId);
         }
         if (StringUtils.isBlank(this.groupId)) {
             throw new IllegalArgumentException("Blank group id" + this.groupId);
         }
         //Adds RPC server to Server.
         NodeManager.getInstance().addAddress(this.serverId.getEndpoint());
 
-        this.node = RaftServiceFactory.createAndInitRaftNode(groupId, serverId, nodeOptions);
+        this.node = RaftServiceFactory.createAndInitRaftNode(this.groupId, this.serverId, this.nodeOptions);
         if (startRpcServer) {
             this.rpcServer.start();
         } else {
@@ -193,7 +190,7 @@ public String getGroupId() {
     /**
      * Set the raft group id
      */
-    public void setGroupId(String groupId) {
+    public void setGroupId(final String groupId) {
         if (this.started) {
             throw new IllegalStateException("Raft group service already started");
         }
@@ -210,7 +207,7 @@ public PeerId getServerId() {
     /**
      * Set the node serverId
      */
-    public void setServerId(PeerId serverId) {
+    public void setServerId(final PeerId serverId) {
         if (this.started) {
             throw new IllegalStateException("Raft group service already started");
         }
@@ -227,7 +224,7 @@ public RpcOptions getNodeOptions() {
     /**
      * Set node options.
      */
-    public void setNodeOptions(NodeOptions nodeOptions) {
+    public void setNodeOptions(final NodeOptions nodeOptions) {
         if (this.started) {
             throw new IllegalStateException("Raft group service already started");
         }
@@ -248,7 +245,7 @@ public RpcServer getRpcServer() {
     /**
      * Set rpc server.
      */
-    public void setRpcServer(RpcServer rpcServer) {
+    public void setRpcServer(final RpcServer rpcServer) {
         if (this.started) {
             throw new IllegalStateException("Raft group service already started");
         }

jraft-core/src/main/java/com/alipay/sofa/jraft/conf/Configuration.java

@@ -23,6 +23,8 @@
 import java.util.List;
 import java.util.Set;
 
+import org.apache.commons.lang.StringUtils;
+
 import com.alipay.sofa.jraft.entity.PeerId;
 import com.alipay.sofa.jraft.util.Copiable;
 
@@ -40,7 +42,7 @@ public Configuration() {
         super();
     }
 
-    public Configuration(Iterable<PeerId> conf) {
+    public Configuration(final Iterable<PeerId> conf) {
         for (final PeerId peer : conf) {
             this.peers.add(peer.copy());
         }
@@ -56,50 +58,50 @@ public void reset() {
     }
 
     public boolean isEmpty() {
-        return peers.isEmpty();
+        return this.peers.isEmpty();
     }
 
     public int size() {
-        return peers.size();
+        return this.peers.size();
     }
 
     @Override
     public Iterator<PeerId> iterator() {
-        return peers.iterator();
+        return this.peers.iterator();
     }
 
     public Set<PeerId> getPeerSet() {
-        return new HashSet<>(peers);
+        return new HashSet<>(this.peers);
     }
 
     public List<PeerId> listPeers() {
         return new ArrayList<>(this.peers);
     }
 
     public List<PeerId> getPeers() {
-        return peers;
+        return this.peers;
     }
 
-    public void setPeers(List<PeerId> peers) {
+    public void setPeers(final List<PeerId> peers) {
         this.peers.clear();
         for (final PeerId peer : peers) {
             this.peers.add(peer.copy());
         }
     }
 
-    public void appendPeers(Collection<PeerId> set) {
+    public void appendPeers(final Collection<PeerId> set) {
         this.peers.addAll(set);
     }
 
-    public boolean addPeer(PeerId peer) {
+    public boolean addPeer(final PeerId peer) {
         return this.peers.add(peer);
     }
 
-    public boolean removePeer(PeerId peer) {
+    public boolean removePeer(final PeerId peer) {
         return this.peers.remove(peer);
     }
 
-    public boolean contains(PeerId peer) {
+    public boolean contains(final PeerId peer) {
         return this.peers.contains(peer);
     }
 
@@ -112,7 +114,7 @@ public int hashCode() {
     }
 
     @Override
-    public boolean equals(Object obj) {
+    public boolean equals(final Object obj) {
         if (this == obj) {
             return true;
         }
@@ -133,7 +135,7 @@ public boolean equals(Object obj) {
     @Override
     public String toString() {
         final StringBuilder sb = new StringBuilder();
-        final List<PeerId> peers = this.listPeers();
+        final List<PeerId> peers = listPeers();
         int i = 0;
         final int size = peers.size();
         for (final PeerId peer : peers) {
@@ -146,12 +148,12 @@ public String toString() {
         return sb.toString();
     }
 
-    public boolean parse(String conf) {
+    public boolean parse(final String conf) {
         if (conf == null) {
             return false;
         }
         reset();
-        final String[] peerStrs = conf.split(",");
+        final String[] peerStrs = StringUtils.split(conf, ",");
         for (final String peerStr : peerStrs) {
             final PeerId peer = new PeerId();
             if (peer.parse(peerStr)) {
@@ -166,7 +168,7 @@ public boolean parse(String conf) {
      *  |included| would be assigned to |*this| - |rhs|
      *  |excluded| would be assigned to |rhs| - |*this|
      */
-    public void diff(Configuration rhs, Configuration included, Configuration excluded) {
+    public void diff(final Configuration rhs, final Configuration included, final Configuration excluded) {
         included.peers = new ArrayList<>(this.peers);
         included.peers.removeAll(rhs.peers);
         excluded.peers = new ArrayList<>(rhs.peers);

jraft-core/src/main/java/com/alipay/sofa/jraft/core/DefaultJRaftServiceFactory.java

@@ -19,8 +19,8 @@
 import org.apache.commons.lang.StringUtils;
 
 import com.alipay.sofa.jraft.JRaftServiceFactory;
-import com.alipay.sofa.jraft.entity.codec.DefaultLogEntryCodecFactory;
 import com.alipay.sofa.jraft.entity.codec.LogEntryCodecFactory;
+import com.alipay.sofa.jraft.entity.codec.v2.LogEntryV2CodecFactory;
 import com.alipay.sofa.jraft.option.RaftOptions;
 import com.alipay.sofa.jraft.storage.LogStorage;
 import com.alipay.sofa.jraft.storage.RaftMetaStorage;
@@ -65,6 +65,6 @@ public RaftMetaStorage createRaftMetaStorage(final String uri, final RaftOptions
 
     @Override
     public LogEntryCodecFactory createLogEntryCodecFactory() {
-        return DefaultLogEntryCodecFactory.getInstance();
+        return LogEntryV2CodecFactory.getInstance();
     }
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

@@ -176,7 +176,7 @@ public boolean init(final FSMCallerOptions opts) {
         this.node = opts.getNode();
         this.nodeMetrics = this.node.getNodeMetrics();
         this.lastAppliedIndex.set(opts.getBootstrapId().getIndex());
-        this.notifyLastAppliedIndexUpdated(this.lastAppliedIndex.get());
+        notifyLastAppliedIndexUpdated(this.lastAppliedIndex.get());
         this.lastAppliedTerm = opts.getBootstrapId().getTerm();
         this.disruptor = new Disruptor<>(new ApplyTaskFactory(), opts.getDisruptorBufferSize(), new NamedThreadFactory(
             "JRaft-FSMCaller-disruptor-", true));
@@ -262,15 +262,15 @@ public boolean onSnapshotSave(final SaveSnapshotClosure done) {
 
     @Override
     public boolean onLeaderStop(final Status status) {
-        return this.enqueueTask((task, sequence) -> {
+        return enqueueTask((task, sequence) -> {
             task.type = TaskType.LEADER_STOP;
             task.status = new Status(status);
         });
     }
 
     @Override
     public boolean onLeaderStart(final long term) {
-        return this.enqueueTask((task, sequence) -> {
+        return enqueueTask((task, sequence) -> {
             task.type = TaskType.LEADER_START;
             task.term = term;
         });
@@ -301,7 +301,7 @@ public boolean onStopFollowing(final LeaderChangeContext ctx) {
     public class OnErrorClosure implements Closure {
         private RaftException error;
 
-        public OnErrorClosure(RaftException error) {
+        public OnErrorClosure(final RaftException error) {
             super();
             this.error = error;
         }
@@ -310,7 +310,7 @@ public RaftException getError() {
             return this.error;
         }
 
-        public void setError(RaftException error) {
+        public void setError(final RaftException error) {
             this.error = error;
         }
 
@@ -322,7 +322,7 @@ public void run(final Status st) {
     @Override
     public boolean onError(final RaftException error) {
         final OnErrorClosure c = new OnErrorClosure(error);
-        return this.enqueueTask((task, sequence) -> {
+        return enqueueTask((task, sequence) -> {
             task.type = TaskType.ERROR;
             task.done = c;
         });
@@ -406,7 +406,7 @@ private long runApplyTask(final ApplyTask task, long maxCommittedIndex, final bo
                         break;
                 }
             } finally {
-                nodeMetrics.recordLatency(task.type.metricName(), Utils.monotonicMs() - startMs);
+                this.nodeMetrics.recordLatency(task.type.metricName(), Utils.monotonicMs() - startMs);
             }
         }
         try {
@@ -680,7 +680,7 @@ private void setError(final RaftException e) {
 
     @OnlyForTest
     RaftException getError() {
-        return error;
+        return this.error;
     }
 
     private boolean passByStatus(final Closure done) {