Leader extends AbstractLeader { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object originalMessage) { requireNonNull(sender, "sender should not be null"); if (ISOLATED_LEADER_CHECK.equals(originalMessage)) { if (isLeaderIsolated()) { log.warn("{}: At least {} followers need to be active, Switching {} from Leader to IsolatedLeader", context.getId(), getMinIsolatedLeaderPeerCount(), getLeaderId()); return internalSwitchBehavior(new IsolatedLeader(context, this)); } else { return this; } } else { return super.handleMessage(sender, originalMessage); } } Leader(final RaftActorContext context, @Nullable final AbstractLeader initializeFromLeader);  Leader(final RaftActorContext context); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object originalMessage); void transferLeadership(@NonNull final RaftActorLeadershipTransferCohort leadershipTransferCohort); @Override void close();  }
@Test public void testHandleMessageForUnknownMessage() { logStart("testHandleMessageForUnknownMessage"); leader = new Leader(createActorContext()); assertNull(leader.handleMessage(followerActor, "foo")); }
@Test public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() { logStart("testThatLeaderSendsAHeartbeatMessageToAllFollowers"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setCommitIndex(-1); actorContext.setPayloadVersion(payloadVersion); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); actorContext.setCurrentBehavior(leader); final long lastIndex = actorContext.getReplicatedLog().lastIndex(); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getTerm", term, appendEntries.getTerm()); assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", -1, appendEntries.getPrevLogTerm()); assertEquals("Entries size", 0, appendEntries.getEntries().size()); assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion()); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex - 1, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams() .getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getPrevLogIndex", lastIndex - 1, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm()); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", lastIndex, appendEntries.getEntries().get(0).getIndex()); assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm()); assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion()); }
@Test public void testHandleReplicateMessageSendAppendEntriesToFollower() { logStart("testHandleReplicateMessageSendAppendEntriesToFollower"); MockRaftActorContext actorContext = createActorContextWithFollower(); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1); assertTrue(raftBehavior instanceof Leader); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm()); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex()); assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm()); assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString()); assertEquals("Commit Index", lastIndex, actorContext.getCommitIndex()); }
@Test public void testHandleReplicateMessageWithHigherTermThanPreviousEntry() { logStart("testHandleReplicateMessageWithHigherTermThanPreviousEntry"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setCommitIndex(-1); actorContext.setLastApplied(-1); long prevTerm = actorContext.getTermInformation().getCurrentTerm(); long newTerm = prevTerm + 1; actorContext.getTermInformation().update(newTerm, ""); leader = new Leader(actorContext); actorContext.setCurrentBehavior(leader); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, newTerm, true, lastIndex, prevTerm, (short)0)); assertEquals("Commit Index", -1, actorContext.getCommitIndex()); followerActor.underlyingActor().clear(); long newIndex = lastIndex + 1; sendReplicate(actorContext, newTerm, newIndex); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", prevTerm, appendEntries.getPrevLogTerm()); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", newIndex, appendEntries.getEntries().get(0).getIndex()); assertEquals("Entry getTerm", newTerm, appendEntries.getEntries().get(0).getTerm()); assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString()); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, newTerm, true, newIndex, newTerm, (short)0)); assertEquals("Commit Index", newIndex, actorContext.getCommitIndex()); }
@Test public void testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus() { logStart("testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setRaftPolicy(createRaftPolicy(true, true)); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short) 0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1); assertTrue(raftBehavior instanceof Leader); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm()); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex()); assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm()); assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString()); assertEquals("Commit Index", lastIndex + 1, actorContext.getCommitIndex()); }
@Test public void testMultipleReplicateShouldNotCauseDuplicateAppendEntriesToBeSent() { logStart("testHandleReplicateMessageSendAppendEntriesToFollower"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getHeartBeatInterval() { return FiniteDuration.apply(5, TimeUnit.SECONDS); } }); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); for (int i = 0; i < 5; i++) { sendReplicate(actorContext, lastIndex + i + 1); } List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of append entries collected should be 1", 1, allMessages.size()); }
@Test public void testMultipleReplicateWithReplyShouldResultInAppendEntries() { logStart("testMultipleReplicateWithReplyShouldResultInAppendEntries"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getHeartBeatInterval() { return FiniteDuration.apply(5, TimeUnit.SECONDS); } }); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); for (int i = 0; i < 3; i++) { sendReplicate(actorContext, lastIndex + i + 1); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex + i + 1, term, (short)0)); } List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of request/consensus appends collected", 6, allMessages.size()); for (int i = 0; i < 3; i++) { assertRequestEntry(lastIndex, allMessages, i); assertCommitEntry(lastIndex, allMessages, i); } sendReplicate(actorContext, lastIndex + 3 + 1); allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of request entries collected", 7, allMessages.size()); assertRequestEntry(lastIndex, allMessages, 3); sendReplicate(actorContext, lastIndex + 4 + 1); allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of request entries collected", 7, allMessages.size()); }
@Test public void testDuplicateAppendEntriesWillBeSentOnHeartBeat() { logStart("testDuplicateAppendEntriesWillBeSentOnHeartBeat"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getHeartBeatInterval() { return FiniteDuration.apply(500, TimeUnit.MILLISECONDS); } }); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); sendReplicate(actorContext, lastIndex + 1); Uninterruptibles.sleepUninterruptibly(750, TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of append entries collected should be 2", 2, allMessages.size()); assertEquals(1, allMessages.get(0).getEntries().size()); assertEquals(lastIndex + 1, allMessages.get(0).getEntries().get(0).getIndex()); assertEquals(1, allMessages.get(1).getEntries().size()); assertEquals(lastIndex + 1, allMessages.get(0).getEntries().get(0).getIndex()); }
@Test public void testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed() { logStart("testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getHeartBeatInterval() { return FiniteDuration.apply(100, TimeUnit.MILLISECONDS); } }); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); for (int i = 0; i < 3; i++) { Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); } List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of append entries collected should be 3", 3, allMessages.size()); }
@Test public void testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate() { logStart("testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate"); MockRaftActorContext actorContext = createActorContextWithFollower(); actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public FiniteDuration getHeartBeatInterval() { return FiniteDuration.apply(100, TimeUnit.MILLISECONDS); } }); long term = 1; actorContext.getTermInformation().update(term, ""); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); long lastIndex = actorContext.getReplicatedLog().lastIndex(); leader.handleMessage(followerActor, new AppendEntriesReply( FOLLOWER_ID, term, true, lastIndex, term, (short)0)); assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive()); followerActor.underlyingActor().clear(); Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); sendReplicate(actorContext, lastIndex + 1); List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class); assertEquals("The number of append entries collected should be 2", 2, allMessages.size()); assertEquals(0, allMessages.get(0).getEntries().size()); assertEquals(1, allMessages.get(1).getEntries().size()); }
@Test public void testHandleReplicateMessageWhenThereAreNoFollowers() { logStart("testHandleReplicateMessageWhenThereAreNoFollowers"); MockRaftActorContext actorContext = createActorContext(); leader = new Leader(actorContext); actorContext.setLastApplied(0); long newLogIndex = actorContext.getReplicatedLog().lastIndex() + 1; long term = actorContext.getTermInformation().getCurrentTerm(); ReplicatedLogEntry newEntry = new SimpleReplicatedLogEntry( newLogIndex, term, new MockRaftActorContext.MockPayload("foo")); actorContext.getReplicatedLog().append(newEntry); final Identifier id = new MockIdentifier("state-id"); RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new Replicate(leaderActor, id, newEntry, true)); assertTrue(raftBehavior instanceof Leader); assertEquals("getCommitIndex", newLogIndex, actorContext.getCommitIndex()); List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching( leaderActor, ApplyState.class); assertEquals("ApplyState count", newLogIndex, applyStateList.size()); for (int i = 0; i <= newLogIndex - 1; i++) { ApplyState applyState = applyStateList.get(i); assertEquals("getIndex", i + 1, applyState.getReplicatedLogEntry().getIndex()); assertEquals("getTerm", term, applyState.getReplicatedLogEntry().getTerm()); } ApplyState last = applyStateList.get((int) newLogIndex - 1); assertEquals("getData", newEntry.getData(), last.getReplicatedLogEntry().getData()); assertEquals("getIdentifier", id, last.getIdentifier()); }
@Test public void testInstallSnapshot() { logStart("testInstallSnapshot"); final MockRaftActorContext actorContext = createActorContextWithFollower(); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); actorContext.getReplicatedLog().removeFrom(0); final int lastAppliedIndex = 3; final int snapshotIndex = 2; final int snapshotTerm = 1; final int currentTerm = 2; actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); actorContext.setCommitIndex(lastAppliedIndex); actorContext.setLastApplied(lastAppliedIndex); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(0); byte[] bytes = toByteString(leadersSnapshot).toByteArray(); Snapshot snapshot = Snapshot.create(ByteState.of(bytes), Collections.<ReplicatedLogEntry>emptyList(), lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm, -1, null, null); RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot, ByteSource.wrap(bytes))); assertTrue(raftBehavior instanceof Leader); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertNotNull(installSnapshot.getData()); assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex()); assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm()); assertEquals(currentTerm, installSnapshot.getTerm()); }
@Test public void testForceInstallSnapshot() { logStart("testForceInstallSnapshot"); final MockRaftActorContext actorContext = createActorContextWithFollower(); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); final int lastAppliedIndex = 3; final int snapshotIndex = -1; final int snapshotTerm = -1; final int currentTerm = 2; actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); actorContext.setCommitIndex(lastAppliedIndex); actorContext.setLastApplied(lastAppliedIndex); leader = new Leader(actorContext); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(-1); byte[] bytes = toByteString(leadersSnapshot).toByteArray(); Snapshot snapshot = Snapshot.create(ByteState.of(bytes), Collections.<ReplicatedLogEntry>emptyList(), lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm, -1, null, null); RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot, ByteSource.wrap(bytes))); assertTrue(raftBehavior instanceof Leader); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertNotNull(installSnapshot.getData()); assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex()); assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm()); assertEquals(currentTerm, installSnapshot.getTerm()); }
@Test public void testHandleInstallSnapshotReplyLastChunk() throws Exception { logStart("testHandleInstallSnapshotReplyLastChunk"); MockRaftActorContext actorContext = createActorContextWithFollower(); final int commitIndex = 3; final int snapshotIndex = 2; final int snapshotTerm = 1; final int currentTerm = 2; actorContext.setCommitIndex(commitIndex); leader = new Leader(actorContext); actorContext.setCurrentBehavior(leader); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(0); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); ByteString bs = toByteString(leadersSnapshot); leader.setSnapshotHolder(new SnapshotHolder(Snapshot.create(ByteState.of(bs.toByteArray()), Collections.<ReplicatedLogEntry>emptyList(), commitIndex, snapshotTerm, commitIndex, snapshotTerm, -1, null, null), ByteSource.wrap(bs.toByteArray()))); LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState( actorContext.getConfigParams().getSnapshotChunkSize(), leader.logName()); fts.setSnapshotBytes(ByteSource.wrap(bs.toByteArray())); leader.getFollower(FOLLOWER_ID).setLeaderInstallSnapshotState(fts); while (!fts.isLastChunk(fts.getChunkIndex())) { fts.getNextChunk(); fts.incrementChunkIndex(); } actorContext.getReplicatedLog().removeFrom(0); RaftActorBehavior raftBehavior = leader.handleMessage(followerActor, new InstallSnapshotReply(currentTerm, FOLLOWER_ID, fts.getChunkIndex(), true)); assertTrue(raftBehavior instanceof Leader); assertEquals(1, leader.followerLogSize()); FollowerLogInformation fli = leader.getFollower(FOLLOWER_ID); assertNotNull(fli); assertNull(fli.getInstallSnapshotState()); assertEquals(commitIndex, fli.getMatchIndex()); assertEquals(commitIndex + 1, fli.getNextIndex()); assertFalse(leader.hasSnapshot()); }
@Test public void testSendSnapshotfromInstallSnapshotReply() { logStart("testSendSnapshotfromInstallSnapshotReply"); MockRaftActorContext actorContext = createActorContextWithFollower(); final int commitIndex = 3; final int snapshotIndex = 2; final int snapshotTerm = 1; final int currentTerm = 2; DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl() { @Override public int getSnapshotChunkSize() { return 50; } }; configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS)); configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS)); actorContext.setConfigParams(configParams); actorContext.setCommitIndex(commitIndex); leader = new Leader(actorContext); actorContext.setCurrentBehavior(leader); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(0); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); ByteString bs = toByteString(leadersSnapshot); Snapshot snapshot = Snapshot.create(ByteState.of(bs.toByteArray()), Collections.<ReplicatedLogEntry>emptyList(), commitIndex, snapshotTerm, commitIndex, snapshotTerm, -1, null, null); leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot, ByteSource.wrap(bs.toByteArray()))); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(1, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); followerActor.underlyingActor().clear(); leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), FOLLOWER_ID, installSnapshot.getChunkIndex(), true)); installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(2, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); followerActor.underlyingActor().clear(); leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), FOLLOWER_ID, installSnapshot.getChunkIndex(), true)); installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); followerActor.underlyingActor().clear(); leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), FOLLOWER_ID, installSnapshot.getChunkIndex(), true)); installSnapshot = MessageCollectorActor.getFirstMatching(followerActor, InstallSnapshot.class); assertNull(installSnapshot); }
@Test public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() { logStart("testHandleInstallSnapshotReplyWithInvalidChunkIndex"); MockRaftActorContext actorContext = createActorContextWithFollower(); final int commitIndex = 3; final int snapshotIndex = 2; final int snapshotTerm = 1; final int currentTerm = 2; actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public int getSnapshotChunkSize() { return 50; } }); actorContext.setCommitIndex(commitIndex); leader = new Leader(actorContext); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(0); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); ByteString bs = toByteString(leadersSnapshot); Snapshot snapshot = Snapshot.create(ByteState.of(bs.toByteArray()), Collections.<ReplicatedLogEntry>emptyList(), commitIndex, snapshotTerm, commitIndex, snapshotTerm, -1, null, null); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot, ByteSource.wrap(bs.toByteArray()))); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(1, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); followerActor.underlyingActor().clear(); leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), FOLLOWER_ID, -1, false)); Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(1, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); }
@Test public void testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk() { logStart("testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk"); MockRaftActorContext actorContext = createActorContextWithFollower(); final int commitIndex = 3; final int snapshotIndex = 2; final int snapshotTerm = 1; final int currentTerm = 2; actorContext.setConfigParams(new DefaultConfigParamsImpl() { @Override public int getSnapshotChunkSize() { return 50; } }); actorContext.setCommitIndex(commitIndex); leader = new Leader(actorContext); leader.getFollower(FOLLOWER_ID).setMatchIndex(-1); leader.getFollower(FOLLOWER_ID).setNextIndex(0); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex); actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm); actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString()); ByteString bs = toByteString(leadersSnapshot); Snapshot snapshot = Snapshot.create(ByteState.of(bs.toByteArray()), Collections.<ReplicatedLogEntry>emptyList(), commitIndex, snapshotTerm, commitIndex, snapshotTerm, -1, null, null); leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot, ByteSource.wrap(bs.toByteArray()))); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(1, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); assertEquals(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE, installSnapshot.getLastChunkHashCode().getAsInt()); final int hashCode = Arrays.hashCode(installSnapshot.getData()); followerActor.underlyingActor().clear(); leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(), FOLLOWER_ID, 1, true)); installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class); assertEquals(2, installSnapshot.getChunkIndex()); assertEquals(3, installSnapshot.getTotalChunks()); assertEquals(hashCode, installSnapshot.getLastChunkHashCode().getAsInt()); }
@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); 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()); }
@Test public void testHandleAppendEntriesReplyFailureWithFollowersLogEmpty() { logStart("testHandleAppendEntriesReplyFailureWithFollowersLogEmpty"); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( new FiniteDuration(1000, TimeUnit.SECONDS)); leaderActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build()); long leaderCommitIndex = 1; leaderActorContext.setCommitIndex(leaderCommitIndex); leaderActorContext.setLastApplied(leaderCommitIndex); final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0); final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1); 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); assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit()); assertEquals("Log entries size", 0, appendEntries.getEntries().size()); assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex()); leaderActor.underlyingActor().setBehavior(leader); leaderActorContext.setCurrentBehavior(leader); leader.handleMessage(followerActor, appendEntriesReply); MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); 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()); FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID); assertEquals("getNextIndex", 2, followerInfo.getNextIndex()); List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2); ApplyState applyState = applyStateList.get(0); assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex()); assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm()); assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(), applyState.getReplicatedLogEntry().getData()); applyState = applyStateList.get(1); assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex()); assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm()); assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(), applyState.getReplicatedLogEntry().getData()); assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex()); assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex()); }
@Test public void testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent() { logStart("testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent"); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( new FiniteDuration(1000, TimeUnit.SECONDS)); leaderActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build()); long leaderCommitIndex = 1; leaderActorContext.setCommitIndex(leaderCommitIndex); leaderActorContext.setLastApplied(leaderCommitIndex); final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0); final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1); MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader(); followerActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).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); assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit()); assertEquals("Log entries size", 0, appendEntries.getEntries().size()); assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex()); leaderActor.underlyingActor().setBehavior(leader); leaderActorContext.setCurrentBehavior(leader); leader.handleMessage(followerActor, appendEntriesReply); MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); 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 term", 2, appendEntries.getEntries().get(0).getTerm()); assertEquals("First entry data", leadersFirstLogEntry.getData(), appendEntries.getEntries().get(0).getData()); assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex()); assertEquals("Second entry term", 2, appendEntries.getEntries().get(1).getTerm()); assertEquals("Second entry data", leadersSecondLogEntry.getData(), appendEntries.getEntries().get(1).getData()); FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID); assertEquals("getNextIndex", 2, followerInfo.getNextIndex()); List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2); ApplyState applyState = applyStateList.get(0); assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex()); assertEquals("Follower's first ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm()); assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(), applyState.getReplicatedLogEntry().getData()); applyState = applyStateList.get(1); assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex()); assertEquals("Follower's second ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm()); assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(), applyState.getReplicatedLogEntry().getData()); assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex()); assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex()); assertEquals("Follower's lastTerm", 2, followerActorContext.getReplicatedLog().lastTerm()); }
@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); 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()); }
@Test public void testIsolatedLeaderCheckNoFollowers() { logStart("testIsolatedLeaderCheckNoFollowers"); MockRaftActorContext leaderActorContext = createActorContext(); leader = new Leader(leaderActorContext); RaftActorBehavior newBehavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK); assertTrue(newBehavior instanceof Leader); }
@Test public void testLaggingFollowerStarvation() { logStart("testLaggingFollowerStarvation"); String leaderActorId = actorFactory.generateActorId("leader"); String follower1ActorId = actorFactory.generateActorId("follower"); String follower2ActorId = actorFactory.generateActorId("follower"); final ActorRef follower1Actor = actorFactory.createActor(MessageCollectorActor.props(), follower1ActorId); final ActorRef follower2Actor = actorFactory.createActor(MessageCollectorActor.props(), follower2ActorId); MockRaftActorContext leaderActorContext = new MockRaftActorContext(leaderActorId, getSystem(), leaderActor); DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(200, TimeUnit.MILLISECONDS)); configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS)); leaderActorContext.setConfigParams(configParams); leaderActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(1,5,1).build()); Map<String, String> peerAddresses = new HashMap<>(); peerAddresses.put(follower1ActorId, follower1Actor.path().toString()); peerAddresses.put(follower2ActorId, follower2Actor.path().toString()); leaderActorContext.setPeerAddresses(peerAddresses); leaderActorContext.getTermInformation().update(1, leaderActorId); leader = createBehavior(leaderActorContext); leaderActor.underlyingActor().setBehavior(leader); for (int i = 1; i < 6; i++) { RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor, new AppendEntriesReply(follower1ActorId, 1, true, i, 1, (short)0)); assertTrue(newBehavior == leader); Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS); } List<SendHeartBeat> heartbeats = MessageCollectorActor.getAllMatching(leaderActor, SendHeartBeat.class); assertTrue(String.format("%s heartbeat(s) is less than expected", heartbeats.size()), heartbeats.size() > 1); List<AppendEntries> appendEntries = MessageCollectorActor.getAllMatching(follower2Actor, AppendEntries.class); assertTrue(String.format("%s append entries is less than expected", appendEntries.size()), appendEntries.size() > 1); }
@Test public void testReplicationConsensusWithNonVotingFollower() { logStart("testReplicationConsensusWithNonVotingFollower"); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( new FiniteDuration(1000, TimeUnit.SECONDS)); leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); leaderActorContext.setCommitIndex(-1); leaderActorContext.setLastApplied(-1); String nonVotingFollowerId = "nonvoting-follower"; ActorRef nonVotingFollowerActor = actorFactory.createActor( MessageCollectorActor.props(), actorFactory.generateActorId(nonVotingFollowerId)); leaderActorContext.addToPeers(nonVotingFollowerId, nonVotingFollowerActor.path().toString(), VotingState.NON_VOTING); leader = new Leader(leaderActorContext); leaderActorContext.setCurrentBehavior(leader); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class); MessageCollectorActor.clearMessages(followerActor); MessageCollectorActor.clearMessages(nonVotingFollowerActor); MessageCollectorActor.clearMessages(leaderActor); sendReplicate(leaderActorContext, 0); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class); leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0)); MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class); leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 0, 1, (short)0)); MessageCollectorActor.clearMessages(followerActor); MessageCollectorActor.clearMessages(nonVotingFollowerActor); MessageCollectorActor.clearMessages(leaderActor); sendReplicate(leaderActorContext, 1); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class); assertEquals("Log entries size", 1, appendEntries.getEntries().size()); assertEquals("Log entry index", 1, appendEntries.getEntries().get(0).getIndex()); leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 1, 1, (short)0)); MessageCollectorActor.assertNoneMatching(leaderActor, ApplyState.class, 500); leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 1, 1, (short)0)); MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class); }
@Test public void testReplicationWithPayloadSizeThatExceedsThreshold() { logStart("testReplicationWithPayloadSizeThatExceedsThreshold"); final int serializedSize = SerializationUtils.serialize(new AppendEntries(1, LEADER_ID, -1, -1, Arrays.asList(new SimpleReplicatedLogEntry(0, 1, new MockRaftActorContext.MockPayload("large"))), 0, -1, (short)0)).length; final MockRaftActorContext.MockPayload largePayload = new MockRaftActorContext.MockPayload("large", serializedSize); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( new FiniteDuration(300, TimeUnit.MILLISECONDS)); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setSnapshotChunkSize(serializedSize - 50); leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); leaderActorContext.setCommitIndex(-1); leaderActorContext.setLastApplied(-1); leader = new Leader(leaderActorContext); leaderActorContext.setCurrentBehavior(leader); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); leader.handleMessage(followerActor, new AppendEntriesReply(FOLLOWER_ID, -1, true, -1, -1, (short)0)); MessageCollectorActor.clearMessages(followerActor); final long term = leaderActorContext.getTermInformation().getCurrentTerm(); sendReplicate(leaderActorContext, term, 0); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", 0, appendEntries.getEntries().get(0).getIndex()); leader.handleMessage(followerActor, new AppendEntriesReply(FOLLOWER_ID, term, true, 0, term, (short)0)); assertEquals("getCommitIndex", 0, leaderActorContext.getCommitIndex()); MessageCollectorActor.clearMessages(followerActor); sendReplicate(leaderActorContext, term, 1, largePayload); MessageSlice messageSlice = MessageCollectorActor.expectFirstMatching(followerActor, MessageSlice.class); assertEquals("getSliceIndex", 1, messageSlice.getSliceIndex()); assertEquals("getTotalSlices", 2, messageSlice.getTotalSlices()); final Identifier slicingId = messageSlice.getIdentifier(); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex()); assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm()); assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit()); assertEquals("Entries size", 0, appendEntries.getEntries().size()); MessageCollectorActor.clearMessages(followerActor); Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams() .getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit()); assertEquals("Entries size", 0, appendEntries.getEntries().size()); MessageCollectorActor.clearMessages(followerActor); leader.handleMessage(followerActor, MessageSliceReply.success(slicingId, 1, followerActor)); messageSlice = MessageCollectorActor.expectFirstMatching(followerActor, MessageSlice.class); assertEquals("getSliceIndex", 2, messageSlice.getSliceIndex()); leader.handleMessage(followerActor, MessageSliceReply.success(slicingId, 2, followerActor)); leader.handleMessage(followerActor, new AppendEntriesReply(FOLLOWER_ID, term, true, 1, term, (short)0)); MessageCollectorActor.clearMessages(followerActor); sendReplicate(leaderActorContext, term, 2); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("Entries size", 1, appendEntries.getEntries().size()); assertEquals("Entry getIndex", 2, appendEntries.getEntries().get(0).getIndex()); assertEquals("getLeaderCommit", 1, appendEntries.getLeaderCommit()); }
@Test public void testLargePayloadSlicingExpiration() { logStart("testLargePayloadSlicingExpiration"); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( new FiniteDuration(100, TimeUnit.MILLISECONDS)); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setSnapshotChunkSize(10); leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); leaderActorContext.setCommitIndex(-1); leaderActorContext.setLastApplied(-1); final long term = leaderActorContext.getTermInformation().getCurrentTerm(); leader = new Leader(leaderActorContext); leaderActorContext.setCurrentBehavior(leader); MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); leader.handleMessage(followerActor, new AppendEntriesReply(FOLLOWER_ID, -1, true, -1, -1, (short)0)); MessageCollectorActor.clearMessages(followerActor); sendReplicate(leaderActorContext, term, 0, new MockRaftActorContext.MockPayload("large", leaderActorContext.getConfigParams().getSnapshotChunkSize() + 1)); MessageCollectorActor.expectFirstMatching(followerActor, MessageSlice.class); Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams() .getElectionTimeOutInterval().toMillis() * 3 + 50, TimeUnit.MILLISECONDS); MessageCollectorActor.clearMessages(followerActor); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit()); assertEquals("Entries size", 0, appendEntries.getEntries().size()); MessageCollectorActor.assertNoneMatching(followerActor, MessageSlice.class, 300); MessageCollectorActor.clearMessages(followerActor); Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams() .getHeartBeatInterval().toMillis() + 50, TimeUnit.MILLISECONDS); leader.handleMessage(followerActor, new AppendEntriesReply(FOLLOWER_ID, term, true, -1, term, (short)0)); MessageCollectorActor.expectFirstMatching(followerActor, MessageSlice.class); }
@Test public void testLeaderAddressInAppendEntries() { logStart("testLeaderAddressInAppendEntries"); MockRaftActorContext leaderActorContext = createActorContextWithFollower(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval( FiniteDuration.create(50, TimeUnit.MILLISECONDS)); leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); leaderActorContext.setCommitIndex(-1); leaderActorContext.setLastApplied(-1); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setPeerAddressResolver( peerId -> leaderActor.path().toString()); leader = new Leader(leaderActorContext); leaderActorContext.setCurrentBehavior(leader); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertFalse(appendEntries.getLeaderAddress().isPresent()); MessageCollectorActor.clearMessages(followerActor); leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0, false, true, RaftVersions.CURRENT_VERSION)); Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams() .getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertTrue(appendEntries.getLeaderAddress().isPresent()); assertEquals(leaderActor.path().toString(), appendEntries.getLeaderAddress().get()); MessageCollectorActor.clearMessages(followerActor); leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0, false, false, RaftVersions.CURRENT_VERSION)); Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams() .getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE); appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertFalse(appendEntries.getLeaderAddress().isPresent()); }