Follower extends AbstractRaftActorBehavior { private void handleInstallSnapshot(final ActorRef sender, final InstallSnapshot installSnapshot) { log.debug("{}: handleInstallSnapshot: {}", logName(), installSnapshot); leaderId = installSnapshot.getLeaderId(); if (snapshotTracker == null) { snapshotTracker = new SnapshotTracker(log, installSnapshot.getTotalChunks(), installSnapshot.getLeaderId(), context); } updateInitialSyncStatus(installSnapshot.getLastIncludedIndex(), installSnapshot.getLeaderId()); try { final InstallSnapshotReply reply = new InstallSnapshotReply( currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true); if (snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(), installSnapshot.getLastChunkHashCode())) { log.info("{}: Snapshot installed from leader: {}", logName(), installSnapshot.getLeaderId()); Snapshot snapshot = Snapshot.create( context.getSnapshotManager().convertSnapshot(snapshotTracker.getSnapshotBytes()), new ArrayList<>(), installSnapshot.getLastIncludedIndex(), installSnapshot.getLastIncludedTerm(), installSnapshot.getLastIncludedIndex(), installSnapshot.getLastIncludedTerm(), context.getTermInformation().getCurrentTerm(), context.getTermInformation().getVotedFor(), installSnapshot.getServerConfig().orElse(null)); ApplySnapshot.Callback applySnapshotCallback = new ApplySnapshot.Callback() { @Override public void onSuccess() { log.debug("{}: handleInstallSnapshot returning: {}", logName(), reply); sender.tell(reply, actor()); } @Override public void onFailure() { sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), -1, false), actor()); } }; actor().tell(new ApplySnapshot(snapshot, applySnapshotCallback), actor()); closeSnapshotTracker(); } else { log.debug("{}: handleInstallSnapshot returning: {}", logName(), reply); sender.tell(reply, actor()); } } catch (IOException e) { log.debug("{}: Exception in InstallSnapshot of follower", logName(), e); sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), -1, false), actor()); closeSnapshotTracker(); } } 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 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()); }