Leader extends AbstractLeader { @Override public void close() { if (leadershipTransferContext != null) { LeadershipTransferContext localLeadershipTransferContext = leadershipTransferContext; leadershipTransferContext = null; localLeadershipTransferContext.transferCohort.abortTransfer(); } super.close(); } 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 testLeaderInstallSnapshotState() throws IOException { logStart("testLeaderInstallSnapshotState"); Map<String, String> leadersSnapshot = new HashMap<>(); leadersSnapshot.put("1", "A"); leadersSnapshot.put("2", "B"); leadersSnapshot.put("3", "C"); ByteString bs = toByteString(leadersSnapshot); byte[] barray = bs.toByteArray(); LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState(50, "test"); fts.setSnapshotBytes(ByteSource.wrap(barray)); assertEquals(bs.size(), barray.length); int chunkIndex = 0; for (int i = 0; i < barray.length; i = i + 50) { int length = i + 50; chunkIndex++; if (i + 50 > barray.length) { length = barray.length; } byte[] chunk = fts.getNextChunk(); assertEquals("bytestring size not matching for chunk:" + chunkIndex, length - i, chunk.length); assertEquals("chunkindex not matching", chunkIndex, fts.getChunkIndex()); fts.markSendStatus(true); if (!fts.isLastChunk(chunkIndex)) { fts.incrementChunkIndex(); } } assertEquals("totalChunks not matching", chunkIndex, fts.getTotalChunks()); fts.close(); }
@Test public void testLeaderCreatedWithCommitIndexLessThanLastIndex() { logStart("testLeaderCreatedWithCommitIndexLessThanLastIndex"); final MockRaftActorContext leaderActorContext = createActorContextWithFollower(); MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor); Follower follower = new Follower(followerActorContext); followerActor.underlyingActor().setBehavior(follower); followerActorContext.setCurrentBehavior(follower); Map<String, String> peerAddresses = new HashMap<>(); peerAddresses.put(FOLLOWER_ID, followerActor.path().toString()); leaderActorContext.setPeerAddresses(peerAddresses); leaderActorContext.getReplicatedLog().removeFrom(0); leaderActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build()); leaderActorContext.setCommitIndex(1); followerActorContext.getReplicatedLog().removeFrom(0); followerActorContext.setReplicatedLog( new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build()); followerActorContext.setCommitIndex(1); leader = new Leader(leaderActorContext); leaderActorContext.setCurrentBehavior(leader); AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class); assertEquals(-1, appendEntries.getLeaderCommit()); assertEquals(0, appendEntries.getEntries().size()); assertEquals(0, appendEntries.getPrevLogIndex()); AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching( leaderActor, AppendEntriesReply.class); assertEquals(2, appendEntriesReply.getLogLastIndex()); assertEquals(1, appendEntriesReply.getLogLastTerm()); assertEquals(2, appendEntriesReply.getLogLastIndex()); assertEquals(1, appendEntriesReply.getLogLastTerm()); follower.close(); }