@Test
public void testHandleInstallSnapshotReplyLastChunk() throws Exception {

    logStart("testHandleInstallSnapshotReplyLastChunk");
    MockRaftActorContext actorContext = createActorContextWithFollower();
    final int commitIndex = 3;
    final int snapshotIndex = 2;
    final int snapshotTerm = 1;
    final int currentTerm = 2;
    actorContext.setCommitIndex(commitIndex);
    leader = new Leader(actorContext);
    actorContext.setCurrentBehavior(leader);
    leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
    leader.getFollower(FOLLOWER_ID).setNextIndex(0);
    // Ignore initial heartbeat.
    MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
    Map<String, String> leadersSnapshot = new HashMap<>();
    leadersSnapshot.put("1", "A");
    leadersSnapshot.put("2", "B");
    leadersSnapshot.put("3", "C");
    // set the snapshot variables in replicatedlog
    actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
    actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
    actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
    ByteString bs = toByteString(leadersSnapshot);
    leader.setSnapshotHolder(new SnapshotHolder(Snapshot.create(ByteState.of(bs.toByteArray()),
            Collections.<ReplicatedLogEntry>emptyList(), commitIndex, snapshotTerm, commitIndex, snapshotTerm,
            -1, null, null), ByteSource.wrap(bs.toByteArray())));
    LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState(
            actorContext.getConfigParams().getSnapshotChunkSize(), leader.logName());
    fts.setSnapshotBytes(ByteSource.wrap(bs.toByteArray()));
    leader.getFollower(FOLLOWER_ID).setLeaderInstallSnapshotState(fts);
    while (!fts.isLastChunk(fts.getChunkIndex())) {
        fts.getNextChunk();
        fts.incrementChunkIndex();
    }
    //clears leaders log
    actorContext.getReplicatedLog().removeFrom(0);
    RaftActorBehavior raftBehavior = leader.handleMessage(followerActor,
            new InstallSnapshotReply(currentTerm, FOLLOWER_ID, fts.getChunkIndex(), true));
    assertTrue(raftBehavior instanceof Leader);
    assertEquals(1, leader.followerLogSize());
    FollowerLogInformation fli = leader.getFollower(FOLLOWER_ID);
    assertNotNull(fli);
    assertNull(fli.getInstallSnapshotState());
    assertEquals(commitIndex, fli.getMatchIndex());
    assertEquals(commitIndex + 1, fli.getNextIndex());
    assertFalse(leader.hasSnapshot());
}