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=7e6654d5d6b9849b106a36bf56c76928798d912d;hb=b8c6400766f7324dd57d059bd48e435569fe1a27;hp=afb8f29e70e2772f7244fb8ed80a4a263bb427e4;hpb=497d356874c8033e0d0f9fccfa7ed6695d6889fc;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 afb8f29e70..7e6654d5d6 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
@@ -9,7 +9,9 @@
 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;
@@ -37,12 +39,20 @@ import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  * </ul>
  */
 public class Follower extends AbstractRaftActorBehavior {
-
-    private SnapshotTracker snapshotTracker = null;
+    private static final int SYNC_THRESHOLD = 10;
 
     private final SyncStatusTracker initialSyncStatusTracker;
 
-    private static final int SYNC_THRESHOLD = 10;
+    private final Procedure<ReplicatedLogEntry> appendAndPersistCallback = new Procedure<ReplicatedLogEntry>() {
+        @Override
+        public void apply(ReplicatedLogEntry logEntry) {
+            context.getReplicatedLog().captureSnapshotIfReady(logEntry);
+        }
+    };
+
+    private SnapshotTracker snapshotTracker = null;
+    private String leaderId;
+    private short leaderPayloadVersion;
 
     public Follower(RaftActorContext context) {
         this(context, null, (short)-1);
@@ -50,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);
 
@@ -62,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){
@@ -96,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()) {
@@ -125,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
@@ -195,7 +223,7 @@ public class Follower extends AbstractRaftActorBehavior {
 
                 LOG.debug("{}: Append entry to log {}", logName(), entry.getData());
 
-                context.getReplicatedLog().appendAndPersist(entry);
+                context.getReplicatedLog().appendAndPersist(entry, appendAndPersistCallback);
 
                 if(entry.getData() instanceof ServerConfigurationPayload) {
                     context.updatePeerIds((ServerConfigurationPayload)entry.getData());
@@ -305,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;
             }
+        }
+
+        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());
 
-        } else if (message instanceof InstallSnapshot) {
-            InstallSnapshot installSnapshot = (InstallSnapshot) message;
+            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) {
@@ -420,7 +455,7 @@ public class Follower extends AbstractRaftActorBehavior {
     }
 
     @Override
-    public void close() throws Exception {
+    public void close() {
         stopElection();
     }