X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FFollower.java;h=88558cac16f76f6f1cc4ddae32db7781a5b186c0;hb=fdab53ef9033fc83c812f7d3d6d3327d3d176f0f;hp=d93271072ce60f86747a5ef6ff19f825ce1473c5;hpb=7be62e955c32ff7fa10753c4307199b287b1904c;p=controller.git

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
index d93271072c..88558cac16 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java
@@ -11,6 +11,7 @@ package org.opendaylight.controller.cluster.raft.behaviors;
 import akka.actor.ActorRef;
 import org.opendaylight.controller.cluster.raft.RaftActorContext;
 import org.opendaylight.controller.cluster.raft.RaftState;
+import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
 import org.opendaylight.controller.cluster.raft.internal.messages.ElectionTimeout;
 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
@@ -36,6 +37,114 @@ public class Follower extends AbstractRaftActorBehavior {
 
     @Override protected RaftState handleAppendEntries(ActorRef sender,
         AppendEntries appendEntries, RaftState suggestedState) {
+
+        // If we got here then we do appear to be talking to the leader
+        leaderId = appendEntries.getLeaderId();
+
+        // 2. Reply false if log doesn’t contain an entry at prevLogIndex
+        // whose term matches prevLogTerm (§5.3)
+        ReplicatedLogEntry previousEntry = context.getReplicatedLog()
+            .get(appendEntries.getPrevLogIndex());
+
+
+        if (lastIndex() > -1 && previousEntry != null
+            && previousEntry.getTerm() != appendEntries
+            .getPrevLogTerm()) {
+
+            context.getLogger().debug(
+                "Cannot append entries because previous entry term "
+                    + previousEntry.getTerm()
+                    + " is not equal to append entries prevLogTerm "
+                    + appendEntries.getPrevLogTerm());
+
+            sender.tell(
+                new AppendEntriesReply(context.getId(), currentTerm(), false,
+                    lastIndex(), lastTerm()), actor()
+            );
+            return state();
+        }
+
+        if (appendEntries.getEntries() != null
+            && appendEntries.getEntries().size() > 0) {
+            context.getLogger().debug(
+                "Number of entries to be appended = " + appendEntries
+                    .getEntries().size());
+
+            // 3. If an existing entry conflicts with a new one (same index
+            // but different terms), delete the existing entry and all that
+            // follow it (§5.3)
+            int addEntriesFrom = 0;
+            if (context.getReplicatedLog().size() > 0) {
+                for (int i = 0;
+                     i < appendEntries.getEntries()
+                         .size(); i++, addEntriesFrom++) {
+                    ReplicatedLogEntry matchEntry =
+                        appendEntries.getEntries().get(i);
+                    ReplicatedLogEntry newEntry = context.getReplicatedLog()
+                        .get(matchEntry.getIndex());
+
+                    if (newEntry == null) {
+                        //newEntry not found in the log
+                        break;
+                    }
+
+                    if (newEntry != null && newEntry.getTerm() == matchEntry
+                        .getTerm()) {
+                        continue;
+                    }
+                    if (newEntry != null && newEntry.getTerm() != matchEntry
+                        .getTerm()) {
+                        context.getLogger().debug(
+                            "Removing entries from log starting at "
+                                + matchEntry.getIndex());
+                        context.getReplicatedLog()
+                            .removeFrom(matchEntry.getIndex());
+                        break;
+                    }
+                }
+            }
+
+            context.getLogger().debug(
+                "After cleanup entries to be added from = " + (addEntriesFrom
+                    + lastIndex()));
+
+            // 4. Append any new entries not already in the log
+            for (int i = addEntriesFrom;
+                 i < appendEntries.getEntries().size(); i++) {
+                context.getLogger().debug(
+                    "Append entry to log " + appendEntries.getEntries().get(i)
+                        .toString());
+                context.getReplicatedLog()
+                    .appendAndPersist(appendEntries.getEntries().get(i));
+            }
+
+            context.getLogger().debug(
+                "Log size is now " + context.getReplicatedLog().size());
+        }
+
+
+        // 5. If leaderCommit > commitIndex, set commitIndex =
+        // min(leaderCommit, index of last new entry)
+
+        long prevCommitIndex = context.getCommitIndex();
+
+        context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(),
+            context.getReplicatedLog().lastIndex()));
+
+        if (prevCommitIndex != context.getCommitIndex()) {
+            context.getLogger()
+                .debug("Commit index set to " + context.getCommitIndex());
+        }
+
+        // If commitIndex > lastApplied: increment lastApplied, apply
+        // log[lastApplied] to state machine (§5.3)
+        if (appendEntries.getLeaderCommit() > context.getLastApplied()) {
+            applyLogToStateMachine(appendEntries.getLeaderCommit());
+        }
+
+        sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
+            lastIndex(), lastTerm()), actor());
+
         return suggestedState;
     }
 
@@ -49,7 +158,7 @@ public class Follower extends AbstractRaftActorBehavior {
         return suggestedState;
     }
 
-    @Override protected RaftState state() {
+    @Override public RaftState state() {
         return RaftState.Follower;
     }
 
@@ -62,4 +171,8 @@ public class Follower extends AbstractRaftActorBehavior {
 
         return super.handleMessage(sender, message);
     }
+
+    @Override public void close() throws Exception {
+        stopElection();
+    }
 }