Log entry checksum validation

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

@@ -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;
  }
@@ -85,9 +85,9 @@ public static ThreadFactory createThreadFactory(final String prefixName) {
  private final AtomicInteger c = new AtomicInteger(0);
 
  @Override
- public Thread newThread(Runnable r) {
+ public Thread newThread(final Runnable r) {
  final Thread t = new Thread(r);
- t.setName(prefixName + c.getAndIncrement());
+ t.setName(prefixName + this.c.getAndIncrement());
  t.setDaemon(true);
  return t;
  }
@@ -98,7 +98,7 @@ public Thread newThread(Runnable r) {
  * 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 +113,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 +128,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) {