@Test
public void testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded() {

    logStart("testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded");
    MockRaftActorContext leaderActorContext = createActorContextWithFollower();
    ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
            new FiniteDuration(1000, TimeUnit.SECONDS));
    ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setSnapshotChunkSize(2);
    leaderActorContext.setReplicatedLog(
            new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 4, 1).build());
    long leaderCommitIndex = 3;
    leaderActorContext.setCommitIndex(leaderCommitIndex);
    leaderActorContext.setLastApplied(leaderCommitIndex);
    final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
    final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
    final ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
    final ReplicatedLogEntry leadersFourthLogEntry = leaderActorContext.getReplicatedLog().get(3);
    MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
    followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
    followerActorContext.setCommitIndex(-1);
    followerActorContext.setLastApplied(-1);
    Follower follower = new Follower(followerActorContext);
    followerActor.underlyingActor().setBehavior(follower);
    followerActorContext.setCurrentBehavior(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 with the leader's current commit index.
    assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
    assertEquals("Log entries size", 0, appendEntries.getEntries().size());
    assertEquals("getPrevLogIndex", 2, appendEntries.getPrevLogIndex());
    leaderActor.underlyingActor().setBehavior(leader);
    leaderActorContext.setCurrentBehavior(leader);
    leader.handleMessage(followerActor, appendEntriesReply);
    List<AppendEntries> appendEntriesList = MessageCollectorActor.expectMatching(followerActor,
            AppendEntries.class, 2);
    MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 2);
    appendEntries = appendEntriesList.get(0);
    assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
    assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
    assertEquals("Log entries size", 2, appendEntries.getEntries().size());
    assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
    assertEquals("First entry data", leadersFirstLogEntry.getData(),
            appendEntries.getEntries().get(0).getData());
    assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
    assertEquals("Second entry data", leadersSecondLogEntry.getData(),
            appendEntries.getEntries().get(1).getData());
    appendEntries = appendEntriesList.get(1);
    assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
    assertEquals("getPrevLogIndex", 1, appendEntries.getPrevLogIndex());
    assertEquals("Log entries size", 2, appendEntries.getEntries().size());
    assertEquals("First entry index", 2, appendEntries.getEntries().get(0).getIndex());
    assertEquals("First entry data", leadersThirdLogEntry.getData(),
            appendEntries.getEntries().get(0).getData());
    assertEquals("Second entry index", 3, appendEntries.getEntries().get(1).getIndex());
    assertEquals("Second entry data", leadersFourthLogEntry.getData(),
            appendEntries.getEntries().get(1).getData());
    FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
    assertEquals("getNextIndex", 4, followerInfo.getNextIndex());
    MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 4);
    assertEquals("Follower's commit index", 3, followerActorContext.getCommitIndex());
    assertEquals("Follower's lastIndex", 3, followerActorContext.getReplicatedLog().lastIndex());
}