+ public void testHandleInstallSnapshot() {
+ logStart("testHandleInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+ context.getTermInformation().update(1, "leader");
+
+ 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++;
+ }
+
+ ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
+ ApplySnapshot.class);
+ Snapshot snapshot = applySnapshot.getSnapshot();
+ assertNotNull(lastInstallSnapshot);
+ assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
+ assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
+ snapshot.getLastAppliedTerm());
+ assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
+ snapshot.getLastAppliedIndex());
+ assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
+ assertEquals("getState type", ByteState.class, snapshot.getState().getClass());
+ Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), ((ByteState)snapshot.getState()).getBytes());
+ assertEquals("getElectionTerm", 1, snapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", "leader", snapshot.getElectionVotedFor());
+ applySnapshot.getCallback().onSuccess();
+
+ List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
+ leaderActor, InstallSnapshotReply.class);
+ assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
+
+ chunkIndex = 1;
+ for (InstallSnapshotReply reply: replies) {
+ assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
+ assertEquals("getTerm", 1, reply.getTerm());
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
+ }
+
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
+ }
+
+ /**
+ * Verify that when an AppendEntries is sent to a follower during a snapshot install
+ * the Follower short-circuits the processing of the AppendEntries message.
+ */
+ @Test
+ public void testReceivingAppendEntriesDuringInstallSnapshot() {
+ logStart("testReceivingAppendEntriesDuringInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+
+ // Check that snapshot installation is not in progress
+ assertNull(follower.getSnapshotTracker());
+
+ // Make sure that we have more than 1 chunk to send
+ assertTrue(totalChunks > 1);
+
+ // Send an install snapshot with the first chunk to start the process of installing a snapshot
+ byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, 1, totalChunks));
+
+ // Check if snapshot installation is in progress now
+ assertNotNull(follower.getSnapshotTracker());
+
+ // Send an append entry
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 1, "3")), 2, -1, (short)1);
+
+ follower.handleMessage(leaderActor, appendEntries);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
+ assertEquals("getTerm", context.getTermInformation().getCurrentTerm(), reply.getTerm());
+
+ assertNotNull(follower.getSnapshotTracker());
+ }
+
+ @Test
+ public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() {
+ logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+
+ // Check that snapshot installation is not in progress
+ assertNull(follower.getSnapshotTracker());
+
+ // Make sure that we have more than 1 chunk to send
+ assertTrue(totalChunks > 1);
+
+ // Send an install snapshot with the first chunk to start the process of installing a snapshot
+ byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, 1, totalChunks));
+
+ // Check if snapshot installation is in progress now
+ assertNotNull(follower.getSnapshotTracker());
+
+ // Send appendEntries with a new term and leader.
+ AppendEntries appendEntries = new AppendEntries(2, "new-leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 2, "3")), 2, -1, (short)1);
+
+ follower.handleMessage(leaderActor, appendEntries);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", 2, reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", 2, reply.getLogLastTerm());
+ assertEquals("getTerm", 2, reply.getTerm());
+
+ assertNull(follower.getSnapshotTracker());
+ }
+
+ @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());
+
+ // Clear all the messages
+ 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"));
+
+ // The new commitIndex is 101
+ 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 testHandleOutOfSequenceInstallSnapshot() {
+ logStart("testHandleOutOfSequenceInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+
+ InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
+ getNextChunk(bsSnapshot, 10, 50), 3, 3);
+ follower.handleMessage(leaderActor, installSnapshot);
+
+ InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
+ InstallSnapshotReply.class);
+
+ assertEquals("isSuccess", false, reply.isSuccess());
+ assertEquals("getChunkIndex", -1, reply.getChunkIndex());
+ assertEquals("getTerm", 1, reply.getTerm());
+ assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
+
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
+ }
+
+ @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 testFollowerSchedulesElectionIfNonVoting() {
+ MockRaftActorContext context = createActorContext();
+ context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo(context.getId(), false))));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setHeartBeatInterval(
+ FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setElectionTimeoutFactor(1);
+
+ follower = new Follower(context, "leader", (short)1);
+
+ ElectionTimeout electionTimeout = MessageCollectorActor.expectFirstMatching(followerActor,
+ ElectionTimeout.class);
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), electionTimeout);
+ assertSame("handleMessage result", follower, newBehavior);
+ assertNull("Expected null leaderId", follower.getLeaderId());
+ }
+
+ @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));
+ }
+
+ @Test
+ public void testCaptureSnapshotOnLastEntryInAppendEntries() {
+ String id = "testCaptureSnapshotOnLastEntryInAppendEntries";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(2);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 1, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 1 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 1 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 1, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 1, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 1, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData()),
+ MockRaftActor.fromState(snapshot.getState()));
+ }
+
+ @Test
+ public void testCaptureSnapshotOnMiddleEntryInAppendEntries() {
+ String id = "testCaptureSnapshotOnMiddleEntryInAppendEntries";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(2);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
+ newReplicatedLogEntry(1, 2, "three"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 2, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 2 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 2 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 2, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 2, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
+ entries.get(2).getData()), MockRaftActor.fromState(snapshot.getState()));
+
+ assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
+ assertEquals("Snapshot index", 2, followerRaftActor.get().getReplicatedLog().getSnapshotIndex());
+
+ // Reinstate the actor from persistence
+
+ actorFactory.killActor(followerActorRef, new JavaTestKit(getSystem()));
+
+ followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
+ assertEquals("Last index", 2, followerRaftActor.get().getReplicatedLog().lastIndex());
+ assertEquals("Last applied index", 2, followerRaftActor.get().getRaftActorContext().getLastApplied());
+ assertEquals("Commit index", 2, followerRaftActor.get().getRaftActorContext().getCommitIndex());
+ assertEquals("State", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
+ entries.get(2).getData()), followerRaftActor.get().getState());
+ }
+
+ @Test
+ public void testCaptureSnapshotOnAppendEntriesWithUnapplied() {
+ String id = "testCaptureSnapshotOnAppendEntriesWithUnapplied";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(1);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
+ newReplicatedLogEntry(1, 2, "three"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 0, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 0 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 0 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 0, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 2, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot unapplied entry index", 1, snapshot.getUnAppliedEntries().get(0).getIndex());
+ assertEquals("Snapshot unapplied entry index", 2, snapshot.getUnAppliedEntries().get(1).getIndex());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 0, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData()),
+ MockRaftActor.fromState(snapshot.getState()));
+ }
+
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private static RaftActorSnapshotCohort newRaftActorSnapshotCohort(
+ final AtomicReference<MockRaftActor> followerRaftActor) {
+ RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
+ @Override
+ public void createSnapshot(final ActorRef actorRef,
+ final java.util.Optional<OutputStream> installSnapshotStream) {
+ try {
+ actorRef.tell(new CaptureSnapshotReply(new MockSnapshotState(followerRaftActor.get().getState()),
+ installSnapshotStream), actorRef);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);