@Test public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() { logStart("testInitialSyncUpWithHandleInstallSnapshot"); MockRaftActorContext context = createActorContext(); context.setCommitIndex(-1); follower = createBehavior(context); ByteString bsSnapshot = createSnapshot(); int offset = 0; int snapshotLength = bsSnapshot.size(); int chunkSize = 50; int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0); int lastIncludedIndex = 1; int chunkIndex = 1; InstallSnapshot lastInstallSnapshot = null; for (int i = 0; i < totalChunks; i++) { byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize); lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1, chunkData, chunkIndex, totalChunks); follower.handleMessage(leaderActor, lastInstallSnapshot); offset = offset + 50; lastIncludedIndex++; chunkIndex++; } FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertFalse(syncStatus.isInitialSyncDone()); MessageCollectorActor.clearMessages(followerActor); context.setLastApplied(101); context.setCommitIndex(101); setLastLogEntry(context, 1, 101, new MockRaftActorContext.MockPayload("")); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101, (short)0); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertTrue(syncStatus.isInitialSyncDone()); }
@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();  }