X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FFollower.java;h=8a0788702d81f3dfe04d713bcc99e527273c82d2;hp=b1c73f6f4155e045935fa37912245e3d03d45fc9;hb=726ee824671781b5031c0108794c22bd0d96eaad;hpb=3a71a222b896b9d07e638af62300180799bdac67 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 b1c73f6f41..8a0788702d 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 @@ -45,12 +45,39 @@ public class Follower extends AbstractRaftActorBehavior { scheduleElection(electionDuration()); } + private boolean isLogEntryPresent(long index){ + if(index == context.getReplicatedLog().getSnapshotIndex()){ + return true; + } + + ReplicatedLogEntry previousEntry = context.getReplicatedLog() + .get(index); + + return previousEntry != null; + + } + + private long getLogEntryTerm(long index){ + if(index == context.getReplicatedLog().getSnapshotIndex()){ + return context.getReplicatedLog().getSnapshotTerm(); + } + + ReplicatedLogEntry previousEntry = context.getReplicatedLog() + .get(index); + + if(previousEntry != null){ + return previousEntry.getTerm(); + } + + return -1; + } + @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender, - AppendEntries appendEntries) { + AppendEntries appendEntries) { if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) { if(LOG.isDebugEnabled()) { - LOG.debug(appendEntries.toString()); + LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries); } } @@ -67,50 +94,50 @@ public class Follower extends AbstractRaftActorBehavior { // 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()); + long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex()); + boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex()); boolean outOfSync = true; // First check if the logs are in sync or not if (lastIndex() == -1 - && appendEntries.getPrevLogIndex() != -1) { + && appendEntries.getPrevLogIndex() != -1) { // The follower's log is out of sync because the leader does have // an entry at prevLogIndex and this follower has no entries in // it's log. if(LOG.isDebugEnabled()) { - LOG.debug("The followers log is empty and the senders prevLogIndex is {}", - appendEntries.getPrevLogIndex()); + LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}", + context.getId(), appendEntries.getPrevLogIndex()); } } else if (lastIndex() > -1 - && appendEntries.getPrevLogIndex() != -1 - && previousEntry == null) { + && appendEntries.getPrevLogIndex() != -1 + && !prevEntryPresent) { // The follower's log is out of sync because the Leader's // prevLogIndex entry was not found in it's log if(LOG.isDebugEnabled()) { - LOG.debug("The log is not empty but the prevLogIndex {} was not found in it", - appendEntries.getPrevLogIndex()); + LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it", + context.getId(), appendEntries.getPrevLogIndex()); } } else if (lastIndex() > -1 - && previousEntry != null - && previousEntry.getTerm()!= appendEntries.getPrevLogTerm()) { + && prevEntryPresent + && prevLogTerm != appendEntries.getPrevLogTerm()) { // The follower's log is out of sync because the Leader's // prevLogIndex entry does exist in the follower's log but it has // a different term in it - if(LOG.isDebugEnabled()) { + if (LOG.isDebugEnabled()) { LOG.debug( - "Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}" - , previousEntry.getTerm() - , appendEntries.getPrevLogTerm()); + "{}: Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}" + , context.getId(), prevLogTerm + , appendEntries.getPrevLogTerm()); } } else { outOfSync = false; @@ -120,9 +147,9 @@ public class Follower extends AbstractRaftActorBehavior { // We found that the log was out of sync so just send a negative // reply and return if(LOG.isDebugEnabled()) { - LOG.debug("Follower is out-of-sync, " + + LOG.debug("{}: Follower ({}) is out-of-sync, " + "so sending negative reply, lastIndex():{}, lastTerm():{}", - lastIndex(), lastTerm() + context.getId(), context.getId(), lastIndex(), lastTerm() ); } sender.tell( @@ -135,9 +162,8 @@ public class Follower extends AbstractRaftActorBehavior { if (appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) { if(LOG.isDebugEnabled()) { - LOG.debug( - "Number of entries to be appended = {}", appendEntries.getEntries().size() - ); + LOG.debug("{}: Number of entries to be appended = {}", context.getId(), + appendEntries.getEntries().size()); } // 3. If an existing entry conflicts with a new one (same index @@ -162,9 +188,8 @@ public class Follower extends AbstractRaftActorBehavior { } if(LOG.isDebugEnabled()) { - LOG.debug( - "Removing entries from log starting at {}", matchEntry.getIndex() - ); + LOG.debug("{}: Removing entries from log starting at {}", context.getId(), + matchEntry.getIndex()); } // Entries do not match so remove all subsequent entries @@ -175,8 +200,8 @@ public class Follower extends AbstractRaftActorBehavior { } if(LOG.isDebugEnabled()) { - LOG.debug("After cleanup entries to be added from = {}", (addEntriesFrom + lastIndex()) - ); + LOG.debug("{}: After cleanup entries to be added from = {}", context.getId(), + (addEntriesFrom + lastIndex())); } // 4. Append any new entries not already in the log @@ -184,13 +209,14 @@ public class Follower extends AbstractRaftActorBehavior { i < appendEntries.getEntries().size(); i++) { if(LOG.isDebugEnabled()) { - LOG.debug("Append entry to log {}", appendEntries.getEntries().get(i).getData()); + LOG.debug("{}: Append entry to log {}", context.getId(), + appendEntries.getEntries().get(i).getData()); } context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i)); } if(LOG.isDebugEnabled()) { - LOG.debug("Log size is now {}", context.getReplicatedLog().size()); + LOG.debug("{}: Log size is now {}", context.getId(), context.getReplicatedLog().size()); } } @@ -205,7 +231,7 @@ public class Follower extends AbstractRaftActorBehavior { if (prevCommitIndex != context.getCommitIndex()) { if(LOG.isDebugEnabled()) { - LOG.debug("Commit index set to {}", context.getCommitIndex()); + LOG.debug("{}: Commit index set to {}", context.getId(), context.getCommitIndex()); } } @@ -215,9 +241,9 @@ public class Follower extends AbstractRaftActorBehavior { if (appendEntries.getLeaderCommit() > context.getLastApplied() && context.getLastApplied() < lastIndex()) { if(LOG.isDebugEnabled()) { - LOG.debug("applyLogToStateMachine, " + + LOG.debug("{}: applyLogToStateMachine, " + "appendEntries.getLeaderCommit():{}," + - "context.getLastApplied():{}, lastIndex():{}", + "context.getLastApplied():{}, lastIndex():{}", context.getId(), appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex() ); } @@ -228,6 +254,10 @@ public class Follower extends AbstractRaftActorBehavior { sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true, lastIndex(), lastTerm()), actor()); + if (!context.isSnapshotCaptureInitiated()) { + fakeSnapshot(appendEntries.getReplicatedToAllIndex(), appendEntries.getReplicatedToAllIndex()); + } + return this; } @@ -275,8 +305,8 @@ public class Follower extends AbstractRaftActorBehavior { private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) { if(LOG.isDebugEnabled()) { - LOG.debug("InstallSnapshot received by follower " + - "datasize:{} , Chunk:{}/{}", installSnapshot.getData().size(), + LOG.debug("{}: InstallSnapshot received by follower " + + "datasize:{} , Chunk:{}/{}", context.getId(), installSnapshot.getData().size(), installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks() ); } @@ -312,8 +342,7 @@ public class Follower extends AbstractRaftActorBehavior { snapshotTracker = null; } catch (Exception e){ - - LOG.error(e, "Exception in InstallSnapshot of follower:"); + LOG.error("{}: Exception in InstallSnapshot of follower", context.getId(), e); //send reply with success as false. The chunk will be sent again on failure sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), installSnapshot.getChunkIndex(), false), actor());