@Test
public void testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader() {

    logStart("testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader");
    MockRaftActorContext leaderActorContext = createActorContextWithFollower();
    ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
            new FiniteDuration(1000, TimeUnit.SECONDS));
    leaderActorContext.setReplicatedLog(
            new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
    long leaderCommitIndex = 2;
    leaderActorContext.setCommitIndex(leaderCommitIndex);
    leaderActorContext.setLastApplied(leaderCommitIndex);
    final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
    final ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
    MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
    followerActorContext.setReplicatedLog(
            new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
    followerActorContext.setCommitIndex(0);
    followerActorContext.setLastApplied(0);
    Follower follower = new Follower(followerActorContext);
    followerActor.underlyingActor().setBehavior(follower);
    leader = new Leader(leaderActorContext);
    AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
    final AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
            AppendEntriesReply.class);
    MessageCollectorActor.clearMessages(followerActor);
    MessageCollectorActor.clearMessages(leaderActor);
    // Verify initial AppendEntries sent.
    assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
    assertEquals("Log entries size", 0, appendEntries.getEntries().size());
    assertEquals("getPrevLogIndex", 1, appendEntries.getPrevLogIndex());
    leaderActor.underlyingActor().setBehavior(leader);
    leader.handleMessage(followerActor, appendEntriesReply);
    MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
    appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
    assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
    assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
    assertEquals("Log entries size", 2, appendEntries.getEntries().size());
    assertEquals("First entry index", 1, appendEntries.getEntries().get(0).getIndex());
    assertEquals("First entry data", leadersSecondLogEntry.getData(),
            appendEntries.getEntries().get(0).getData());
    assertEquals("Second entry index", 2, appendEntries.getEntries().get(1).getIndex());
    assertEquals("Second entry data", leadersThirdLogEntry.getData(),
            appendEntries.getEntries().get(1).getData());
    FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
    assertEquals("getNextIndex", 3, followerInfo.getNextIndex());
    List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
    ApplyState applyState = applyStateList.get(0);
    assertEquals("Follower's first ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
    assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
    assertEquals("Follower's first ApplyState data", leadersSecondLogEntry.getData(),
            applyState.getReplicatedLogEntry().getData());
    applyState = applyStateList.get(1);
    assertEquals("Follower's second ApplyState index", 2, applyState.getReplicatedLogEntry().getIndex());
    assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
    assertEquals("Follower's second ApplyState data", leadersThirdLogEntry.getData(),
            applyState.getReplicatedLogEntry().getData());
    assertEquals("Follower's commit index", 2, followerActorContext.getCommitIndex());
    assertEquals("Follower's lastIndex", 2, followerActorContext.getReplicatedLog().lastIndex());
}