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();  }
@Test public void testHandleElectionTimeoutWhenNoLeaderMessageReceived() { logStart("testHandleElectionTimeoutWhenNoLeaderMessageReceived"); MockRaftActorContext context = createActorContext(); follower = new Follower(context); Uninterruptibles.sleepUninterruptibly(context.getConfigParams().getElectionTimeOutInterval().toMillis(), TimeUnit.MILLISECONDS); RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE); assertTrue(raftBehavior instanceof Candidate); }
@Test public void testHandleElectionTimeoutWhenLeaderMessageReceived() { logStart("testHandleElectionTimeoutWhenLeaderMessageReceived"); MockRaftActorContext context = createActorContext(); ((DefaultConfigParamsImpl) context.getConfigParams()) .setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); ((DefaultConfigParamsImpl) context.getConfigParams()).setElectionTimeoutFactor(4); follower = new Follower(context); context.setCurrentBehavior(follower); Uninterruptibles.sleepUninterruptibly(context.getConfigParams() .getElectionTimeOutInterval().toMillis() - 100, TimeUnit.MILLISECONDS); follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(), -1, -1, (short) 1)); Uninterruptibles.sleepUninterruptibly(130, TimeUnit.MILLISECONDS); RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE); assertTrue(raftBehavior instanceof Follower); Uninterruptibles.sleepUninterruptibly(context.getConfigParams() .getElectionTimeOutInterval().toMillis() - 150, TimeUnit.MILLISECONDS); follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(), -1, -1, (short) 1)); Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS); raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE); assertTrue(raftBehavior instanceof Follower); }
@Test public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull() { logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull"); MockRaftActorContext context = createActorContext(); long term = 1000; context.getTermInformation().update(term, null); follower = createBehavior(context); follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999)); RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class); assertEquals("isVoteGranted", true, reply.isVoteGranted()); assertEquals("getTerm", term, reply.getTerm()); verify(follower).scheduleElection(any(FiniteDuration.class)); }
@Test public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId() { logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId"); MockRaftActorContext context = createActorContext(); long term = 1000; context.getTermInformation().update(term, "test"); follower = createBehavior(context); follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999)); RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class); assertEquals("isVoteGranted", false, reply.isVoteGranted()); verify(follower, never()).scheduleElection(any(FiniteDuration.class)); }
@Test public void testHandleFirstAppendEntries() { logStart("testHandleFirstAppendEntries"); MockRaftActorContext context = createActorContext(); context.getReplicatedLog().clear(0,2); context.getReplicatedLog().append(newReplicatedLogEntry(1,100, "bar")); context.getReplicatedLog().setSnapshotIndex(99); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); Assert.assertEquals(1, context.getReplicatedLog().size()); AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 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()); assertTrue("append entries reply should be true", reply.isSuccess()); }
@Test public void testHandleFirstAppendEntriesWithPrevIndexMinusOne() { logStart("testHandleFirstAppendEntries"); MockRaftActorContext context = createActorContext(); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 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()); }
@Test public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog() { logStart("testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog"); MockRaftActorContext context = createActorContext(); context.getReplicatedLog().clear(0,2); context.getReplicatedLog().append(newReplicatedLogEntry(1, 100, "bar")); context.getReplicatedLog().setSnapshotIndex(99); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 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()); assertTrue("append entries reply should be true", reply.isSuccess()); }
@Test public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot() { logStart("testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot"); MockRaftActorContext context = createActorContext(); context.getReplicatedLog().clear(0,2); context.getReplicatedLog().setSnapshotIndex(100); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 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()); assertTrue("append entries reply should be true", reply.isSuccess()); }
@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()); }
@Test public void testHandleSyncUpAppendEntries() { logStart("testHandleSyncUpAppendEntries"); MockRaftActorContext context = createActorContext(); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0); follower = createBehavior(context); follower.handleMessage(leaderActor, appendEntries); 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("")); entries = Arrays.asList(newReplicatedLogEntry(2, 101, "foo")); appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertTrue(syncStatus.isInitialSyncDone()); MessageCollectorActor.clearMessages(followerActor); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.getFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertNull(syncStatus); }
@Test public void testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete() { logStart("testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete"); MockRaftActorContext context = createActorContext(); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0); follower = createBehavior(context); follower.handleMessage(leaderActor, appendEntries); FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertFalse(syncStatus.isInitialSyncDone()); MessageCollectorActor.clearMessages(followerActor); context.setLastApplied(100); setLastLogEntry(context, 1, 100, new MockRaftActorContext.MockPayload("")); entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertFalse(syncStatus.isInitialSyncDone()); }
@Test public void testHandleAppendEntriesLeaderChangedAfterSyncUpComplete() { logStart("testHandleAppendEntriesLeaderChangedAfterSyncUpComplete"); MockRaftActorContext context = createActorContext(); List<ReplicatedLogEntry> entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0); follower = createBehavior(context); follower.handleMessage(leaderActor, appendEntries); 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("")); entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertTrue(syncStatus.isInitialSyncDone()); MessageCollectorActor.clearMessages(followerActor); context.setLastApplied(100); setLastLogEntry(context, 1, 100, new MockRaftActorContext.MockPayload("")); entries = Arrays.asList( newReplicatedLogEntry(2, 101, "foo")); appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0); follower.handleMessage(leaderActor, appendEntries); syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class); assertFalse(syncStatus.isInitialSyncDone()); }
@Test public void testHandleAppendEntriesWithNewerCommitIndex() { logStart("testHandleAppendEntriesWithNewerCommitIndex"); MockRaftActorContext context = createActorContext(); context.setLastApplied(100); setLastLogEntry(context, 1, 100, new MockRaftActorContext.MockPayload("")); context.getReplicatedLog().setSnapshotIndex(99); List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList( newReplicatedLogEntry(2, 101, "foo")); AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0); follower = createBehavior(context); follower.handleMessage(leaderActor, appendEntries); assertEquals("getLastApplied", 101L, context.getLastApplied()); }
@Test public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() { logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm"); MockRaftActorContext context = createActorContext(); AppendEntries appendEntries = new AppendEntries(2, "leader", 0, 2, Collections.emptyList(), 101, -1, (short)0); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class); assertEquals("isSuccess", false, reply.isSuccess()); }
@Test public void testHandleAppendEntriesSenderPrevLogIndexIsInTheSnapshot() { logStart("testHandleAppendEntriesSenderPrevLogIndexIsInTheSnapshot"); MockRaftActorContext context = createActorContext(); context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(5, 8, 3).build()); context.getReplicatedLog().setSnapshotIndex(4); context.getReplicatedLog().setSnapshotTerm(3); AppendEntries appendEntries = new AppendEntries(3, "leader", 1, 3, Collections.emptyList(), 8, -1, (short)0); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class); assertEquals("isSuccess", true, reply.isSuccess()); }
@Test public void testHandleAppendEntriesCorrectReceiverLogEntries() { logStart("testHandleAppendEntriesCorrectReceiverLogEntries"); MockRaftActorContext context = createActorContext(); context.getTermInformation().update(1, "test"); MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); log.append(newReplicatedLogEntry(1, 0, "zero")); log.append(newReplicatedLogEntry(1, 1, "one")); log.append(newReplicatedLogEntry(1, 2, "two")); context.setReplicatedLog(log); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(newReplicatedLogEntry(2, 2, "two-1")); entries.add(newReplicatedLogEntry(2, 3, "three")); AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); assertEquals("Next index", 4, log.last().getIndex() + 1); assertEquals("Entry 1 data", "one", log.get(1).getData().toString()); assertEquals("Entry 2", entries.get(0), log.get(2)); assertEquals("Entry 3", entries.get(1), log.get(3)); expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3); }
@Test public void testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot() { logStart("testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot"); MockRaftActorContext context = createActorContext(); context.getTermInformation().update(1, "test"); MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); log.append(newReplicatedLogEntry(1, 0, "zero")); log.append(newReplicatedLogEntry(1, 1, "one")); log.append(newReplicatedLogEntry(1, 2, "two")); context.setReplicatedLog(log); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(newReplicatedLogEntry(2, 2, "two-1")); entries.add(newReplicatedLogEntry(2, 3, "three")); AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0); context.setRaftPolicy(createRaftPolicy(false, true)); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); expectAndVerifyAppendEntriesReply(2, false, context.getId(), 1, 2, true); }
@Test public void testHandleAppendEntriesPreviousLogEntryMissing() { logStart("testHandleAppendEntriesPreviousLogEntryMissing"); final MockRaftActorContext context = createActorContext(); MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); log.append(newReplicatedLogEntry(1, 0, "zero")); log.append(newReplicatedLogEntry(1, 1, "one")); log.append(newReplicatedLogEntry(1, 2, "two")); context.setReplicatedLog(log); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(newReplicatedLogEntry(1, 4, "four")); AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2); }
@Test public void testHandleAppendEntriesWithExistingLogEntry() { logStart("testHandleAppendEntriesWithExistingLogEntry"); MockRaftActorContext context = createActorContext(); context.getTermInformation().update(1, "test"); MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); log.append(newReplicatedLogEntry(1, 0, "zero")); log.append(newReplicatedLogEntry(1, 1, "one")); context.setReplicatedLog(log); List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one")); follower = createBehavior(context); follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0)); assertEquals("Next index", 2, log.last().getIndex() + 1); assertEquals("Entry 1", entries.get(0), log.get(1)); expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1); entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two")); MessageCollectorActor.clearMessages(leaderActor); follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0)); assertEquals("Next index", 3, log.last().getIndex() + 1); assertEquals("Entry 1", entries.get(0), log.get(1)); assertEquals("Entry 2", entries.get(1), log.get(2)); expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2); }
@Test public void testHandleAppendEntriesAfterInstallingSnapshot() { logStart("testHandleAppendAfterInstallingSnapshot"); MockRaftActorContext context = createActorContext(); MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); log.setSnapshotIndex(3); log.setSnapshotTerm(1); context.setReplicatedLog(log); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(newReplicatedLogEntry(1, 4, "four")); AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0); follower = createBehavior(context); RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries); Assert.assertSame(follower, newBehavior); expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4); }
@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()); }
@Test public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() { MockRaftActorContext context = createActorContext(); Stopwatch stopwatch = Stopwatch.createStarted(); follower = createBehavior(context); TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class); long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS); assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis()); RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow); assertTrue("Expected Candidate", newBehavior instanceof Candidate); }
@Test public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() { MockRaftActorContext context = createActorContext(); context.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getElectionTimeOutInterval() { return FiniteDuration.apply(100, TimeUnit.MILLISECONDS); } }); context.setRaftPolicy(createRaftPolicy(false, false)); follower = createBehavior(context); TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class); RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow); assertSame("handleMessage result", follower, newBehavior); }
@Test public void testElectionScheduledWhenAnyRaftRPCReceived() { MockRaftActorContext context = createActorContext(); follower = createBehavior(context); follower.handleMessage(leaderActor, new RaftRPC() { private static final long serialVersionUID = 1L; @Override public long getTerm() { return 100; } }); verify(follower).scheduleElection(any(FiniteDuration.class)); }
@Test public void testElectionNotScheduledWhenNonRaftRPCMessageReceived() { MockRaftActorContext context = createActorContext(); follower = createBehavior(context); follower.handleMessage(leaderActor, "non-raft-rpc"); verify(follower, never()).scheduleElection(any(FiniteDuration.class)); }