opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft.behaviors;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSelection;
  13 import akka.actor.Cancellable;
  14 import com.google.common.annotations.VisibleForTesting;
  15 import com.google.common.base.Optional;
  16 import com.google.common.base.Preconditions;
  17 import com.google.protobuf.ByteString;
  18 import java.io.IOException;
  19 import java.util.Collection;
  20 import java.util.Collections;
  21 import java.util.HashMap;
  22 import java.util.Iterator;
  23 import java.util.LinkedList;
  24 import java.util.List;
  25 import java.util.Map;
  26 import java.util.Map.Entry;
  27 import javax.annotation.Nullable;
  28 import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
  29 import org.opendaylight.controller.cluster.raft.ClientRequestTrackerImpl;
  30 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
  31 import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
  32 import org.opendaylight.controller.cluster.raft.PeerInfo;
  33 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  34 import org.opendaylight.controller.cluster.raft.RaftState;
  35 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  36 import org.opendaylight.controller.cluster.raft.Snapshot;
  37 import org.opendaylight.controller.cluster.raft.VotingState;
  38 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
  39 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
  40 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  41 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  42 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  43 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  44 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  45 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  46 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  47 import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply;
  48 import scala.concurrent.duration.FiniteDuration;
  49
  50 /**
  51  * The behavior of a RaftActor when it is in the Leader state
  52  * <p/>
  53  * Leaders:
  54  * <ul>
  55  * <li> Upon election: send initial empty AppendEntries RPCs
  56  * (heartbeat) to each server; repeat during idle periods to
  57  * prevent election timeouts (§5.2)
  58  * <li> If command received from client: append entry to local log,
  59  * respond after entry applied to state machine (§5.3)
  60  * <li> If last log index ≥ nextIndex for a follower: send
  61  * AppendEntries RPC with log entries starting at nextIndex
  62  * <ul>
  63  * <li> If successful: update nextIndex and matchIndex for
  64  * follower (§5.3)
  65  * <li> If AppendEntries fails because of log inconsistency:
  66  * decrement nextIndex and retry (§5.3)
  67  * </ul>
  68  * <li> If there exists an N such that N > commitIndex, a majority
  69  * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
  70  * set commitIndex = N (§5.3, §5.4).
  71  */
  72 public abstract class AbstractLeader extends AbstractRaftActorBehavior {
  73
  74     // The index of the first chunk that is sent when installing a snapshot
  75     public static final int FIRST_CHUNK_INDEX = 1;
  76
  77     // The index that the follower should respond with if it needs the install snapshot to be reset
  78     public static final int INVALID_CHUNK_INDEX = -1;
  79
  80     // This would be passed as the hash code of the last chunk when sending the first chunk
  81     public static final int INITIAL_LAST_CHUNK_HASH_CODE = -1;
  82
  83     private final Map<String, FollowerLogInformation> followerToLog = new HashMap<>();
  84     private final Map<String, FollowerToSnapshot> mapFollowerToSnapshot = new HashMap<>();
  85
  86     private Cancellable heartbeatSchedule = null;
  87
  88     private final Collection<ClientRequestTracker> trackerList = new LinkedList<>();
  89
  90     private int minReplicationCount;
  91
  92     private Optional<SnapshotHolder> snapshot;
  93
  94     public AbstractLeader(RaftActorContext context) {
  95         super(context, RaftState.Leader);
  96
  97         setLeaderPayloadVersion(context.getPayloadVersion());
  98
  99         for(PeerInfo peerInfo: context.getPeers()) {
 100             FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(peerInfo, -1, context);
 101             followerToLog.put(peerInfo.getId(), followerLogInformation);
 102         }
 103
 104         leaderId = context.getId();
 105
 106         LOG.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds());
 107
 108         updateMinReplicaCount();
 109
 110         snapshot = Optional.absent();
 111
 112         // Immediately schedule a heartbeat
 113         // Upon election: send initial empty AppendEntries RPCs
 114         // (heartbeat) to each server; repeat during idle periods to
 115         // prevent election timeouts (§5.2)
 116         sendAppendEntries(0, false);
 117
 118         // It is important to schedule this heartbeat here
 119         scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
 120     }
 121
 122     /**
 123      * Return an immutable collection of follower identifiers.
 124      *
 125      * @return Collection of follower IDs
 126      */
 127     public final Collection<String> getFollowerIds() {
 128         return followerToLog.keySet();
 129     }
 130
 131     public void addFollower(String followerId) {
 132         FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(
 133                 context.getPeerInfo(followerId), -1, context);
 134         followerToLog.put(followerId, followerLogInformation);
 135
 136         if(heartbeatSchedule == null) {
 137             scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
 138         }
 139     }
 140
 141     public void removeFollower(String followerId) {
 142         followerToLog.remove(followerId);
 143         mapFollowerToSnapshot.remove(followerId);
 144     }
 145
 146     public void updateMinReplicaCount() {
 147         int numVoting = 0;
 148         for(PeerInfo peer: context.getPeers()) {
 149             if(peer.isVoting()) {
 150                 numVoting++;
 151             }
 152         }
 153
 154         minReplicationCount = getMajorityVoteCount(numVoting);
 155     }
 156
 157     protected int getMinIsolatedLeaderPeerCount(){
 158       //the isolated Leader peer count will be 1 less than the majority vote count.
 159         //this is because the vote count has the self vote counted in it
 160         //for e.g
 161         //0 peers = 1 votesRequired , minIsolatedLeaderPeerCount = 0
 162         //2 peers = 2 votesRequired , minIsolatedLeaderPeerCount = 1
 163         //4 peers = 3 votesRequired, minIsolatedLeaderPeerCount = 2
 164
 165         return minReplicationCount > 0 ? (minReplicationCount - 1) : 0;
 166     }
 167
 168     @VisibleForTesting
 169     void setSnapshot(@Nullable Snapshot snapshot) {
 170         if(snapshot != null) {
 171             this.snapshot = Optional.of(new SnapshotHolder(snapshot));
 172         } else {
 173             this.snapshot = Optional.absent();
 174         }
 175     }
 176
 177     @Override
 178     protected RaftActorBehavior handleAppendEntries(ActorRef sender,
 179         AppendEntries appendEntries) {
 180
 181         LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
 182
 183         return this;
 184     }
 185
 186     @Override
 187     protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
 188         AppendEntriesReply appendEntriesReply) {
 189
 190         if(LOG.isTraceEnabled()) {
 191             LOG.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply);
 192         }
 193
 194         // Update the FollowerLogInformation
 195         String followerId = appendEntriesReply.getFollowerId();
 196         FollowerLogInformation followerLogInformation =
 197             followerToLog.get(followerId);
 198
 199         if(followerLogInformation == null){
 200             LOG.error("{}: handleAppendEntriesReply - unknown follower {}", logName(), followerId);
 201             return this;
 202         }
 203
 204         if(followerLogInformation.timeSinceLastActivity() >
 205                 context.getConfigParams().getElectionTimeOutInterval().toMillis()) {
 206             LOG.warn("{} : handleAppendEntriesReply delayed beyond election timeout, " +
 207                             "appendEntriesReply : {}, timeSinceLastActivity : {}, lastApplied : {}, commitIndex : {}",
 208                     logName(), appendEntriesReply, followerLogInformation.timeSinceLastActivity(),
 209                     context.getLastApplied(), context.getCommitIndex());
 210         }
 211
 212         followerLogInformation.markFollowerActive();
 213         followerLogInformation.setPayloadVersion(appendEntriesReply.getPayloadVersion());
 214         followerLogInformation.setRaftVersion(appendEntriesReply.getRaftVersion());
 215
 216         boolean updated = false;
 217         if (appendEntriesReply.isSuccess()) {
 218             updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
 219         } else {
 220             LOG.debug("{}: handleAppendEntriesReply: received unsuccessful reply: {}", logName(), appendEntriesReply);
 221
 222             long followerLastLogIndex = appendEntriesReply.getLogLastIndex();
 223             ReplicatedLogEntry followersLastLogEntry = context.getReplicatedLog().get(followerLastLogIndex);
 224             if(appendEntriesReply.isForceInstallSnapshot()) {
 225                 // Reset the followers match and next index. This is to signal that this follower has nothing
 226                 // in common with this Leader and so would require a snapshot to be installed
 227                 followerLogInformation.setMatchIndex(-1);
 228                 followerLogInformation.setNextIndex(-1);
 229
 230                 // Force initiate a snapshot capture
 231                 initiateCaptureSnapshot(followerId);
 232             } else if(followerLastLogIndex < 0 || (followersLastLogEntry != null &&
 233                     followersLastLogEntry.getTerm() == appendEntriesReply.getLogLastTerm())) {
 234                 // The follower's log is empty or the last entry is present in the leader's journal
 235                 // and the terms match so the follower is just behind the leader's journal from
 236                 // the last snapshot, if any. We'll catch up the follower quickly by starting at the
 237                 // follower's last log index.
 238
 239                 updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
 240             } else {
 241                 // TODO: When we find that the follower is out of sync with the
 242                 // Leader we simply decrement that followers next index by 1.
 243                 // Would it be possible to do better than this? The RAFT spec
 244                 // does not explicitly deal with it but may be something for us to
 245                 // think about.
 246
 247                 followerLogInformation.decrNextIndex();
 248             }
 249         }
 250
 251         // Now figure out if this reply warrants a change in the commitIndex
 252         // If there exists an N such that N > commitIndex, a majority
 253         // of matchIndex[i] ≥ N, and log[N].term == currentTerm:
 254         // set commitIndex = N (§5.3, §5.4).
 255         for (long N = context.getCommitIndex() + 1; ; N++) {
 256             int replicatedCount = 1;
 257
 258             for (FollowerLogInformation info : followerToLog.values()) {
 259                 final PeerInfo peerInfo = context.getPeerInfo(info.getId());
 260                 if(info.getMatchIndex() >= N && (peerInfo != null && peerInfo.isVoting())) {
 261                     replicatedCount++;
 262                 }
 263             }
 264
 265             if (replicatedCount >= minReplicationCount) {
 266                 ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(N);
 267                 if (replicatedLogEntry == null) {
 268                     break;
 269                 }
 270
 271                 // Don't update the commit index if the log entry is from a previous term, as per §5.4.1:
 272                 // "Raft never commits log entries from previous terms by counting replicas".
 273                 // However we keep looping so we can make progress when new entries in the current term
 274                 // reach consensus, as per §5.4.1: "once an entry from the current term is committed by
 275                 // counting replicas, then all prior entries are committed indirectly".
 276                 if (replicatedLogEntry.getTerm() == currentTerm()) {
 277                     context.setCommitIndex(N);
 278                 }
 279             } else {
 280                 break;
 281             }
 282         }
 283
 284         // Apply the change to the state machine
 285         if (context.getCommitIndex() > context.getLastApplied()) {
 286             if(LOG.isDebugEnabled()) {
 287                 LOG.debug("{}: handleAppendEntriesReply from {}: applying to log - commitIndex: {}, lastAppliedIndex: {}",
 288                         logName(), followerId, context.getCommitIndex(), context.getLastApplied());
 289             }
 290
 291             applyLogToStateMachine(context.getCommitIndex());
 292         }
 293
 294         if (!context.getSnapshotManager().isCapturing()) {
 295             purgeInMemoryLog();
 296         }
 297
 298         //Send the next log entry immediately, if possible, no need to wait for heartbeat to trigger that event
 299         sendUpdatesToFollower(followerId, followerLogInformation, false, !updated);
 300         return this;
 301     }
 302
 303     private boolean updateFollowerLogInformation(FollowerLogInformation followerLogInformation,
 304             AppendEntriesReply appendEntriesReply) {
 305         boolean updated = followerLogInformation.setMatchIndex(appendEntriesReply.getLogLastIndex());
 306         updated = followerLogInformation.setNextIndex(appendEntriesReply.getLogLastIndex() + 1) || updated;
 307
 308         if(updated && LOG.isDebugEnabled()) {
 309             LOG.debug("{}: handleAppendEntriesReply - FollowerLogInformation for {} updated: matchIndex: {}, nextIndex: {}",
 310                     logName(), followerLogInformation.getId(), followerLogInformation.getMatchIndex(),
 311                     followerLogInformation.getNextIndex());
 312         }
 313         return updated;
 314     }
 315
 316     private void purgeInMemoryLog() {
 317         //find the lowest index across followers which has been replicated to all.
 318         // lastApplied if there are no followers, so that we keep clearing the log for single-node
 319         // we would delete the in-mem log from that index on, in-order to minimize mem usage
 320         // we would also share this info thru AE with the followers so that they can delete their log entries as well.
 321         long minReplicatedToAllIndex = followerToLog.isEmpty() ? context.getLastApplied() : Long.MAX_VALUE;
 322         for (FollowerLogInformation info : followerToLog.values()) {
 323             minReplicatedToAllIndex = Math.min(minReplicatedToAllIndex, info.getMatchIndex());
 324         }
 325
 326         super.performSnapshotWithoutCapture(minReplicatedToAllIndex);
 327     }
 328
 329     @Override
 330     protected ClientRequestTracker removeClientRequestTracker(long logIndex) {
 331         final Iterator<ClientRequestTracker> it = trackerList.iterator();
 332         while (it.hasNext()) {
 333             final ClientRequestTracker t = it.next();
 334             if (t.getIndex() == logIndex) {
 335                 it.remove();
 336                 return t;
 337             }
 338         }
 339
 340         return null;
 341     }
 342
 343     @Override
 344     protected ClientRequestTracker findClientRequestTracker(long logIndex) {
 345         for (ClientRequestTracker tracker : trackerList) {
 346             if (tracker.getIndex() == logIndex) {
 347                 return tracker;
 348             }
 349         }
 350         return null;
 351     }
 352
 353     @Override
 354     protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
 355         RequestVoteReply requestVoteReply) {
 356         return this;
 357     }
 358
 359     protected void beforeSendHeartbeat(){}
 360
 361     @Override
 362     public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 363         Preconditions.checkNotNull(sender, "sender should not be null");
 364
 365         Object message = fromSerializableMessage(originalMessage);
 366
 367         if (message instanceof RaftRPC) {
 368             RaftRPC rpc = (RaftRPC) message;
 369             // If RPC request or response contains term T > currentTerm:
 370             // set currentTerm = T, convert to follower (§5.1)
 371             // This applies to all RPC messages and responses
 372             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 373                 LOG.debug("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower",
 374                         logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
 375
 376                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 377
 378                 return internalSwitchBehavior(RaftState.Follower);
 379             }
 380         }
 381
 382         if (message instanceof SendHeartBeat) {
 383             beforeSendHeartbeat();
 384             sendHeartBeat();
 385             scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
 386             return this;
 387
 388         } else if(message instanceof SendInstallSnapshot) {
 389             // received from RaftActor
 390             setSnapshot(((SendInstallSnapshot) message).getSnapshot());
 391             sendInstallSnapshot();
 392
 393         } else if (message instanceof Replicate) {
 394             replicate((Replicate) message);
 395
 396         } else if (message instanceof InstallSnapshotReply){
 397             handleInstallSnapshotReply((InstallSnapshotReply) message);
 398
 399         }
 400
 401
 402         return super.handleMessage(sender, message);
 403     }
 404
 405     private void handleInstallSnapshotReply(InstallSnapshotReply reply) {
 406         LOG.debug("{}: handleInstallSnapshotReply: {}", logName(), reply);
 407
 408         String followerId = reply.getFollowerId();
 409         FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
 410
 411         if (followerToSnapshot == null) {
 412             LOG.error("{}: FollowerToSnapshot not found for follower {} in InstallSnapshotReply",
 413                     logName(), followerId);
 414             return;
 415         }
 416
 417         FollowerLogInformation followerLogInformation = followerToLog.get(followerId);
 418         if(followerLogInformation == null) {
 419             // This can happen during AddServer if it times out.
 420             LOG.error("{}: FollowerLogInformation not found for follower {} in InstallSnapshotReply",
 421                     logName(), followerId);
 422             mapFollowerToSnapshot.remove(followerId);
 423             return;
 424         }
 425
 426         followerLogInformation.markFollowerActive();
 427
 428         if (followerToSnapshot.getChunkIndex() == reply.getChunkIndex()) {
 429             boolean wasLastChunk = false;
 430             if (reply.isSuccess()) {
 431                 if(followerToSnapshot.isLastChunk(reply.getChunkIndex())) {
 432                     //this was the last chunk reply
 433                     if(LOG.isDebugEnabled()) {
 434                         LOG.debug("{}: InstallSnapshotReply received, " +
 435                                 "last chunk received, Chunk: {}. Follower: {} Setting nextIndex: {}",
 436                                 logName(), reply.getChunkIndex(), followerId,
 437                             context.getReplicatedLog().getSnapshotIndex() + 1
 438                         );
 439                     }
 440
 441                     long followerMatchIndex = snapshot.get().getLastIncludedIndex();
 442                     followerLogInformation.setMatchIndex(followerMatchIndex);
 443                     followerLogInformation.setNextIndex(followerMatchIndex + 1);
 444                     mapFollowerToSnapshot.remove(followerId);
 445
 446                     LOG.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}",
 447                         logName(), followerId, followerLogInformation.getMatchIndex(),
 448                         followerLogInformation.getNextIndex());
 449
 450                     if (mapFollowerToSnapshot.isEmpty()) {
 451                         // once there are no pending followers receiving snapshots
 452                         // we can remove snapshot from the memory
 453                         setSnapshot(null);
 454                     }
 455                     wasLastChunk = true;
 456                     if(context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED){
 457                         UnInitializedFollowerSnapshotReply unInitFollowerSnapshotSuccess =
 458                                              new UnInitializedFollowerSnapshotReply(followerId);
 459                         context.getActor().tell(unInitFollowerSnapshotSuccess, context.getActor());
 460                         LOG.debug("Sent message UnInitializedFollowerSnapshotReply to self");
 461                     }
 462                 } else {
 463                     followerToSnapshot.markSendStatus(true);
 464                 }
 465             } else {
 466                 LOG.info("{}: InstallSnapshotReply received sending snapshot chunk failed, Will retry, Chunk: {}",
 467                         logName(), reply.getChunkIndex());
 468
 469                 followerToSnapshot.markSendStatus(false);
 470             }
 471
 472             if (wasLastChunk && !context.getSnapshotManager().isCapturing()) {
 473                 // Since the follower is now caught up try to purge the log.
 474                 purgeInMemoryLog();
 475             } else if (!wasLastChunk && followerToSnapshot.canSendNextChunk()) {
 476                 ActorSelection followerActor = context.getPeerActorSelection(followerId);
 477                 if(followerActor != null) {
 478                     sendSnapshotChunk(followerActor, followerId);
 479                 }
 480             }
 481
 482         } else {
 483             LOG.error("{}: Chunk index {} in InstallSnapshotReply from follower {} does not match expected index {}",
 484                     logName(), reply.getChunkIndex(), followerId,
 485                     followerToSnapshot.getChunkIndex());
 486
 487             if(reply.getChunkIndex() == INVALID_CHUNK_INDEX){
 488                 // Since the Follower did not find this index to be valid we should reset the follower snapshot
 489                 // so that Installing the snapshot can resume from the beginning
 490                 followerToSnapshot.reset();
 491             }
 492         }
 493     }
 494
 495     private void replicate(Replicate replicate) {
 496         long logIndex = replicate.getReplicatedLogEntry().getIndex();
 497
 498         LOG.debug("{}: Replicate message: identifier: {}, logIndex: {}", logName(),
 499                 replicate.getIdentifier(), logIndex);
 500
 501         // Create a tracker entry we will use this later to notify the
 502         // client actor
 503         trackerList.add(
 504             new ClientRequestTrackerImpl(replicate.getClientActor(),
 505                 replicate.getIdentifier(),
 506                 logIndex)
 507         );
 508
 509         boolean applyModificationToState = followerToLog.isEmpty()
 510                 || context.getRaftPolicy().applyModificationToStateBeforeConsensus();
 511
 512         if(applyModificationToState){
 513             context.setCommitIndex(logIndex);
 514             applyLogToStateMachine(logIndex);
 515         }
 516
 517         if (!followerToLog.isEmpty()) {
 518             sendAppendEntries(0, false);
 519         }
 520     }
 521
 522     protected void sendAppendEntries(long timeSinceLastActivityInterval, boolean isHeartbeat) {
 523         // Send an AppendEntries to all followers
 524         for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
 525             final String followerId = e.getKey();
 526             final FollowerLogInformation followerLogInformation = e.getValue();
 527             // This checks helps not to send a repeat message to the follower
 528             if(!followerLogInformation.isFollowerActive() ||
 529                     followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) {
 530                 sendUpdatesToFollower(followerId, followerLogInformation, true, isHeartbeat);
 531             }
 532         }
 533     }
 534
 535     /**
 536      *
 537      * This method checks if any update needs to be sent to the given follower. This includes append log entries,
 538      * sending next snapshot chunk, and initiating a snapshot.
 539      * @return true if any update is sent, false otherwise
 540      */
 541
 542     private void sendUpdatesToFollower(String followerId, FollowerLogInformation followerLogInformation,
 543                                        boolean sendHeartbeat, boolean isHeartbeat) {
 544
 545         ActorSelection followerActor = context.getPeerActorSelection(followerId);
 546         if (followerActor != null) {
 547             long followerNextIndex = followerLogInformation.getNextIndex();
 548             boolean isFollowerActive = followerLogInformation.isFollowerActive();
 549             boolean sendAppendEntries = false;
 550             List<ReplicatedLogEntry> entries = Collections.emptyList();
 551
 552             if (mapFollowerToSnapshot.get(followerId) != null) {
 553                 // if install snapshot is in process , then sent next chunk if possible
 554                 if (isFollowerActive && mapFollowerToSnapshot.get(followerId).canSendNextChunk()) {
 555                     sendSnapshotChunk(followerActor, followerId);
 556                 } else if(sendHeartbeat) {
 557                     // we send a heartbeat even if we have not received a reply for the last chunk
 558                     sendAppendEntries = true;
 559                 }
 560             } else {
 561                 long leaderLastIndex = context.getReplicatedLog().lastIndex();
 562                 long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex();
 563
 564                 if((!isHeartbeat && LOG.isDebugEnabled()) || LOG.isTraceEnabled()) {
 565                     LOG.debug("{}: Checking sendAppendEntries for follower {}: active: {}, followerNextIndex: {}, leaderLastIndex: {}, leaderSnapShotIndex: {}",
 566                             logName(), followerId, isFollowerActive, followerNextIndex, leaderLastIndex, leaderSnapShotIndex);
 567                 }
 568
 569                 if (isFollowerActive && context.getReplicatedLog().isPresent(followerNextIndex)) {
 570
 571                     LOG.debug("{}: sendAppendEntries: {} is present for follower {}", logName(),
 572                             followerNextIndex, followerId);
 573
 574                     if(followerLogInformation.okToReplicate()) {
 575                         // Try to send all the entries in the journal but not exceeding the max data size
 576                         // for a single AppendEntries message.
 577                         int maxEntries = (int) context.getReplicatedLog().size();
 578                         entries = context.getReplicatedLog().getFrom(followerNextIndex, maxEntries,
 579                                 context.getConfigParams().getSnapshotChunkSize());
 580                         sendAppendEntries = true;
 581                     }
 582                 } else if (isFollowerActive && followerNextIndex >= 0 &&
 583                     leaderLastIndex > followerNextIndex && !context.getSnapshotManager().isCapturing()) {
 584                     // if the followers next index is not present in the leaders log, and
 585                     // if the follower is just not starting and if leader's index is more than followers index
 586                     // then snapshot should be sent
 587
 588                     if (LOG.isDebugEnabled()) {
 589                         LOG.debug(String.format("%s: InitiateInstallSnapshot to follower: %s," +
 590                                     "follower-nextIndex: %d, leader-snapshot-index: %d,  " +
 591                                     "leader-last-index: %d", logName(), followerId,
 592                                     followerNextIndex, leaderSnapShotIndex, leaderLastIndex));
 593                     }
 594
 595                     // Send heartbeat to follower whenever install snapshot is initiated.
 596                     sendAppendEntries = true;
 597                     if (canInstallSnapshot(followerNextIndex)) {
 598                         initiateCaptureSnapshot(followerId);
 599                     }
 600
 601                 } else if(sendHeartbeat) {
 602                     // we send an AppendEntries, even if the follower is inactive
 603                     // in-order to update the followers timestamp, in case it becomes active again
 604                     sendAppendEntries = true;
 605                 }
 606
 607             }
 608
 609             if(sendAppendEntries) {
 610                 sendAppendEntriesToFollower(followerActor, followerNextIndex,
 611                         entries, followerId);
 612             }
 613         }
 614     }
 615
 616     private void sendAppendEntriesToFollower(ActorSelection followerActor, long followerNextIndex,
 617         List<ReplicatedLogEntry> entries, String followerId) {
 618         AppendEntries appendEntries = new AppendEntries(currentTerm(), context.getId(),
 619             prevLogIndex(followerNextIndex),
 620             prevLogTerm(followerNextIndex), entries,
 621             context.getCommitIndex(), super.getReplicatedToAllIndex(), context.getPayloadVersion());
 622
 623         if(!entries.isEmpty() || LOG.isTraceEnabled()) {
 624             LOG.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerId,
 625                     appendEntries);
 626         }
 627
 628         followerActor.tell(appendEntries, actor());
 629     }
 630
 631     /**
 632      * Install Snapshot works as follows
 633      * 1. Leader initiates the capture snapshot by sending a CaptureSnapshot message to actor
 634      * 2. RaftActor on receipt of the CaptureSnapshotReply (from Shard), stores the received snapshot in the replicated log
 635      * and makes a call to Leader's handleMessage , with SendInstallSnapshot message.
 636      * 3. Leader , picks the snapshot from im-mem ReplicatedLog and sends it in chunks to the Follower
 637      * 4. On complete, Follower sends back a InstallSnapshotReply.
 638      * 5. On receipt of the InstallSnapshotReply for the last chunk, Leader marks the install complete for that follower
 639      * and replenishes the memory by deleting the snapshot in Replicated log.
 640      * 6. If another follower requires a snapshot and a snapshot has been collected (via CaptureSnapshotReply)
 641      * then send the existing snapshot in chunks to the follower.
 642      * @param followerId
 643      */
 644     public boolean initiateCaptureSnapshot(String followerId) {
 645         if (snapshot.isPresent()) {
 646             // if a snapshot is present in the memory, most likely another install is in progress
 647             // no need to capture snapshot.
 648             // This could happen if another follower needs an install when one is going on.
 649             final ActorSelection followerActor = context.getPeerActorSelection(followerId);
 650             sendSnapshotChunk(followerActor, followerId);
 651             return true;
 652         } else {
 653             return context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(),
 654                     this.getReplicatedToAllIndex(), followerId);
 655         }
 656     }
 657
 658     private boolean canInstallSnapshot(long nextIndex){
 659         // If the follower's nextIndex is -1 then we might as well send it a snapshot
 660         // Otherwise send it a snapshot only if the nextIndex is not present in the log but is present
 661         // in the snapshot
 662         return (nextIndex == -1 ||
 663                 (!context.getReplicatedLog().isPresent(nextIndex)
 664                         && context.getReplicatedLog().isInSnapshot(nextIndex)));
 665
 666     }
 667
 668
 669     private void sendInstallSnapshot() {
 670         LOG.debug("{}: sendInstallSnapshot", logName());
 671         for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
 672             String followerId = e.getKey();
 673             ActorSelection followerActor = context.getPeerActorSelection(followerId);
 674             FollowerLogInformation followerLogInfo = e.getValue();
 675
 676             if (followerActor != null) {
 677                 long nextIndex = followerLogInfo.getNextIndex();
 678                 if (context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED ||
 679                         canInstallSnapshot(nextIndex)) {
 680                     sendSnapshotChunk(followerActor, followerId);
 681                 }
 682             }
 683         }
 684     }
 685
 686     /**
 687      *  Sends a snapshot chunk to a given follower
 688      *  InstallSnapshot should qualify as a heartbeat too.
 689      */
 690     private void sendSnapshotChunk(ActorSelection followerActor, String followerId) {
 691         try {
 692             if (snapshot.isPresent()) {
 693                 ByteString nextSnapshotChunk = getNextSnapshotChunk(followerId, snapshot.get().getSnapshotBytes());
 694
 695                 // Note: the previous call to getNextSnapshotChunk has the side-effect of adding
 696                 // followerId to the followerToSnapshot map.
 697                 FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
 698
 699                 followerActor.tell(
 700                     new InstallSnapshot(currentTerm(), context.getId(),
 701                         snapshot.get().getLastIncludedIndex(),
 702                         snapshot.get().getLastIncludedTerm(),
 703                         nextSnapshotChunk,
 704                         followerToSnapshot.incrementChunkIndex(),
 705                         followerToSnapshot.getTotalChunks(),
 706                         Optional.of(followerToSnapshot.getLastChunkHashCode())
 707                     ).toSerializable(),
 708                     actor()
 709                 );
 710
 711                 if(LOG.isDebugEnabled()) {
 712                     LOG.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}",
 713                             logName(), followerActor.path(), followerToSnapshot.getChunkIndex(),
 714                             followerToSnapshot.getTotalChunks());
 715                 }
 716             }
 717         } catch (IOException e) {
 718             LOG.error("{}: InstallSnapshot failed for Leader.", logName(), e);
 719         }
 720     }
 721
 722     /**
 723      * Acccepts snaphot as ByteString, enters into map for future chunks
 724      * creates and return a ByteString chunk
 725      */
 726     private ByteString getNextSnapshotChunk(String followerId, ByteString snapshotBytes) throws IOException {
 727         FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
 728         if (followerToSnapshot == null) {
 729             followerToSnapshot = new FollowerToSnapshot(snapshotBytes);
 730             mapFollowerToSnapshot.put(followerId, followerToSnapshot);
 731         }
 732         ByteString nextChunk = followerToSnapshot.getNextChunk();
 733
 734         LOG.debug("{}: next snapshot chunk size for follower {}: {}", logName(), followerId, nextChunk.size());
 735
 736         return nextChunk;
 737     }
 738
 739     private void sendHeartBeat() {
 740         if (!followerToLog.isEmpty()) {
 741             LOG.trace("{}: Sending heartbeat", logName());
 742             sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis(), true);
 743         }
 744     }
 745
 746     private void stopHeartBeat() {
 747         if (heartbeatSchedule != null && !heartbeatSchedule.isCancelled()) {
 748             heartbeatSchedule.cancel();
 749         }
 750     }
 751
 752     private void scheduleHeartBeat(FiniteDuration interval) {
 753         if (followerToLog.isEmpty()) {
 754             // Optimization - do not bother scheduling a heartbeat as there are
 755             // no followers
 756             return;
 757         }
 758
 759         stopHeartBeat();
 760
 761         // Schedule a heartbeat. When the scheduler triggers a SendHeartbeat
 762         // message is sent to itself.
 763         // Scheduling the heartbeat only once here because heartbeats do not
 764         // need to be sent if there are other messages being sent to the remote
 765         // actor.
 766         heartbeatSchedule = context.getActorSystem().scheduler().scheduleOnce(
 767             interval, context.getActor(), new SendHeartBeat(),
 768             context.getActorSystem().dispatcher(), context.getActor());
 769     }
 770
 771     @Override
 772     public void close() throws Exception {
 773         stopHeartBeat();
 774     }
 775
 776     @Override
 777     public String getLeaderId() {
 778         return context.getId();
 779     }
 780
 781     protected boolean isLeaderIsolated() {
 782         int minPresent = getMinIsolatedLeaderPeerCount();
 783         for (FollowerLogInformation followerLogInformation : followerToLog.values()) {
 784             if (followerLogInformation.isFollowerActive()) {
 785                 --minPresent;
 786                 if (minPresent == 0) {
 787                     break;
 788                 }
 789             }
 790         }
 791         return (minPresent != 0);
 792     }
 793
 794     /**
 795      * Encapsulates the snapshot bytestring and handles the logic of sending
 796      * snapshot chunks
 797      */
 798     protected class FollowerToSnapshot {
 799         private final ByteString snapshotBytes;
 800         private int offset = 0;
 801         // the next snapshot chunk is sent only if the replyReceivedForOffset matches offset
 802         private int replyReceivedForOffset;
 803         // if replyStatus is false, the previous chunk is attempted
 804         private boolean replyStatus = false;
 805         private int chunkIndex;
 806         private final int totalChunks;
 807         private int lastChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 808         private int nextChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 809
 810         public FollowerToSnapshot(ByteString snapshotBytes) {
 811             this.snapshotBytes = snapshotBytes;
 812             int size = snapshotBytes.size();
 813             totalChunks = ( size / context.getConfigParams().getSnapshotChunkSize()) +
 814                 ((size % context.getConfigParams().getSnapshotChunkSize()) > 0 ? 1 : 0);
 815             if(LOG.isDebugEnabled()) {
 816                 LOG.debug("{}: Snapshot {} bytes, total chunks to send:{}",
 817                         logName(), size, totalChunks);
 818             }
 819             replyReceivedForOffset = -1;
 820             chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
 821         }
 822
 823         public ByteString getSnapshotBytes() {
 824             return snapshotBytes;
 825         }
 826
 827         public int incrementOffset() {
 828             if(replyStatus) {
 829                 // if prev chunk failed, we would want to sent the same chunk again
 830                 offset = offset + context.getConfigParams().getSnapshotChunkSize();
 831             }
 832             return offset;
 833         }
 834
 835         public int incrementChunkIndex() {
 836             if (replyStatus) {
 837                 // if prev chunk failed, we would want to sent the same chunk again
 838                 chunkIndex =  chunkIndex + 1;
 839             }
 840             return chunkIndex;
 841         }
 842
 843         public int getChunkIndex() {
 844             return chunkIndex;
 845         }
 846
 847         public int getTotalChunks() {
 848             return totalChunks;
 849         }
 850
 851         public boolean canSendNextChunk() {
 852             // we only send a false if a chunk is sent but we have not received a reply yet
 853             return replyReceivedForOffset == offset;
 854         }
 855
 856         public boolean isLastChunk(int chunkIndex) {
 857             return totalChunks == chunkIndex;
 858         }
 859
 860         public void markSendStatus(boolean success) {
 861             if (success) {
 862                 // if the chunk sent was successful
 863                 replyReceivedForOffset = offset;
 864                 replyStatus = true;
 865                 lastChunkHashCode = nextChunkHashCode;
 866             } else {
 867                 // if the chunk sent was failure
 868                 replyReceivedForOffset = offset;
 869                 replyStatus = false;
 870             }
 871         }
 872
 873         public ByteString getNextChunk() {
 874             int snapshotLength = getSnapshotBytes().size();
 875             int start = incrementOffset();
 876             int size = context.getConfigParams().getSnapshotChunkSize();
 877             if (context.getConfigParams().getSnapshotChunkSize() > snapshotLength) {
 878                 size = snapshotLength;
 879             } else {
 880                 if ((start + context.getConfigParams().getSnapshotChunkSize()) > snapshotLength) {
 881                     size = snapshotLength - start;
 882                 }
 883             }
 884
 885
 886             LOG.debug("{}: Next chunk: length={}, offset={},size={}", logName(),
 887                     snapshotLength, start, size);
 888
 889             ByteString substring = getSnapshotBytes().substring(start, start + size);
 890             nextChunkHashCode = substring.hashCode();
 891             return substring;
 892         }
 893
 894         /**
 895          * reset should be called when the Follower needs to be sent the snapshot from the beginning
 896          */
 897         public void reset(){
 898             offset = 0;
 899             replyStatus = false;
 900             replyReceivedForOffset = offset;
 901             chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
 902             lastChunkHashCode = AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE;
 903         }
 904
 905         public int getLastChunkHashCode() {
 906             return lastChunkHashCode;
 907         }
 908     }
 909
 910     // called from example-actor for printing the follower-states
 911     public String printFollowerStates() {
 912         final StringBuilder sb = new StringBuilder();
 913
 914         sb.append('[');
 915         for (FollowerLogInformation followerLogInformation : followerToLog.values()) {
 916             sb.append('{');
 917             sb.append(followerLogInformation.getId());
 918             sb.append(" state:");
 919             sb.append(followerLogInformation.isFollowerActive());
 920             sb.append("},");
 921         }
 922         sb.append(']');
 923
 924         return sb.toString();
 925     }
 926
 927     @VisibleForTesting
 928     public FollowerLogInformation getFollower(String followerId) {
 929         return followerToLog.get(followerId);
 930     }
 931
 932     @VisibleForTesting
 933     protected void setFollowerSnapshot(String followerId, FollowerToSnapshot snapshot) {
 934         mapFollowerToSnapshot.put(followerId, snapshot);
 935     }
 936
 937     @VisibleForTesting
 938     public int followerSnapshotSize() {
 939         return mapFollowerToSnapshot.size();
 940     }
 941
 942     @VisibleForTesting
 943     public int followerLogSize() {
 944         return followerToLog.size();
 945     }
 946
 947     private static class SnapshotHolder {
 948         private final long lastIncludedTerm;
 949         private final long lastIncludedIndex;
 950         private final ByteString snapshotBytes;
 951
 952         SnapshotHolder(Snapshot snapshot) {
 953             this.lastIncludedTerm = snapshot.getLastAppliedTerm();
 954             this.lastIncludedIndex = snapshot.getLastAppliedIndex();
 955             this.snapshotBytes = ByteString.copyFrom(snapshot.getState());
 956         }
 957
 958         long getLastIncludedTerm() {
 959             return lastIncludedTerm;
 960         }
 961
 962         long getLastIncludedIndex() {
 963             return lastIncludedIndex;
 964         }
 965
 966         ByteString getSnapshotBytes() {
 967             return snapshotBytes;
 968         }
 969     }
 970 }