[Fix for Bug 1631]

[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / behaviors / AbstractRaftActorBehavior.java

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 35d563b784cf3f4705784a78952249ce06badbea..eed74bba82fbf8a4f8e4c1b5431887451c49d59d 100644 (file)
--- 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
@@ -10,11 +10,12 @@ package org.opendaylight.controller.cluster.raft.behaviors;
 
 import akka.actor.ActorRef;
 import akka.actor.Cancellable;
+import akka.event.LoggingAdapter;
 import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
 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.SerializationUtils;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
@@ -43,6 +44,11 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      */
     protected final RaftActorContext context;
 
+    /**
+     *
+     */
+    protected final LoggingAdapter LOG;
+
     /**
      *
      */
@@ -56,6 +62,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
 
     protected AbstractRaftActorBehavior(RaftActorContext context) {
         this.context = context;
+        this.LOG = context.getLogger();
     }
 
     /**
@@ -70,7 +77,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @param appendEntries  The AppendEntries message
      * @return
      */
-    protected abstract RaftState handleAppendEntries(ActorRef sender,
+    protected abstract RaftActorBehavior handleAppendEntries(ActorRef sender,
         AppendEntries appendEntries);
 
 
@@ -82,19 +89,21 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @param appendEntries
      * @return
      */
-    protected RaftState appendEntries(ActorRef sender,
+    protected RaftActorBehavior appendEntries(ActorRef sender,
         AppendEntries appendEntries) {
 
         // 1. Reply false if term < currentTerm (§5.1)
         if (appendEntries.getTerm() < currentTerm()) {
-            context.getLogger().debug(
-                "Cannot append entries because sender term " + appendEntries
-                    .getTerm() + " is less than " + currentTerm());
+            if(LOG.isDebugEnabled()) {
+                LOG.debug("Cannot append entries because sender term {} is less than {}",
+                        appendEntries.getTerm(), currentTerm());
+            }
+
             sender.tell(
                 new AppendEntriesReply(context.getId(), currentTerm(), false,
                     lastIndex(), lastTerm()), actor()
             );
-            return state();
+            return this;
         }
 
 
@@ -113,7 +122,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @param appendEntriesReply The AppendEntriesReply message
      * @return
      */
-    protected abstract RaftState handleAppendEntriesReply(ActorRef sender,
+    protected abstract RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
         AppendEntriesReply appendEntriesReply);
 
     /**
@@ -124,11 +133,12 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @param requestVote
      * @return
      */
-    protected RaftState requestVote(ActorRef sender,
+    protected RaftActorBehavior requestVote(ActorRef sender,
         RequestVote requestVote) {
 
-
-        context.getLogger().debug(requestVote.toString());
+        if(LOG.isDebugEnabled()) {
+            LOG.debug(requestVote.toString());
+        }
 
         boolean grantVote = false;
 
@@ -166,7 +176,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
 
         sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor());
 
-        return state();
+        return this;
     }
 
     /**
@@ -181,7 +191,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
      * @param requestVoteReply The RequestVoteReply message
      * @return
      */
-    protected abstract RaftState handleRequestVoteReply(ActorRef sender,
+    protected abstract RaftActorBehavior handleRequestVoteReply(ActorRef sender,
         RequestVoteReply requestVoteReply);
 
     /**
@@ -340,13 +350,21 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
             } else {
                 //if one index is not present in the log, no point in looping
                 // around as the rest wont be present either
-                context.getLogger().warning(
-                    "Missing index {} from log. Cannot apply state. Ignoring {} to {}", i, i, index );
+                LOG.warning(
+                        "Missing index {} from log. Cannot apply state. Ignoring {} to {}", i, i, index);
                 break;
             }
         }
-        context.getLogger().debug("Setting last applied to {}", newLastApplied);
+        if(LOG.isDebugEnabled()) {
+            LOG.debug("Setting last applied to {}", newLastApplied);
+        }
         context.setLastApplied(newLastApplied);
+
+        // send a message to persist a ApplyLogEntries marker message into akka's persistent journal
+        // will be used during recovery
+        //in case if the above code throws an error and this message is not sent, it would be fine
+        // as the  append entries received later would initiate add this message to the journal
+        actor().tell(new ApplyLogEntries((int) context.getLastApplied()), actor());
     }
 
     protected Object fromSerializableMessage(Object serializable){
@@ -354,7 +372,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
     }
 
     @Override
-    public RaftState handleMessage(ActorRef sender, Object message) {
+    public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
         if (message instanceof AppendEntries) {
             return appendEntries(sender, (AppendEntries) message);
         } else if (message instanceof AppendEntriesReply) {
@@ -364,10 +382,21 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior {
         } else if (message instanceof RequestVoteReply) {
             return handleRequestVoteReply(sender, (RequestVoteReply) message);
         }
-        return state();
+        return this;
     }
 
     @Override public String getLeaderId() {
         return leaderId;
     }
+
+    protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
+        LOG.info("Switching from behavior {} to {}", this.state(), behavior.state());
+        try {
+            close();
+        } catch (Exception e) {
+            LOG.error(e, "Failed to close behavior : {}", this.state());
+        }
+
+        return behavior;
+    }
 }