bugfix: pesist commitIndex into raft meta log

Author: wenweihu86

pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>com.github.wenweihu86.raft</groupId>
-    <artifactId>raft-java-all</artifactId>
+    <artifactId>raft-java-parent</artifactId>
     <version>1.9.0</version>
     <packaging>pom</packaging>
 

raft-java-core/src/main/java/com/github/wenweihu86/raft/RaftNode.java

@@ -87,7 +87,7 @@ public RaftNode(RaftOptions raftOptions,
 
         currentTerm = raftLog.getMetaData().getCurrentTerm();
         votedFor = raftLog.getMetaData().getVotedFor();
-        commitIndex = Math.max(snapshot.getMetaData().getLastIncludedIndex(), commitIndex);
+        commitIndex = Math.max(snapshot.getMetaData().getLastIncludedIndex(), raftLog.getMetaData().getCommitIndex());
         // discard old log entries
         if (snapshot.getMetaData().getLastIncludedIndex() > 0
                 && raftLog.getFirstLogIndex() <= snapshot.getMetaData().getLastIncludedIndex()) {
@@ -156,7 +156,7 @@ public boolean replicate(byte[] data, RaftProto.EntryType entryType) {
             List<RaftProto.LogEntry> entries = new ArrayList<>();
             entries.add(logEntry);
             newLastLogIndex = raftLog.append(entries);
-            raftLog.updateMetaData(currentTerm, null, raftLog.getFirstLogIndex());
+//            raftLog.updateMetaData(currentTerm, null, raftLog.getFirstLogIndex());
 
             for (RaftProto.Server server : configuration.getServersList()) {
                 final Peer peer = peerMap.get(server.getServerId());
@@ -304,7 +304,7 @@ public void stepDown(long newTerm) {
             currentTerm = newTerm;
             leaderId = 0;
             votedFor = 0;
-            raftLog.updateMetaData(currentTerm, votedFor, null);
+            raftLog.updateMetaData(currentTerm, votedFor, null, null);
         }
         state = NodeState.STATE_FOLLOWER;
         // stop heartbeat
@@ -760,6 +760,7 @@ private void advanceCommitIndex() {
         }
         long oldCommitIndex = commitIndex;
         commitIndex = newCommitIndex;
+        raftLog.updateMetaData(currentTerm, null, raftLog.getFirstLogIndex(), commitIndex);
         // 同步到状态机
         for (long index = oldCommitIndex + 1; index <= newCommitIndex; index++) {
             RaftProto.LogEntry entry = raftLog.getEntry(index);

raft-java-core/src/main/java/com/github/wenweihu86/raft/proto/RaftProto.java

@@ -33,6 +33,7 @@ message LogMetaData {
     optional uint64 current_term = 1;
     optional uint32 voted_for = 2;
     optional uint64 first_log_index = 3;
+    optional uint64 commit_index = 4;
 }
 
 message SnapshotMetaData {

raft-java-core/src/main/java/com/github/wenweihu86/raft/proto/raft.proto

@@ -84,7 +84,7 @@ public RaftProto.VoteResponse requestVote(RaftProto.VoteRequest request) {
             if (raftNode.getVotedFor() == 0 && logIsOk) {
                 raftNode.stepDown(request.getTerm());
                 raftNode.setVotedFor(request.getServerId());
-                raftNode.getRaftLog().updateMetaData(raftNode.getCurrentTerm(), raftNode.getVotedFor(), null);
+                raftNode.getRaftLog().updateMetaData(raftNode.getCurrentTerm(), raftNode.getVotedFor(), null, null);
                 responseBuilder.setGranted(true);
                 responseBuilder.setTerm(raftNode.getCurrentTerm());
             }
@@ -174,8 +174,8 @@ public RaftProto.AppendEntriesResponse appendEntries(RaftProto.AppendEntriesRequ
                 entries.add(entry);
             }
             raftNode.getRaftLog().append(entries);
-            raftNode.getRaftLog().updateMetaData(raftNode.getCurrentTerm(),
-                    null, raftNode.getRaftLog().getFirstLogIndex());
+//            raftNode.getRaftLog().updateMetaData(raftNode.getCurrentTerm(),
+//                    null, raftNode.getRaftLog().getFirstLogIndex());
             responseBuilder.setLastLogIndex(raftNode.getRaftLog().getLastLogIndex());
 
             advanceCommitIndex(request);
@@ -313,6 +313,7 @@ private void advanceCommitIndex(RaftProto.AppendEntriesRequest request) {
         long newCommitIndex = Math.min(request.getCommitIndex(),
                 request.getPrevLogIndex() + request.getEntriesCount());
         raftNode.setCommitIndex(newCommitIndex);
+        raftNode.getRaftLog().updateMetaData(null,null, null, newCommitIndex);
         if (raftNode.getLastAppliedIndex() < raftNode.getCommitIndex()) {
             // apply state machine
             for (long index = raftNode.getLastAppliedIndex() + 1;

raft-java-core/src/main/java/com/github/wenweihu86/raft/service/impl/RaftConsensusServiceImpl.java

@@ -191,7 +191,7 @@ public void truncatePrefix(long newFirstIndex) {
         } else {
             newActualFirstIndex = startLogIndexSegmentMap.firstKey();
         }
-        updateMetaData(null, null, newActualFirstIndex);
+        updateMetaData(null, null, newActualFirstIndex, null);
         LOG.info("Truncating log from old first index {} to new first index {}",
                 oldFirstIndex, newActualFirstIndex);
     }
@@ -316,7 +316,15 @@ public RaftProto.LogMetaData readMetaData() {
         }
     }
 
-    public void updateMetaData(Long currentTerm, Integer votedFor, Long firstLogIndex) {
+    /**
+     * 更新raft log meta data，
+     * 包括commitIndex， fix bug: https://github.com/wenweihu86/raft-java/issues/19
+     * @param currentTerm
+     * @param votedFor
+     * @param firstLogIndex
+     * @param commitIndex
+     */
+    public void updateMetaData(Long currentTerm, Integer votedFor, Long firstLogIndex, Long commitIndex) {
         RaftProto.LogMetaData.Builder builder = RaftProto.LogMetaData.newBuilder(this.metaData);
         if (currentTerm != null) {
             builder.setCurrentTerm(currentTerm);
@@ -327,6 +335,9 @@ public void updateMetaData(Long currentTerm, Integer votedFor, Long firstLogInde
         if (firstLogIndex != null) {
             builder.setFirstLogIndex(firstLogIndex);
         }
+        if (commitIndex != null) {
+            builder.setCommitIndex(commitIndex);
+        }
         this.metaData = builder.build();
 
         String fileName = logDir + File.separator + "metadata";