BUG-5626: do not allow overriding of RaftActor.handleCommand()

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

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 33ed3357d85eab27441c7eafdd6d494974d667eb..7e6654d5d6b9849b106a36bf56c76928798d912d 100644 (file)
--- 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 akka.japi.Procedure;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import java.util.ArrayList;
 import org.opendaylight.controller.cluster.raft.RaftActorContext;
 import org.opendaylight.controller.cluster.raft.RaftState;
@@ -50,6 +51,8 @@ public class Follower extends AbstractRaftActorBehavior {
     };
 
     private SnapshotTracker snapshotTracker = null;
+    private String leaderId;
+    private short leaderPayloadVersion;
 
     public Follower(RaftActorContext context) {
         this(context, null, (short)-1);
@@ -57,8 +60,8 @@ public class Follower extends AbstractRaftActorBehavior {
 
     public Follower(RaftActorContext context, String initialLeaderId, short initialLeaderPayloadVersion) {
         super(context, RaftState.Follower);
-        leaderId = initialLeaderId;
-        setLeaderPayloadVersion(initialLeaderPayloadVersion);
+        this.leaderId = initialLeaderId;
+        this.leaderPayloadVersion = initialLeaderPayloadVersion;
 
         initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), SYNC_THRESHOLD);
 
@@ -69,7 +72,26 @@ public class Follower extends AbstractRaftActorBehavior {
                 scheduleElection(electionDuration());
             }
         }
+    }
+
+    @Override
+    public final String getLeaderId() {
+        return leaderId;
+    }
 
+    @VisibleForTesting
+    protected final void setLeaderId(final String leaderId) {
+        this.leaderId = Preconditions.checkNotNull(leaderId);
+    }
+
+    @Override
+    public short getLeaderPayloadVersion() {
+        return leaderPayloadVersion;
+    }
+
+    @VisibleForTesting
+    protected final void setLeaderPayloadVersion(short leaderPayloadVersion) {
+        this.leaderPayloadVersion = leaderPayloadVersion;
     }
 
     private boolean isLogEntryPresent(long index){
@@ -103,8 +125,8 @@ public class Follower extends AbstractRaftActorBehavior {
         initialSyncStatusTracker.update(leaderId, currentLeaderCommit, context.getCommitIndex());
     }
 
-    @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
-                                                              AppendEntries appendEntries) {
+    @Override
+    protected RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries) {
 
         int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
         if(LOG.isTraceEnabled()) {
@@ -132,8 +154,7 @@ public class Follower extends AbstractRaftActorBehavior {
 
         // If we got here then we do appear to be talking to the leader
         leaderId = appendEntries.getLeaderId();
-
-        setLeaderPayloadVersion(appendEntries.getPayloadVersion());
+        leaderPayloadVersion = appendEntries.getPayloadVersion();
 
         updateInitialSyncStatus(appendEntries.getLeaderCommit(), appendEntries.getLeaderId());
         // First check if the logs are in sync or not
@@ -312,51 +333,58 @@ public class Follower extends AbstractRaftActorBehavior {
         return outOfSync;
     }
 
-    @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
+    @Override
+    protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
         AppendEntriesReply appendEntriesReply) {
         return this;
     }
 
-    @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
+    @Override
+    protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
         RequestVoteReply requestVoteReply) {
         return this;
     }
 
-    @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
-
-        Object message = fromSerializableMessage(originalMessage);
-
-        if (message instanceof RaftRPC) {
-            RaftRPC rpc = (RaftRPC) message;
-            // If RPC request or response contains term T > currentTerm:
-            // set currentTerm = T, convert to follower (§5.1)
-            // This applies to all RPC messages and responses
-            if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
-                LOG.debug("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
-                        logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
-
-                context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
-            }
-        }
-
-        if (message instanceof ElectionTimeout) {
-            if(canStartElection()) {
+    @Override
+    public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
+        if (originalMessage instanceof ElectionTimeout) {
+            if (canStartElection()) {
                 LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName());
                 return internalSwitchBehavior(RaftState.Candidate);
             } else {
                 return this;
             }
+        }
 
-        } else if (message instanceof InstallSnapshot) {
-            InstallSnapshot installSnapshot = (InstallSnapshot) message;
+        final Object message = fromSerializableMessage(originalMessage);
+        if (!(message instanceof RaftRPC)) {
+            // The rest of the processing requires the message to be a RaftRPC
+            return null;
+        }
+
+        final RaftRPC rpc = (RaftRPC) message;
+        // If RPC request or response contains term T > currentTerm:
+        // set currentTerm = T, convert to follower (§5.1)
+        // This applies to all RPC messages and responses
+        if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
+            LOG.debug("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
+                logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
+
+            context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
+        }
+
+        if (rpc instanceof InstallSnapshot) {
+            InstallSnapshot installSnapshot = (InstallSnapshot) rpc;
             handleInstallSnapshot(sender, installSnapshot);
+            scheduleElection(electionDuration());
+            return this;
         }
 
-        if(message instanceof RaftRPC && (!(message instanceof RequestVote) || (canGrantVote((RequestVote) message)))){
+        if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) {
             scheduleElection(electionDuration());
         }
 
-        return super.handleMessage(sender, message);
+        return super.handleMessage(sender, rpc);
     }
 
     private void handleInstallSnapshot(final ActorRef sender, InstallSnapshot installSnapshot) {
@@ -427,7 +455,7 @@ public class Follower extends AbstractRaftActorBehavior {
     }
 
     @Override
-    public void close() throws Exception {
+    public void close() {
         stopElection();
     }