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%2FAbstractLeader.java;h=fab1714989a99407f3d51452d33ef313ac2c14e8;hp=5076a8a38f891c894b9e01fdc61db5c0d5fa11a9;hb=660c3e22ca97bc613ea6f6288503620bba6fb233;hpb=95d3c7975a423951dcbdecfbfa4cb6b7a23591cc diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java index 5076a8a38f..fab1714989 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java @@ -15,7 +15,6 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.protobuf.ByteString; -import java.io.IOException; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -61,15 +60,14 @@ import scala.concurrent.duration.FiniteDuration; * respond after entry applied to state machine (§5.3) *
  • If last log index ≥ nextIndex for a follower: send * AppendEntries RPC with log entries starting at nextIndex - * *
  • If there exists an N such that N > commitIndex, a majority * of matchIndex[i] ≥ N, and log[N].term == currentTerm: * set commitIndex = N (§5.3, §5.4). + * */ public abstract class AbstractLeader extends AbstractRaftActorBehavior { private final Map followerToLog = new HashMap<>(); @@ -77,32 +75,31 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { /** * Lookup table for request contexts based on journal index. We could use a {@link Map} here, but we really * expect the entries to be modified in sequence, hence we open-code the lookup. - * * TODO: Evaluate the use of ArrayDeque(), as that has lower memory overhead. Non-head removals are more costly, * but we already expect those to be far from frequent. */ private final Queue trackers = new LinkedList<>(); private Cancellable heartbeatSchedule = null; - private Optional snapshot = Optional.absent();; + private Optional snapshot = Optional.absent(); private int minReplicationCount; protected AbstractLeader(RaftActorContext context, RaftState state, @Nullable AbstractLeader initializeFromLeader) { super(context, state); - if(initializeFromLeader != null) { + if (initializeFromLeader != null) { followerToLog.putAll(initializeFromLeader.followerToLog); snapshot = initializeFromLeader.snapshot; trackers.addAll(initializeFromLeader.trackers); } else { - for(PeerInfo peerInfo: context.getPeers()) { + for (PeerInfo peerInfo: context.getPeers()) { FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(peerInfo, -1, context); followerToLog.put(peerInfo.getId(), followerLogInformation); } } - LOG.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds()); + log.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds()); updateMinReplicaCount(); @@ -134,7 +131,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { context.getPeerInfo(followerId), -1, context); followerToLog.put(followerId, followerLogInformation); - if(heartbeatSchedule == null) { + if (heartbeatSchedule == null) { scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); } } @@ -145,8 +142,8 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { public void updateMinReplicaCount() { int numVoting = 0; - for(PeerInfo peer: context.getPeers()) { - if(peer.isVoting()) { + for (PeerInfo peer: context.getPeers()) { + if (peer.isVoting()) { numVoting++; } } @@ -154,7 +151,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { minReplicationCount = getMajorityVoteCount(numVoting); } - protected int getMinIsolatedLeaderPeerCount(){ + protected int getMinIsolatedLeaderPeerCount() { //the isolated Leader peer count will be 1 less than the majority vote count. //this is because the vote count has the self vote counted in it //for e.g @@ -162,12 +159,12 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { //2 peers = 2 votesRequired , minIsolatedLeaderPeerCount = 1 //4 peers = 3 votesRequired, minIsolatedLeaderPeerCount = 2 - return minReplicationCount > 0 ? (minReplicationCount - 1) : 0; + return minReplicationCount > 0 ? minReplicationCount - 1 : 0; } @VisibleForTesting void setSnapshot(@Nullable Snapshot snapshot) { - if(snapshot != null) { + if (snapshot != null) { this.snapshot = Optional.of(new SnapshotHolder(snapshot)); } else { this.snapshot = Optional.absent(); @@ -183,33 +180,28 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { protected RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries) { - LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries); + log.debug("{}: handleAppendEntries: {}", logName(), appendEntries); return this; } @Override - protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender, - AppendEntriesReply appendEntriesReply) { - - if(LOG.isTraceEnabled()) { - LOG.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply); - } + protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply) { + log.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply); // Update the FollowerLogInformation String followerId = appendEntriesReply.getFollowerId(); - FollowerLogInformation followerLogInformation = - followerToLog.get(followerId); + FollowerLogInformation followerLogInformation = followerToLog.get(followerId); - if(followerLogInformation == null){ - LOG.error("{}: handleAppendEntriesReply - unknown follower {}", logName(), followerId); + if (followerLogInformation == null) { + log.error("{}: handleAppendEntriesReply - unknown follower {}", logName(), followerId); return this; } - if(followerLogInformation.timeSinceLastActivity() > - context.getConfigParams().getElectionTimeOutInterval().toMillis()) { - LOG.warn("{} : handleAppendEntriesReply delayed beyond election timeout, " + - "appendEntriesReply : {}, timeSinceLastActivity : {}, lastApplied : {}, commitIndex : {}", + if (followerLogInformation.timeSinceLastActivity() + > context.getConfigParams().getElectionTimeOutInterval().toMillis()) { + log.warn("{} : handleAppendEntriesReply delayed beyond election timeout, " + + "appendEntriesReply : {}, timeSinceLastActivity : {}, lastApplied : {}, commitIndex : {}", logName(), appendEntriesReply, followerLogInformation.timeSinceLastActivity(), context.getLastApplied(), context.getCommitIndex()); } @@ -218,8 +210,10 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerLogInformation.setPayloadVersion(appendEntriesReply.getPayloadVersion()); followerLogInformation.setRaftVersion(appendEntriesReply.getRaftVersion()); + long followerLastLogIndex = appendEntriesReply.getLogLastIndex(); + long followersLastLogTermInLeadersLog = getLogEntryTerm(followerLastLogIndex); boolean updated = false; - if(appendEntriesReply.getLogLastIndex() > context.getReplicatedLog().lastIndex()) { + if (appendEntriesReply.getLogLastIndex() > context.getReplicatedLog().lastIndex()) { // The follower's log is actually ahead of the leader's log. Normally this doesn't happen // in raft as a node cannot become leader if it's log is behind another's. However, the // non-voting semantics deviate a bit from raft. Only voting members participate in @@ -233,23 +227,41 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // snapshot. It's also possible that the follower's last log index is behind the leader's. // However in this case the log terms won't match and the logs will conflict - this is handled // elsewhere. - LOG.debug("{}: handleAppendEntriesReply: follower {} lastIndex {} is ahead of our lastIndex {} - forcing install snaphot", - logName(), followerLogInformation.getId(), appendEntriesReply.getLogLastIndex(), - context.getReplicatedLog().lastIndex()); + log.debug("{}: handleAppendEntriesReply: follower {} lastIndex {} is ahead of our lastIndex {} - " + + "forcing install snaphot", logName(), followerLogInformation.getId(), + appendEntriesReply.getLogLastIndex(), context.getReplicatedLog().lastIndex()); followerLogInformation.setMatchIndex(-1); followerLogInformation.setNextIndex(-1); initiateCaptureSnapshot(followerId); + updated = true; } else if (appendEntriesReply.isSuccess()) { - updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply); + if (followerLastLogIndex >= 0 && followersLastLogTermInLeadersLog >= 0 + && followersLastLogTermInLeadersLog != appendEntriesReply.getLogLastTerm()) { + // The follower's last entry is present in the leader's journal but the terms don't match so the + // follower has a conflicting entry. Since the follower didn't report that it's out of sync, this means + // either the previous leader entry sent didn't conflict or the previous leader entry is in the snapshot + // and no longer in the journal. Either way, we set the follower's next index to 1 less than the last + // index reported by the follower. For the former case, the leader will send all entries starting with + // the previous follower's index and the follower will remove and replace the conflicting entries as + // needed. For the latter, the leader will initiate an install snapshot. + + followerLogInformation.setNextIndex(followerLastLogIndex - 1); + updated = true; + + log.debug("{}: handleAppendEntriesReply: follower {} last log term {} for index {} conflicts with the " + + "leader's {} - set the follower's next index to {}", logName(), + followerId, appendEntriesReply.getLogLastTerm(), appendEntriesReply.getLogLastIndex(), + followersLastLogTermInLeadersLog, followerLogInformation.getNextIndex()); + } else { + updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply); + } } else { - LOG.debug("{}: handleAppendEntriesReply: received unsuccessful reply: {}", logName(), appendEntriesReply); + log.debug("{}: handleAppendEntriesReply: received unsuccessful reply: {}", logName(), appendEntriesReply); - long followerLastLogIndex = appendEntriesReply.getLogLastIndex(); - long followersLastLogTerm = getLogEntryTerm(followerLastLogIndex); - if(appendEntriesReply.isForceInstallSnapshot()) { + if (appendEntriesReply.isForceInstallSnapshot()) { // Reset the followers match and next index. This is to signal that this follower has nothing // in common with this Leader and so would require a snapshot to be installed followerLogInformation.setMatchIndex(-1); @@ -257,8 +269,8 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // Force initiate a snapshot capture initiateCaptureSnapshot(followerId); - } else if(followerLastLogIndex < 0 || (followersLastLogTerm >= 0 && - followersLastLogTerm == appendEntriesReply.getLogLastTerm())) { + } else if (followerLastLogIndex < 0 || followersLastLogTermInLeadersLog >= 0 + && followersLastLogTermInLeadersLog == appendEntriesReply.getLogLastTerm()) { // The follower's log is empty or the last entry is present in the leader's journal // and the terms match so the follower is just behind the leader's journal from // the last snapshot, if any. We'll catch up the follower quickly by starting at the @@ -269,10 +281,12 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // The follower's log conflicts with leader's log so decrement follower's next index by 1 // in an attempt to find where the logs match. - LOG.debug("{}: follower's last log term {} conflicts with the leader's {} - dec next index", - logName(), appendEntriesReply.getLogLastTerm(), followersLastLogTerm); - followerLogInformation.decrNextIndex(); + updated = true; + + log.debug("{}: follower's last log term {} conflicts with the leader's {} - dec next index to {}", + logName(), appendEntriesReply.getLogLastTerm(), followersLastLogTermInLeadersLog, + followerLogInformation.getNextIndex()); } } @@ -280,34 +294,36 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // If there exists an N such that N > commitIndex, a majority // of matchIndex[i] ≥ N, and log[N].term == currentTerm: // set commitIndex = N (§5.3, §5.4). - if(LOG.isTraceEnabled()) { - LOG.trace("{}: handleAppendEntriesReply from {}: commitIndex: {}, lastAppliedIndex: {}, currentTerm: {}", + if (log.isTraceEnabled()) { + log.trace("{}: handleAppendEntriesReply from {}: commitIndex: {}, lastAppliedIndex: {}, currentTerm: {}", logName(), followerId, context.getCommitIndex(), context.getLastApplied(), currentTerm()); } - for (long N = context.getCommitIndex() + 1; ; N++) { + for (long index = context.getCommitIndex() + 1; ; index++) { int replicatedCount = 1; - LOG.trace("{}: checking Nth index {}", logName(), N); + log.trace("{}: checking Nth index {}", logName(), index); for (FollowerLogInformation info : followerToLog.values()) { final PeerInfo peerInfo = context.getPeerInfo(info.getId()); - if(info.getMatchIndex() >= N && peerInfo != null && peerInfo.isVoting()) { + if (info.getMatchIndex() >= index && peerInfo != null && peerInfo.isVoting()) { replicatedCount++; - } else if(LOG.isTraceEnabled()) { - LOG.trace("{}: Not counting follower {} - matchIndex: {}, {}", logName(), info.getId(), + } else if (log.isTraceEnabled()) { + log.trace("{}: Not counting follower {} - matchIndex: {}, {}", logName(), info.getId(), info.getMatchIndex(), peerInfo); } } - if(LOG.isTraceEnabled()) { - LOG.trace("{}: replicatedCount {}, minReplicationCount: {}", logName(), replicatedCount, minReplicationCount); + if (log.isTraceEnabled()) { + log.trace("{}: replicatedCount {}, minReplicationCount: {}", logName(), replicatedCount, + minReplicationCount); } if (replicatedCount >= minReplicationCount) { - ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(N); + ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(index); if (replicatedLogEntry == null) { - LOG.debug("{}: ReplicatedLogEntry not found for index {} - snapshotIndex: {}, journal size: {}", - logName(), N, context.getReplicatedLog().getSnapshotIndex(), context.getReplicatedLog().size()); + log.debug("{}: ReplicatedLogEntry not found for index {} - snapshotIndex: {}, journal size: {}", + logName(), index, context.getReplicatedLog().getSnapshotIndex(), + context.getReplicatedLog().size()); break; } @@ -317,23 +333,25 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // reach consensus, as per §5.4.1: "once an entry from the current term is committed by // counting replicas, then all prior entries are committed indirectly". if (replicatedLogEntry.getTerm() == currentTerm()) { - LOG.trace("{}: Setting commit index to {}", logName(), N); - context.setCommitIndex(N); + log.trace("{}: Setting commit index to {}", logName(), index); + context.setCommitIndex(index); } else { - LOG.debug("{}: Not updating commit index to {} - retrieved log entry with index {}, term {} does not match the current term {}", - logName(), N, replicatedLogEntry.getIndex(), replicatedLogEntry.getTerm(), currentTerm()); + log.debug("{}: Not updating commit index to {} - retrieved log entry with index {}, " + + "term {} does not match the current term {}", logName(), index, + replicatedLogEntry.getIndex(), replicatedLogEntry.getTerm(), currentTerm()); } } else { - LOG.trace("{}: minReplicationCount not reached, actual {} - breaking", logName(), replicatedCount); + log.trace("{}: minReplicationCount not reached, actual {} - breaking", logName(), replicatedCount); break; } } // Apply the change to the state machine if (context.getCommitIndex() > context.getLastApplied()) { - if(LOG.isDebugEnabled()) { - LOG.debug("{}: handleAppendEntriesReply from {}: applying to log - commitIndex: {}, lastAppliedIndex: {}", - logName(), followerId, context.getCommitIndex(), context.getLastApplied()); + if (log.isDebugEnabled()) { + log.debug( + "{}: handleAppendEntriesReply from {}: applying to log - commitIndex: {}, lastAppliedIndex: {}", + logName(), followerId, context.getCommitIndex(), context.getLastApplied()); } applyLogToStateMachine(context.getCommitIndex()); @@ -354,10 +372,11 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { boolean updated = followerLogInformation.setMatchIndex(appendEntriesReply.getLogLastIndex()); updated = followerLogInformation.setNextIndex(appendEntriesReply.getLogLastIndex() + 1) || updated; - if(updated && LOG.isDebugEnabled()) { - LOG.debug("{}: handleAppendEntriesReply - FollowerLogInformation for {} updated: matchIndex: {}, nextIndex: {}", - logName(), followerLogInformation.getId(), followerLogInformation.getMatchIndex(), - followerLogInformation.getNextIndex()); + if (updated && log.isDebugEnabled()) { + log.debug( + "{}: handleAppendEntriesReply - FollowerLogInformation for {} updated: matchIndex: {}, nextIndex: {}", + logName(), followerLogInformation.getId(), followerLogInformation.getMatchIndex(), + followerLogInformation.getNextIndex()); } return updated; } @@ -407,7 +426,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // 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 leader's term {} - switching to Follower", + log.debug("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); @@ -420,7 +439,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { beforeSendHeartbeat(); sendHeartBeat(); scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); - } else if(message instanceof SendInstallSnapshot) { + } else if (message instanceof SendInstallSnapshot) { // received from RaftActor setSnapshot(((SendInstallSnapshot) message).getSnapshot()); sendInstallSnapshot(); @@ -436,20 +455,20 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } private void handleInstallSnapshotReply(InstallSnapshotReply reply) { - LOG.debug("{}: handleInstallSnapshotReply: {}", logName(), reply); + log.debug("{}: handleInstallSnapshotReply: {}", logName(), reply); String followerId = reply.getFollowerId(); FollowerLogInformation followerLogInformation = followerToLog.get(followerId); - if(followerLogInformation == null) { + if (followerLogInformation == null) { // This can happen during AddServer if it times out. - LOG.error("{}: FollowerLogInformation not found for follower {} in InstallSnapshotReply", + log.error("{}: FollowerLogInformation not found for follower {} in InstallSnapshotReply", logName(), followerId); return; } LeaderInstallSnapshotState installSnapshotState = followerLogInformation.getInstallSnapshotState(); if (installSnapshotState == null) { - LOG.error("{}: LeaderInstallSnapshotState not found for follower {} in InstallSnapshotReply", + log.error("{}: LeaderInstallSnapshotState not found for follower {} in InstallSnapshotReply", logName(), followerId); return; } @@ -459,22 +478,18 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if (installSnapshotState.getChunkIndex() == reply.getChunkIndex()) { boolean wasLastChunk = false; if (reply.isSuccess()) { - if(installSnapshotState.isLastChunk(reply.getChunkIndex())) { + if (installSnapshotState.isLastChunk(reply.getChunkIndex())) { //this was the last chunk reply - if(LOG.isDebugEnabled()) { - LOG.debug("{}: InstallSnapshotReply received, " + - "last chunk received, Chunk: {}. Follower: {} Setting nextIndex: {}", - logName(), reply.getChunkIndex(), followerId, - context.getReplicatedLog().getSnapshotIndex() + 1 - ); - } + log.debug("{}: InstallSnapshotReply received, last chunk received, Chunk: {}. Follower: {} -" + + " Setting nextIndex: {}", logName(), reply.getChunkIndex(), followerId, + context.getReplicatedLog().getSnapshotIndex() + 1); long followerMatchIndex = snapshot.get().getLastIncludedIndex(); followerLogInformation.setMatchIndex(followerMatchIndex); followerLogInformation.setNextIndex(followerMatchIndex + 1); followerLogInformation.clearLeaderInstallSnapshotState(); - LOG.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}", + log.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}", logName(), followerId, followerLogInformation.getMatchIndex(), followerLogInformation.getNextIndex()); @@ -485,17 +500,17 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } wasLastChunk = true; - if(context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED){ + if (context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED) { UnInitializedFollowerSnapshotReply unInitFollowerSnapshotSuccess = new UnInitializedFollowerSnapshotReply(followerId); context.getActor().tell(unInitFollowerSnapshotSuccess, context.getActor()); - LOG.debug("Sent message UnInitializedFollowerSnapshotReply to self"); + log.debug("Sent message UnInitializedFollowerSnapshotReply to self"); } } else { installSnapshotState.markSendStatus(true); } } else { - LOG.info("{}: InstallSnapshotReply received sending snapshot chunk failed, Will retry, Chunk: {}", + log.info("{}: InstallSnapshotReply received sending snapshot chunk failed, Will retry, Chunk: {}", logName(), reply.getChunkIndex()); installSnapshotState.markSendStatus(false); @@ -506,17 +521,17 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { purgeInMemoryLog(); } else if (!wasLastChunk && installSnapshotState.canSendNextChunk()) { ActorSelection followerActor = context.getPeerActorSelection(followerId); - if(followerActor != null) { - sendSnapshotChunk(followerActor, followerId); + if (followerActor != null) { + sendSnapshotChunk(followerActor, followerLogInformation); } } } else { - LOG.error("{}: Chunk index {} in InstallSnapshotReply from follower {} does not match expected index {}", + log.error("{}: Chunk index {} in InstallSnapshotReply from follower {} does not match expected index {}", logName(), reply.getChunkIndex(), followerId, installSnapshotState.getChunkIndex()); - if(reply.getChunkIndex() == LeaderInstallSnapshotState.INVALID_CHUNK_INDEX){ + if (reply.getChunkIndex() == LeaderInstallSnapshotState.INVALID_CHUNK_INDEX) { // Since the Follower did not find this index to be valid we should reset the follower snapshot // so that Installing the snapshot can resume from the beginning installSnapshotState.reset(); @@ -525,8 +540,8 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } private boolean anyFollowersInstallingSnapshot() { - for(FollowerLogInformation info: followerToLog.values()) { - if(info.getInstallSnapshotState() != null) { + for (FollowerLogInformation info: followerToLog.values()) { + if (info.getInstallSnapshotState() != null) { return true; } @@ -538,12 +553,12 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { private void replicate(Replicate replicate) { long logIndex = replicate.getReplicatedLogEntry().getIndex(); - LOG.debug("{}: Replicate message: identifier: {}, logIndex: {}, payload: {}", logName(), + log.debug("{}: Replicate message: identifier: {}, logIndex: {}, payload: {}", logName(), replicate.getIdentifier(), logIndex, replicate.getReplicatedLogEntry().getData().getClass()); // Create a tracker entry we will use this later to notify the // client actor - if(replicate.getClientActor() != null) { + if (replicate.getClientActor() != null) { trackers.add(new ClientRequestTrackerImpl(replicate.getClientActor(), replicate.getIdentifier(), logIndex)); } @@ -551,7 +566,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { boolean applyModificationToState = !context.anyVotingPeers() || context.getRaftPolicy().applyModificationToStateBeforeConsensus(); - if(applyModificationToState){ + if (applyModificationToState) { context.setCommitIndex(logIndex); applyLogToStateMachine(logIndex); } @@ -567,20 +582,19 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { final String followerId = e.getKey(); final FollowerLogInformation followerLogInformation = e.getValue(); // This checks helps not to send a repeat message to the follower - if(!followerLogInformation.isFollowerActive() || - followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) { + if (!followerLogInformation.isFollowerActive() + || followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) { sendUpdatesToFollower(followerId, followerLogInformation, true, isHeartbeat); } } } /** - * * This method checks if any update needs to be sent to the given follower. This includes append log entries, * sending next snapshot chunk, and initiating a snapshot. + * * @return true if any update is sent, false otherwise */ - private void sendUpdatesToFollower(String followerId, FollowerLogInformation followerLogInformation, boolean sendHeartbeat, boolean isHeartbeat) { @@ -595,8 +609,8 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if (installSnapshotState != null) { // if install snapshot is in process , then sent next chunk if possible if (isFollowerActive && installSnapshotState.canSendNextChunk()) { - sendSnapshotChunk(followerActor, followerId); - } else if(sendHeartbeat) { + sendSnapshotChunk(followerActor, followerLogInformation); + } else if (sendHeartbeat) { // we send a heartbeat even if we have not received a reply for the last chunk sendAppendEntries = true; } @@ -604,17 +618,18 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { long leaderLastIndex = context.getReplicatedLog().lastIndex(); long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex(); - if((!isHeartbeat && LOG.isDebugEnabled()) || LOG.isTraceEnabled()) { - LOG.debug("{}: Checking sendAppendEntries for follower {}: active: {}, followerNextIndex: {}, leaderLastIndex: {}, leaderSnapShotIndex: {}", - logName(), followerId, isFollowerActive, followerNextIndex, leaderLastIndex, leaderSnapShotIndex); + if (!isHeartbeat && log.isDebugEnabled() || log.isTraceEnabled()) { + log.debug("{}: Checking sendAppendEntries for follower {}: active: {}, followerNextIndex: {}, " + + "leaderLastIndex: {}, leaderSnapShotIndex: {}", logName(), followerId, isFollowerActive, + followerNextIndex, leaderLastIndex, leaderSnapShotIndex); } if (isFollowerActive && context.getReplicatedLog().isPresent(followerNextIndex)) { - LOG.debug("{}: sendAppendEntries: {} is present for follower {}", logName(), + log.debug("{}: sendAppendEntries: {} is present for follower {}", logName(), followerNextIndex, followerId); - if(followerLogInformation.okToReplicate()) { + if (followerLogInformation.okToReplicate()) { // Try to send all the entries in the journal but not exceeding the max data size // for a single AppendEntries message. int maxEntries = (int) context.getReplicatedLog().size(); @@ -622,16 +637,16 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { context.getConfigParams().getSnapshotChunkSize()); sendAppendEntries = true; } - } else if (isFollowerActive && followerNextIndex >= 0 && - leaderLastIndex > followerNextIndex && !context.getSnapshotManager().isCapturing()) { + } else if (isFollowerActive && followerNextIndex >= 0 + && leaderLastIndex > followerNextIndex && !context.getSnapshotManager().isCapturing()) { // if the followers next index is not present in the leaders log, and // if the follower is just not starting and if leader's index is more than followers index // then snapshot should be sent - if (LOG.isDebugEnabled()) { - LOG.debug(String.format("%s: InitiateInstallSnapshot to follower: %s, " + - "follower-nextIndex: %d, leader-snapshot-index: %d, " + - "leader-last-index: %d", logName(), followerId, + if (log.isDebugEnabled()) { + log.debug(String.format("%s: InitiateInstallSnapshot to follower: %s, " + + "follower-nextIndex: %d, leader-snapshot-index: %d, " + + "leader-last-index: %d", logName(), followerId, followerNextIndex, leaderSnapShotIndex, leaderLastIndex)); } @@ -641,7 +656,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { initiateCaptureSnapshot(followerId); } - } else if(sendHeartbeat) { + } else if (sendHeartbeat) { // we send an AppendEntries, even if the follower is inactive // in-order to update the followers timestamp, in case it becomes active again sendAppendEntries = true; @@ -649,22 +664,33 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } - if(sendAppendEntries) { - sendAppendEntriesToFollower(followerActor, followerNextIndex, - entries, followerId); + if (sendAppendEntries) { + sendAppendEntriesToFollower(followerActor, entries, followerLogInformation); } } } - private void sendAppendEntriesToFollower(ActorSelection followerActor, long followerNextIndex, - List entries, String followerId) { + private void sendAppendEntriesToFollower(ActorSelection followerActor, List entries, + FollowerLogInformation followerLogInformation) { + // In certain cases outlined below we don't want to send the actual commit index to prevent the follower from + // possibly committing and applying conflicting entries (those with same index, different term) from a prior + // term that weren't replicated to a majority, which would be a violation of raft. + // - if the follower isn't active. In this case we don't know the state of the follower and we send an + // empty AppendEntries as a heart beat to prevent election. + // - if we're in the process of installing a snapshot. In this case we don't send any new entries but still + // need to send AppendEntries to prevent election. + boolean isInstallingSnaphot = followerLogInformation.getInstallSnapshotState() != null; + long leaderCommitIndex = isInstallingSnaphot || !followerLogInformation.isFollowerActive() ? -1 : + context.getCommitIndex(); + + long followerNextIndex = followerLogInformation.getNextIndex(); AppendEntries appendEntries = new AppendEntries(currentTerm(), context.getId(), getLogEntryIndex(followerNextIndex - 1), getLogEntryTerm(followerNextIndex - 1), entries, - context.getCommitIndex(), super.getReplicatedToAllIndex(), context.getPayloadVersion()); + leaderCommitIndex, super.getReplicatedToAllIndex(), context.getPayloadVersion()); - if(!entries.isEmpty() || LOG.isTraceEnabled()) { - LOG.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerId, + if (!entries.isEmpty() || log.isTraceEnabled()) { + log.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerLogInformation.getId(), appendEntries); } @@ -672,45 +698,57 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } /** + * Initiates a snapshot capture to install on a follower. + *

    * Install Snapshot works as follows - * 1. Leader initiates the capture snapshot by sending a CaptureSnapshot message to actor - * 2. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log - * and makes a call to Leader's handleMessage , with SendInstallSnapshot message. - * 3. Leader , picks the snapshot from im-mem ReplicatedLog and sends it in chunks to the Follower - * 4. On complete, Follower sends back a InstallSnapshotReply. - * 5. On receipt of the InstallSnapshotReply for the last chunk, Leader marks the install complete for that follower - * and replenishes the memory by deleting the snapshot in Replicated log. - * 6. If another follower requires a snapshot and a snapshot has been collected (via CaptureSnapshotReply) - * then send the existing snapshot in chunks to the follower. - * @param followerId + * 1. Leader initiates the capture snapshot by calling createSnapshot on the RaftActor. + * 2. On receipt of the CaptureSnapshotReply message, the RaftActor persists the snapshot and makes a call to + * the Leader's handleMessage with a SendInstallSnapshot message. + * 3. The Leader obtains and stores the Snapshot from the SendInstallSnapshot message and sends it in chunks to + * the Follower via InstallSnapshot messages. + * 4. For each chunk, the Follower sends back an InstallSnapshotReply. + * 5. On receipt of the InstallSnapshotReply for the last chunk, the Leader marks the install complete for that + * follower. + * 6. If another follower requires a snapshot and a snapshot has been collected (via SendInstallSnapshot) + * then send the existing snapshot in chunks to the follower. + * + * @param followerId the id of the follower. + * @return true if capture was initiated, false otherwise. */ public boolean initiateCaptureSnapshot(String followerId) { + FollowerLogInformation followerLogInfo = followerToLog.get(followerId); if (snapshot.isPresent()) { - // if a snapshot is present in the memory, most likely another install is in progress - // no need to capture snapshot. - // This could happen if another follower needs an install when one is going on. + // If a snapshot is present in the memory, most likely another install is in progress no need to capture + // snapshot. This could happen if another follower needs an install when one is going on. final ActorSelection followerActor = context.getPeerActorSelection(followerId); - sendSnapshotChunk(followerActor, followerId); + + // Note: sendSnapshotChunk will set the LeaderInstallSnapshotState. + sendSnapshotChunk(followerActor, followerLogInfo); return true; } else { - return context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(), + boolean captureInitiated = context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(), this.getReplicatedToAllIndex(), followerId); + if (captureInitiated) { + followerLogInfo.setLeaderInstallSnapshotState(new LeaderInstallSnapshotState( + context.getConfigParams().getSnapshotChunkSize(), logName())); + } + + return captureInitiated; } } - private boolean canInstallSnapshot(long nextIndex){ + private boolean canInstallSnapshot(long nextIndex) { // If the follower's nextIndex is -1 then we might as well send it a snapshot // Otherwise send it a snapshot only if the nextIndex is not present in the log but is present // in the snapshot - return nextIndex == -1 || - (!context.getReplicatedLog().isPresent(nextIndex) - && context.getReplicatedLog().isInSnapshot(nextIndex)); + return nextIndex == -1 || !context.getReplicatedLog().isPresent(nextIndex) + && context.getReplicatedLog().isInSnapshot(nextIndex); } private void sendInstallSnapshot() { - LOG.debug("{}: sendInstallSnapshot", logName()); + log.debug("{}: sendInstallSnapshot", logName()); for (Entry e : followerToLog.entrySet()) { String followerId = e.getKey(); ActorSelection followerActor = context.getPeerActorSelection(followerId); @@ -718,9 +756,10 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if (followerActor != null) { long nextIndex = followerLogInfo.getNextIndex(); - if (context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED || - canInstallSnapshot(nextIndex)) { - sendSnapshotChunk(followerActor, followerId); + if (followerLogInfo.getInstallSnapshotState() != null + || context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED + || canInstallSnapshot(nextIndex)) { + sendSnapshotChunk(followerActor, followerLogInfo); } } } @@ -730,66 +769,50 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { * Sends a snapshot chunk to a given follower * InstallSnapshot should qualify as a heartbeat too. */ - private void sendSnapshotChunk(ActorSelection followerActor, String followerId) { - try { - if (snapshot.isPresent()) { - byte[] nextSnapshotChunk = getNextSnapshotChunk(followerId, snapshot.get().getSnapshotBytes()); - - // Note: the previous call to getNextSnapshotChunk has the side-effect of adding - // followerId to the followerToSnapshot map. - LeaderInstallSnapshotState installSnapshotState = followerToLog.get(followerId).getInstallSnapshotState(); - - int nextChunkIndex = installSnapshotState.incrementChunkIndex(); - Optional serverConfig = Optional.absent(); - if(installSnapshotState.isLastChunk(nextChunkIndex)) { - serverConfig = Optional.fromNullable(context.getPeerServerInfo(true)); - } - - followerActor.tell( - new InstallSnapshot(currentTerm(), context.getId(), - snapshot.get().getLastIncludedIndex(), - snapshot.get().getLastIncludedTerm(), - nextSnapshotChunk, - nextChunkIndex, - installSnapshotState.getTotalChunks(), - Optional.of(installSnapshotState.getLastChunkHashCode()), - serverConfig - ).toSerializable(followerToLog.get(followerId).getRaftVersion()), - actor() - ); - - if(LOG.isDebugEnabled()) { - LOG.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", - logName(), followerActor.path(), installSnapshotState.getChunkIndex(), - installSnapshotState.getTotalChunks()); - } + private void sendSnapshotChunk(ActorSelection followerActor, FollowerLogInformation followerLogInfo) { + if (snapshot.isPresent()) { + LeaderInstallSnapshotState installSnapshotState = followerLogInfo.getInstallSnapshotState(); + if (installSnapshotState == null) { + installSnapshotState = new LeaderInstallSnapshotState(context.getConfigParams().getSnapshotChunkSize(), + logName()); + followerLogInfo.setLeaderInstallSnapshotState(installSnapshotState); } - } catch (IOException e) { - LOG.error("{}: InstallSnapshot failed for Leader.", logName(), e); - } - } - /** - * Acccepts snaphot as ByteString, enters into map for future chunks - * creates and return a ByteString chunk - */ - private byte[] getNextSnapshotChunk(String followerId, ByteString snapshotBytes) throws IOException { - LeaderInstallSnapshotState installSnapshotState = followerToLog.get(followerId).getInstallSnapshotState(); - if (installSnapshotState == null) { - installSnapshotState = new LeaderInstallSnapshotState(snapshotBytes, context.getConfigParams().getSnapshotChunkSize(), - logName()); - followerToLog.get(followerId).setLeaderInstallSnapshotState(installSnapshotState); - } - byte[] nextChunk = installSnapshotState.getNextChunk(); + // Ensure the snapshot bytes are set - this is a no-op. + installSnapshotState.setSnapshotBytes(snapshot.get().getSnapshotBytes()); + + byte[] nextSnapshotChunk = installSnapshotState.getNextChunk(); - LOG.debug("{}: next snapshot chunk size for follower {}: {}", logName(), followerId, nextChunk.length); + log.debug("{}: next snapshot chunk size for follower {}: {}", logName(), followerLogInfo.getId(), + nextSnapshotChunk.length); - return nextChunk; + int nextChunkIndex = installSnapshotState.incrementChunkIndex(); + Optional serverConfig = Optional.absent(); + if (installSnapshotState.isLastChunk(nextChunkIndex)) { + serverConfig = Optional.fromNullable(context.getPeerServerInfo(true)); + } + + followerActor.tell( + new InstallSnapshot(currentTerm(), context.getId(), + snapshot.get().getLastIncludedIndex(), + snapshot.get().getLastIncludedTerm(), + nextSnapshotChunk, + nextChunkIndex, + installSnapshotState.getTotalChunks(), + Optional.of(installSnapshotState.getLastChunkHashCode()), + serverConfig + ).toSerializable(followerLogInfo.getRaftVersion()), + actor() + ); + + log.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", logName(), followerActor.path(), + installSnapshotState.getChunkIndex(), installSnapshotState.getTotalChunks()); + } } private void sendHeartBeat() { if (!followerToLog.isEmpty()) { - LOG.trace("{}: Sending heartbeat", logName()); + log.trace("{}: Sending heartbeat", logName()); sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis(), true); } } @@ -838,7 +861,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { int minPresent = getMinIsolatedLeaderPeerCount(); for (FollowerLogInformation followerLogInformation : followerToLog.values()) { final PeerInfo peerInfo = context.getPeerInfo(followerLogInformation.getId()); - if(peerInfo != null && peerInfo.isVoting() && followerLogInformation.isFollowerActive()) { + if (peerInfo != null && peerInfo.isVoting() && followerLogInformation.isFollowerActive()) { --minPresent; if (minPresent == 0) { return false;