From 583f6075e842a6a37b83bd01e478aebc70c6af73 Mon Sep 17 00:00:00 2001
From: Moiz Raja <moraja@cisco.com>
Date: Thu, 10 Jul 2014 18:37:22 -0700
Subject: [PATCH] Implement behavior common to a RaftActor in all it's states

The common behavior for a RaftActor is
- Change state to follower when it receives a RaftRPC with a newer term
- Increment the applied state of the commitIndex of the AppendEntries message
  is higher than it's applied state

Change-Id: I1fc0cdcb90368f3c07f555b307ac827286386651
Signed-off-by: Moiz Raja <moraja@cisco.com>
---
 .../controller/cluster/raft/ElectionTerm.java |   2 +-
 .../cluster/raft/ElectionTermImpl.java        |   4 +-
 .../behaviors/AbstractRaftActorBehavior.java  | 130 +++++++++++++++++-
 .../cluster/raft/behaviors/Candidate.java     |  30 +++-
 .../cluster/raft/behaviors/Follower.java      |  38 ++++-
 .../cluster/raft/behaviors/Leader.java        |  32 ++++-
 .../raft/messages/AbstractRaftRPC.java        |  24 ++++
 .../cluster/raft/messages/AppendEntries.java  |  21 +--
 .../raft/messages/AppendEntriesReply.java     |   6 +-
 .../cluster/raft/messages/RaftRPC.java        |  13 ++
 .../cluster/raft/messages/RequestVote.java    |   7 +-
 .../raft/messages/RequestVoteReply.java       |   7 +-
 .../cluster/raft/MockRaftActorContext.java    |  14 +-
 .../AbstractRaftActorBehaviorTest.java        |  91 ++++++++++++
 .../cluster/raft/behaviors/CandidateTest.java |  12 ++
 .../cluster/raft/behaviors/FollowerTest.java  |   9 ++
 .../cluster/raft/behaviors/LeaderTest.java    |   8 +-
 17 files changed, 412 insertions(+), 36 deletions(-)
 create mode 100644 opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AbstractRaftRPC.java
 create mode 100644 opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RaftRPC.java
 create mode 100644 opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java
 create mode 100644 opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/CandidateTest.java
 create mode 100644 opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTerm.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTerm.java
