@Test public void testFirstAppendEntriesWithNoPrevIndexAndReplToAllPresentInSnapshotButCalculatedPrevEntryMissing() { logStart( "testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing"); MockRaftActorContext context = createActorContext(); context.getReplicatedLog().clear(0,2); context.getReplicatedLog().setSnapshotIndex(100); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 105, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 105, 100, (short) 0); follower = createBehavior(context); follower.handleMessage(leaderActor, appendEntries); FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class); assertFalse(syncStatus.isInitialSyncDone()); assertFalse("append entries reply should be false", reply.isSuccess()); }
@Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout || message instanceof TimeoutNow) { return handleElectionTimeout(message); } if (appendEntriesMessageAssembler.handleMessage(message, actor())) { return this; } if (!(message instanceof RaftRPC)) { return null; } final RaftRPC rpc = (RaftRPC) message; if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); } if (rpc instanceof InstallSnapshot) { handleInstallSnapshot(sender, (InstallSnapshot) rpc); restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); return this; } if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) { restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); } return super.handleMessage(sender, rpc); }
Follower extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout || message instanceof TimeoutNow) { return handleElectionTimeout(message); } if (appendEntriesMessageAssembler.handleMessage(message, actor())) { return this; } if (!(message instanceof RaftRPC)) { return null; } final RaftRPC rpc = (RaftRPC) message; if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); } if (rpc instanceof InstallSnapshot) { handleInstallSnapshot(sender, (InstallSnapshot) rpc); restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); return this; } if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) { restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); } return super.handleMessage(sender, rpc); } }
Follower extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout || message instanceof TimeoutNow) { return handleElectionTimeout(message); } if (appendEntriesMessageAssembler.handleMessage(message, actor())) { return this; } if (!(message instanceof RaftRPC)) { return null; } final RaftRPC rpc = (RaftRPC) message; if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); } if (rpc instanceof InstallSnapshot) { handleInstallSnapshot(sender, (InstallSnapshot) rpc); restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); return this; } if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) { restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); } return super.handleMessage(sender, rpc); } Follower(final RaftActorContext context);  Follower(final RaftActorContext context, final String initialLeaderId,
            final short initialLeaderPayloadVersion); }
Follower extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout || message instanceof TimeoutNow) { return handleElectionTimeout(message); } if (appendEntriesMessageAssembler.handleMessage(message, actor())) { return this; } if (!(message instanceof RaftRPC)) { return null; } final RaftRPC rpc = (RaftRPC) message; if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); } if (rpc instanceof InstallSnapshot) { handleInstallSnapshot(sender, (InstallSnapshot) rpc); restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); return this; } if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) { restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); } return super.handleMessage(sender, rpc); } Follower(final RaftActorContext context);  Follower(final RaftActorContext context, final String initialLeaderId,
            final short initialLeaderPayloadVersion); @Override final String getLeaderId(); @Override short getLeaderPayloadVersion(); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object message); @Override void close(); }
Follower extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout || message instanceof TimeoutNow) { return handleElectionTimeout(message); } if (appendEntriesMessageAssembler.handleMessage(message, actor())) { return this; } if (!(message instanceof RaftRPC)) { return null; } final RaftRPC rpc = (RaftRPC) message; if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); } if (rpc instanceof InstallSnapshot) { handleInstallSnapshot(sender, (InstallSnapshot) rpc); restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); return this; } if (!(rpc instanceof RequestVote) || canGrantVote((RequestVote) rpc)) { restartLastLeaderMessageTimer(); scheduleElection(electionDuration()); } return super.handleMessage(sender, rpc); } Follower(final RaftActorContext context);  Follower(final RaftActorContext context, final String initialLeaderId,
            final short initialLeaderPayloadVersion); @Override final String getLeaderId(); @Override short getLeaderPayloadVersion(); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object message); @Override void close();  }