index 2cf39b59f1..4b0367f0ad 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTerm.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTerm.java
@@ -39,5 +39,5 @@ public interface ElectionTerm {
      * @param currentTerm
      * @param votedFor
      */
-    void update(AtomicLong currentTerm, String votedFor);
+    void update(long currentTerm, String votedFor);
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTermImpl.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTermImpl.java
index e918f75273..e75e0c5bb6 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTermImpl.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/ElectionTermImpl.java
@@ -38,8 +38,8 @@ public class ElectionTermImpl implements ElectionTerm{
         return votedFor;
     }
 
-    public void update(AtomicLong currentTerm, String votedFor){
-        this.currentTerm = currentTerm;
+    public void update(long currentTerm, String votedFor){
+        this.currentTerm.set(currentTerm);
         this.votedFor = votedFor;
 
         // TODO : Write to some persistent state
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java
index fde104223b..d7a8d5abb3 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java
@@ -8,11 +8,18 @@
 
 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.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
 
 /**
  * Abstract class that represents the behavior of a RaftActor
- * <p>
+ * <p/>
  * All Servers:
  * <ul>
  * <li> If commitIndex > lastApplied: increment lastApplied, apply
@@ -31,4 +38,125 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
     protected AbstractRaftActorBehavior(RaftActorContext context) {
         this.context = context;
     }
+
+    /**
+     * Derived classes should not directly handle AppendEntries messages it
+     * should let the base class handle it first. Once the base class handles
+     * the AppendEntries message and does the common actions that are applicable
+     * in all RaftState's it will delegate the handling of the AppendEntries
+     * message to the derived class to do more state specific handling by calling
+     * this method
+     *
+     * @param sender         The actor that sent this message
+     * @param appendEntries  The AppendEntries message
+     * @param suggestedState The state that the RaftActor should be in based
+     *                       on the base class's processing of the AppendEntries
+     *                       message
+     * @return
+     */
+    protected abstract RaftState handleAppendEntries(ActorRef sender,
+        AppendEntries appendEntries, RaftState suggestedState);
+
+    /**
+     * Derived classes should not directly handle AppendEntriesReply messages it
+     * should let the base class handle it first. Once the base class handles
+     * the AppendEntriesReply message and does the common actions that are
+     * applicable in all RaftState's it will delegate the handling of the
+     * AppendEntriesReply message to the derived class to do more state specific
+     * handling by calling this method
+     *
+     * @param sender             The actor that sent this message
+     * @param appendEntriesReply The AppendEntriesReply message
+     * @param suggestedState     The state that the RaftActor should be in based
+     *                           on the base class's processing of the
+     *                           AppendEntriesReply message
+     * @return
+     */
+
+    protected abstract RaftState handleAppendEntriesReply(ActorRef sender,
+        AppendEntriesReply appendEntriesReply, RaftState suggestedState);
+
+    /**
+     * Derived classes should not directly handle RequestVote messages it
+     * should let the base class handle it first. Once the base class handles
+     * the RequestVote message and does the common actions that are applicable
+     * in all RaftState's it will delegate the handling of the RequestVote
+     * message to the derived class to do more state specific handling by calling
+     * this method
+     *
+     * @param sender         The actor that sent this message
+     * @param requestVote    The RequestVote message
+     * @param suggestedState The state that the RaftActor should be in based
+     *                       on the base class's processing of the RequestVote
+     *                       message
+     * @return
+     */
+    protected abstract RaftState handleRequestVote(ActorRef sender,
+        RequestVote requestVote, RaftState suggestedState);
+
+    /**
+     * Derived classes should not directly handle RequestVoteReply messages it
+     * should let the base class handle it first. Once the base class handles
+     * the RequestVoteReply message and does the common actions that are
+     * applicable in all RaftState's it will delegate the handling of the
+     * RequestVoteReply message to the derived class to do more state specific
+     * handling by calling this method
+     *
+     * @param sender           The actor that sent this message
+     * @param requestVoteReply The RequestVoteReply message
+     * @param suggestedState   The state that the RaftActor should be in based
+     *                         on the base class's processing of the RequestVote
+     *                         message
+     * @return
+     */
+
+    protected abstract RaftState handleRequestVoteReply(ActorRef sender,
+        RequestVoteReply requestVoteReply, RaftState suggestedState);
+
+    /**
+     * @return The derived class should return the state that corresponds to
+     * it's behavior
+     */
+    protected abstract RaftState state();
+
+    @Override
+    public RaftState handleMessage(ActorRef sender, Object message) {
+        RaftState raftState = state();
+        if (message instanceof RaftRPC) {
+            raftState = applyTerm((RaftRPC) message);
+        }
+        if (message instanceof AppendEntries) {
+            AppendEntries appendEntries = (AppendEntries) message;
+            if (appendEntries.getLeaderCommit() > context.getLastApplied()
+                .get()) {
+                applyLogToStateMachine(appendEntries.getLeaderCommit());
+            }
+            raftState = handleAppendEntries(sender, appendEntries, raftState);
+        } else if (message instanceof AppendEntriesReply) {
+            raftState =
+                handleAppendEntriesReply(sender, (AppendEntriesReply) message,
+                    raftState);
+        } else if (message instanceof RequestVote) {
+            raftState =
+                handleRequestVote(sender, (RequestVote) message, raftState);
+        } else if (message instanceof RequestVoteReply) {
+            raftState =
+                handleRequestVoteReply(sender, (RequestVoteReply) message,
+                    raftState);
+        }
+        return raftState;
+    }
+
+    private RaftState applyTerm(RaftRPC rpc) {
+        if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()
+            .get()) {
+            context.getTermInformation().update(rpc.getTerm(), null);
+            return RaftState.Follower;
+        }
+        return state();
+    }
+
+    private void applyLogToStateMachine(long index) {
+        context.getLastApplied().set(index);
+    }
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java
index fb480a9433..774691154a 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java
@@ -11,6 +11,10 @@ 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.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
 
 import java.util.List;
 
@@ -40,8 +44,32 @@ public class Candidate extends AbstractRaftActorBehavior {
         this.peers = peers;
     }
 
+    @Override protected RaftState handleAppendEntries(ActorRef sender,
+        AppendEntries appendEntries, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+        AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVote(ActorRef sender,
+        RequestVote requestVote, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+        RequestVoteReply requestVoteReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState state() {
+        return RaftState.Candidate;
+    }
+
     @Override
     public RaftState handleMessage(ActorRef sender, Object message) {
-        return RaftState.Candidate;
+        return super.handleMessage(sender, message);
     }
 }
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 4c97d933c8..1bdcc8bdb4 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,16 +11,52 @@ 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.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
 
 /**
  * The behavior of a RaftActor in the Follower state
+ *
+ * <ul>
+ * <li> Respond to RPCs from candidates and leaders
+ * <li> If election timeout elapses without receiving AppendEntries
+ * RPC from current leader or granting vote to candidate:
+ * convert to candidate
+ * </ul>
+ *
  */
 public class Follower extends AbstractRaftActorBehavior {
     public Follower(RaftActorContext context) {
         super(context);
     }
 
-    @Override public RaftState handleMessage(ActorRef sender, Object message) {
+    @Override protected RaftState handleAppendEntries(ActorRef sender,
+        AppendEntries appendEntries, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+        AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVote(ActorRef sender,
+        RequestVote requestVote, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+        RequestVoteReply requestVoteReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState state() {
         return RaftState.Follower;
     }
+
+    @Override public RaftState handleMessage(ActorRef sender, Object message) {
+        return super.handleMessage(sender, message);
+    }
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java
index cf63f4d1e2..4adf8d08fd 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java
@@ -16,6 +16,9 @@ import org.opendaylight.controller.cluster.raft.RaftReplicator;
 import org.opendaylight.controller.cluster.raft.RaftState;
 import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -71,6 +74,30 @@ public class Leader extends AbstractRaftActorBehavior {
 
     }
 
+    @Override protected RaftState handleAppendEntries(ActorRef sender,
+        AppendEntries appendEntries, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+        AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVote(ActorRef sender,
+        RequestVote requestVote, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+        RequestVoteReply requestVoteReply, RaftState suggestedState) {
+        return suggestedState;
+    }
+
+    @Override protected RaftState state() {
+        return RaftState.Leader;
+    }
+
     @Override public RaftState handleMessage(ActorRef sender, Object message) {
         Preconditions.checkNotNull(sender, "sender should not be null");
 
@@ -79,8 +106,9 @@ public class Leader extends AbstractRaftActorBehavior {
                 context.getTermInformation().getCurrentTerm().get() , context.getId(),
                 context.getReplicatedLog().last().getIndex(),
                 context.getReplicatedLog().last().getTerm(),
-                Collections.EMPTY_LIST), context.getActor());
+                Collections.EMPTY_LIST, context.getCommitIndex().get()), context.getActor());
+            return state();
         }
-        return RaftState.Leader;
+        return super.handleMessage(sender, message);
     }
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AbstractRaftRPC.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AbstractRaftRPC.java
new file mode 100644
index 0000000000..3cafda9a15
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AbstractRaftRPC.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.raft.messages;
+
+public class AbstractRaftRPC implements RaftRPC {
+    // term
+    protected long term;
+
+    protected AbstractRaftRPC(long term){
+        this.term = term;
+    }
+
+    public long getTerm() {
+        return term;
+    }
+
+
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java
index e3c6ac0e72..fd4e0a274c 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntries.java
@@ -14,10 +14,7 @@ import java.util.List;
  * Invoked by leader to replicate log entries (§5.3); also used as
  * heartbeat (§5.2).
  */
-public class AppendEntries {
-    // Leaders term
-    private final long term;
-
+public class AppendEntries extends AbstractRaftRPC {
     // So that follower can redirect clients
     private final String leaderId;
 
@@ -31,17 +28,17 @@ public class AppendEntries {
     // may send more than one for efficiency)
     private final List<Object> entries;
 
+    // leader's commitIndex
+    private final long leaderCommit;
+
     public AppendEntries(long term, String leaderId, long prevLogIndex,
-        long prevLogTerm, List<Object> entries) {
-        this.term = term;
+        long prevLogTerm, List<Object> entries, long leaderCommit) {
+        super(term);
         this.leaderId = leaderId;
         this.prevLogIndex = prevLogIndex;
         this.prevLogTerm = prevLogTerm;
         this.entries = entries;
-    }
-
-    public long getTerm() {
-        return term;
+        this.leaderCommit = leaderCommit;
     }
 
     public String getLeaderId() {
@@ -59,4 +56,8 @@ public class AppendEntries {
     public List<Object> getEntries() {
         return entries;
     }
+
+    public long getLeaderCommit() {
+        return leaderCommit;
+    }
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntriesReply.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntriesReply.java
index d811464cba..28f0f6b52b 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntriesReply.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/AppendEntriesReply.java
@@ -11,16 +11,14 @@ package org.opendaylight.controller.cluster.raft.messages;
 /**
  * Reply for the AppendEntriesRpc message
  */
-public class AppendEntriesReply {
-    // currentTerm, for leader to update itself
-    private final long term;
+public class AppendEntriesReply extends AbstractRaftRPC{
 
     // true if follower contained entry matching
     // prevLogIndex and prevLogTerm
     private final boolean success;
 
     public AppendEntriesReply(long term, boolean success) {
-        this.term = term;
+        super(term);
         this.success = success;
     }
 
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RaftRPC.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RaftRPC.java
new file mode 100644
index 0000000000..a770e54f58
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RaftRPC.java
@@ -0,0 +1,13 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.raft.messages;
+
+public interface RaftRPC {
+    public long getTerm();
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVote.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVote.java
index 5828e439eb..981da17ce1 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVote.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVote.java
@@ -11,10 +11,7 @@ package org.opendaylight.controller.cluster.raft.messages;
 /**
  * Invoked by candidates to gather votes (§5.2).
  */
-public class RequestVote {
-
-    // candidate’s term
-    private final long term;
+public class RequestVote extends AbstractRaftRPC{
 
     // candidate requesting vote
     private final String candidateId;
@@ -27,7 +24,7 @@ public class RequestVote {
 
     public RequestVote(long term, String candidateId, long lastLogIndex,
         long lastLogTerm) {
-        this.term = term;
+        super(term);
         this.candidateId = candidateId;
         this.lastLogIndex = lastLogIndex;
         this.lastLogTerm = lastLogTerm;
diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVoteReply.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVoteReply.java
index 7acafecfbb..816120cd93 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVoteReply.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/messages/RequestVoteReply.java
@@ -8,16 +8,13 @@
 
 package org.opendaylight.controller.cluster.raft.messages;
 
-public class RequestVoteReply {
-
-    // currentTerm, for candidate to update itself
-    private final long term;
+public class RequestVoteReply extends AbstractRaftRPC{
 
     // true means candidate received vot
     private final boolean voteGranted;
 
     public RequestVoteReply(long term, boolean voteGranted) {
-        this.term = term;
+        super(term);
         this.voteGranted = voteGranted;
     }
 
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MockRaftActorContext.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MockRaftActorContext.java
index 05dd9e8191..2e1464f383 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MockRaftActorContext.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/MockRaftActorContext.java
@@ -20,6 +20,8 @@ public class MockRaftActorContext implements RaftActorContext {
     private String id;
     private ActorSystem system;
     private ActorRef actor;
+    private AtomicLong index = new AtomicLong(0);
+    private AtomicLong lastApplied = new AtomicLong(0);
 
     public MockRaftActorContext(){
 
@@ -51,12 +53,20 @@ public class MockRaftActorContext implements RaftActorContext {
         return new ElectionTermImpl(this.id);
     }
 
+    public void setIndex(AtomicLong index){
+        this.index = index;
+    }
+
     @Override public AtomicLong getCommitIndex() {
-        throw new UnsupportedOperationException("getCommitIndex");
+        return index;
+    }
+
+    public void setLastApplied(AtomicLong lastApplied){
+        this.lastApplied = lastApplied;
     }
 
     @Override public AtomicLong getLastApplied() {
-        throw new UnsupportedOperationException("getLastApplied");
+        return lastApplied;
     }
 
     @Override public ReplicatedLog getReplicatedLog() {
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java
new file mode 100644
index 0000000000..ea417a172e
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java
@@ -0,0 +1,91 @@
+package org.opendaylight.controller.cluster.raft.behaviors;
+
+import akka.actor.ActorRef;
+import akka.testkit.JavaTestKit;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.raft.AbstractActorTest;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftState;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+
+public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest{
+   @Test
+    public void testHandlingOfRaftRPCWithNewerTerm() throws Exception {
+        new JavaTestKit(getSystem()) {{
+
+            assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(),
+                createAppendEntriesWithNewerTerm());
+
+            assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(),
+                createAppendEntriesReplyWithNewerTerm());
+
+            assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(),
+                createRequestVoteWithNewerTerm());
+
+            assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(),
+                createRequestVoteReplyWithNewerTerm());
+
+
+        }};
+    }
+
+    @Test
+    public void testHandlingOfAppendEntriesWithNewerCommitIndex() throws Exception{
+        new JavaTestKit(getSystem()) {{
+
+            MockRaftActorContext context =
+                new MockRaftActorContext();
+
+            context.setLastApplied(new AtomicLong(100));
+
+            AppendEntries appendEntries =
+                new AppendEntries(100, "leader-1", 0, 0, null, 101);
+
+            RaftState raftState =
+                createBehavior(context).handleMessage(getRef(), appendEntries);
+
+            assertEquals(new AtomicLong(101).get(), context.getLastApplied().get());
+
+        }};
+    }
+
+    protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(
+        ActorRef actorRef, RaftRPC rpc){
+        RaftState raftState = createBehavior()
+            .handleMessage(actorRef, rpc);
+
+        assertEquals(RaftState.Follower, raftState);
+    }
+
+    protected abstract RaftActorBehavior createBehavior(RaftActorContext actorContext);
+
+    protected RaftActorBehavior createBehavior(){
+        return createBehavior(new MockRaftActorContext());
+    }
+
+    protected AppendEntries createAppendEntriesWithNewerTerm(){
+        return new AppendEntries(100, "leader-1", 0, 0, null, 1);
+    }
+
+    protected AppendEntriesReply createAppendEntriesReplyWithNewerTerm(){
+        return new AppendEntriesReply(100, false);
+    }
+
+    protected RequestVote createRequestVoteWithNewerTerm(){
+        return new RequestVote(100, "candidate-1", 10, 100);
+    }
+
+    protected RequestVoteReply createRequestVoteReplyWithNewerTerm(){
+        return new RequestVoteReply(100, false);
+    }
+
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/CandidateTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/CandidateTest.java
new file mode 100644
index 0000000000..6a3040d843
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/CandidateTest.java
@@ -0,0 +1,12 @@
+package org.opendaylight.controller.cluster.raft.behaviors;
+
+import org.opendaylight.controller.cluster.raft.RaftActorContext;
+
+import java.util.Collections;
+
+public class CandidateTest extends AbstractRaftActorBehaviorTest {
+
+    @Override protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
+        return new Candidate(actorContext, Collections.EMPTY_LIST);
+    }
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java
new file mode 100644
index 0000000000..cd3bd2833d
--- /dev/null
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java
@@ -0,0 +1,9 @@
+package org.opendaylight.controller.cluster.raft.behaviors;
+
+import org.opendaylight.controller.cluster.raft.RaftActorContext;
+
+public class FollowerTest extends AbstractRaftActorBehaviorTest {
+    @Override protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
+        return new Follower(actorContext);
+    }
+}
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java
index 08a79e5565..d0497f3d0a 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java
@@ -3,8 +3,8 @@ package org.opendaylight.controller.cluster.raft.behaviors;
 import akka.testkit.JavaTestKit;
 import junit.framework.Assert;
 import org.junit.Test;
-import org.opendaylight.controller.cluster.raft.AbstractActorTest;
 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftActorContext;
 import org.opendaylight.controller.cluster.raft.RaftState;
 import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
@@ -15,7 +15,7 @@ import java.util.List;
 
 import static org.junit.Assert.assertEquals;
 
-public class LeaderTest extends AbstractActorTest {
+public class LeaderTest extends AbstractRaftActorBehaviorTest {
 
     @Test
     public void testHandleMessageForUnknownMessage() throws Exception {
@@ -68,4 +68,8 @@ public class LeaderTest extends AbstractActorTest {
             };
         }};
     }
+
+    @Override protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
+        return new Leader(actorContext, Collections.EMPTY_LIST);
+    }
 }
-- 
2.36.6