import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Test
public void testOnSnapshotOfferWithServerConfiguration() {
    long electionTerm = 2;
    String electionVotedFor = "member-2";
    ServerConfigurationPayload serverPayload = new ServerConfigurationPayload(Arrays.asList(
                                                    new ServerInfo(localId, true),
                                                    new ServerInfo("follower1", true),
                                                    new ServerInfo("follower2", true)));

    MockSnapshotState snapshotState = new MockSnapshotState(Arrays.asList(new MockPayload("1")));
    Snapshot snapshot = Snapshot.create(snapshotState, Collections.<ReplicatedLogEntry>emptyList(),
            -1, -1, -1, -1, electionTerm, electionVotedFor, serverPayload);

    SnapshotMetadata metadata = new SnapshotMetadata("test", 6, 12345);
    SnapshotOffer snapshotOffer = new SnapshotOffer(metadata , snapshot);

    sendMessageToSupport(snapshotOffer);

    assertEquals("Journal log size", 0, context.getReplicatedLog().size());
    assertEquals("Election term", electionTerm, context.getTermInformation().getCurrentTerm());
    assertEquals("Election votedFor", electionVotedFor, context.getTermInformation().getVotedFor());
    assertTrue("Dynamic server configuration", context.isDynamicServerConfigurationInUse());
    assertEquals("Peer List", Sets.newHashSet("follower1", "follower2"),
        Sets.newHashSet(context.getPeerIds()));
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Override
protected final void handleRecover(final Object message) throws Exception {
    if (message instanceof RecoveryCompleted) {
        if (incarnation != null) {
            incarnation = incarnation + 1;
        } else {
            incarnation = 0;
        }

        this.localBucket = new LocalBucket<>(incarnation.intValue(), initialData);
        initialData = null;
        LOG.debug("{}: persisting new incarnation {}", persistenceId(), incarnation);
        persisting = true;
        saveSnapshot(incarnation);
    } else if (message instanceof SnapshotOffer) {
        incarnation = (Integer) ((SnapshotOffer)message).snapshot();
        LOG.debug("{}: recovered incarnation {}", persistenceId(), incarnation);
    } else {
        LOG.warn("{}: ignoring recovery message {}", persistenceId(), message);
    }
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Override
public Receive createReceiveRecover() {
    return receiveBuilder()
            .match(String.class, s -> putCmd(s.toString()))
            .match(RecoveryCompleted.class, s -> log.debug("done"))
            .match(SnapshotOffer.class, ss -> list = (List<String>) ss.snapshot())
            .build();
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
boolean handleRecoveryMessage(Object message, PersistentDataProvider persistentProvider) {
    log.trace("{}: handleRecoveryMessage: {}", context.getId(), message);

    anyDataRecovered = anyDataRecovered || !(message instanceof RecoveryCompleted);

    if (isMigratedSerializable(message)) {
        hasMigratedDataRecovered = true;
    }

    boolean recoveryComplete = false;
    if (message instanceof UpdateElectionTerm) {
        context.getTermInformation().update(((UpdateElectionTerm) message).getCurrentTerm(),
                ((UpdateElectionTerm) message).getVotedFor());
    } else if (message instanceof SnapshotOffer) {
        onRecoveredSnapshot((SnapshotOffer) message);
    } else if (message instanceof ReplicatedLogEntry) {
        onRecoveredJournalLogEntry((ReplicatedLogEntry) message);
    } else if (message instanceof ApplyJournalEntries) {
        onRecoveredApplyLogEntries(((ApplyJournalEntries) message).getToIndex());
    } else if (message instanceof DeleteEntries) {
        onDeleteEntries((DeleteEntries) message);
    } else if (message instanceof ServerConfigurationPayload) {
        context.updatePeerIds((ServerConfigurationPayload)message);
    } else if (message instanceof RecoveryCompleted) {
        recoveryComplete = true;
        onRecoveryCompletedMessage(persistentProvider);
    }

    return recoveryComplete;
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Test
public void testDataRecoveredWithPersistenceDisabled() {
    doNothing().when(mockCohort).applyRecoverySnapshot(anyObject());
    doReturn(false).when(mockPersistence).isRecoveryApplicable();
    doReturn(10L).when(mockPersistentProvider).getLastSequenceNumber();

    Snapshot snapshot = Snapshot.create(new MockSnapshotState(Arrays.asList(new MockPayload("1"))),
            Collections.<ReplicatedLogEntry>emptyList(), 3, 1, 3, 1, -1, null, null);
    SnapshotOffer snapshotOffer = new SnapshotOffer(new SnapshotMetadata("test", 6, 12345), snapshot);

    sendMessageToSupport(snapshotOffer);

    sendMessageToSupport(new UpdateElectionTerm(5, "member2"));

    sendMessageToSupport(new SimpleReplicatedLogEntry(4, 1, new MockRaftActorContext.MockPayload("4")));
    sendMessageToSupport(new SimpleReplicatedLogEntry(5, 1, new MockRaftActorContext.MockPayload("5")));

    sendMessageToSupport(new ApplyJournalEntries(4));

    sendMessageToSupport(new DeleteEntries(5));

    assertEquals("Journal log size", 0, context.getReplicatedLog().size());
    assertEquals("Last index", -1, context.getReplicatedLog().lastIndex());
    assertEquals("Last applied", -1, context.getLastApplied());
    assertEquals("Commit index", -1, context.getCommitIndex());
    assertEquals("Snapshot term", -1, context.getReplicatedLog().getSnapshotTerm());
    assertEquals("Snapshot index", -1, context.getReplicatedLog().getSnapshotIndex());

    assertEquals("Current term", 5, context.getTermInformation().getCurrentTerm());
    assertEquals("Voted For", "member2", context.getTermInformation().getVotedFor());

    sendMessageToSupport(RecoveryCompleted.getInstance(), true);

    verify(mockCohort, never()).applyRecoverySnapshot(anyObject());
    verify(mockCohort, never()).getRestoreFromSnapshot();
    verifyNoMoreInteractions(mockCohort);

    verify(mockPersistentProvider).deleteMessages(10L);
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Override
protected void handleRecover(final Object message) throws Exception {
    if (message instanceof RecoveryCompleted) {
        onRecoveryCompleted();
    } else if (message instanceof SnapshotOffer) {
        applyShardManagerSnapshot((ShardManagerSnapshot)((SnapshotOffer) message).snapshot());
    }
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Override
AbstractClientActorBehavior<?> onReceiveRecover(final Object recover) {
    if (recover instanceof RecoveryCompleted) {
        final ClientIdentifier nextId;
        if (lastId != null) {
            if (!currentFrontend.equals(lastId.getFrontendId())) {
                LOG.error("{}: Mismatched frontend identifier, shutting down. Current: {} Saved: {}",
                    persistenceId(), currentFrontend, lastId.getFrontendId());
                return null;
            }

            nextId = ClientIdentifier.create(currentFrontend, lastId.getGeneration() + 1);
        } else {
            nextId = ClientIdentifier.create(currentFrontend, 0);
        }

        LOG.debug("{}: persisting new identifier {}", persistenceId(), nextId);
        context().saveSnapshot(nextId);
        return new SavingClientActorBehavior(context(), nextId);
    } else if (recover instanceof SnapshotOffer) {
        lastId = (ClientIdentifier) ((SnapshotOffer)recover).snapshot();
        LOG.debug("{}: recovered identifier {}", persistenceId(), lastId);
    } else {
        LOG.warn("{}: ignoring recovery message {}", persistenceId(), recover);
    }

    return this;
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Override
public Receive createReceiveRecover() {
    return ReceiveBuilder.create()
        .match(eventType, evt -> { updateState(evt); })
        .match(SnapshotOffer.class, snapshot -> {
            // Snapshots support is not implemented yet.
        })
        .build();
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
/**
 * Called on restart. Loads from Snapshot first, and then replays Journal Events to update state.
 *
 * @param msg
 */
public void onReceiveRecover(Object msg) {
    log.info("Received Recover: " + msg);
    if (msg instanceof Event) {
        processorState.update((Event) msg);

    } else if (msg instanceof SnapshotOffer) {
        processorState = (ProcessorState) ((SnapshotOffer) msg).snapshot();
    }
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
private void onRecoveredSnapshot(SnapshotOffer offer) {
    log.debug("{}: SnapshotOffer called.", context.getId());

    initRecoveryTimer();

    Snapshot snapshot = (Snapshot) offer.snapshot();

    for (ReplicatedLogEntry entry: snapshot.getUnAppliedEntries()) {
        if (isMigratedPayload(entry)) {
            hasMigratedDataRecovered = true;
        }
    }

    if (!context.getPersistenceProvider().isRecoveryApplicable()) {
        // We may have just transitioned to disabled and have a snapshot containing state data and/or log
        // entries - we don't want to preserve these, only the server config and election term info.

        snapshot = Snapshot.create(
                EmptyState.INSTANCE, Collections.emptyList(), -1, -1, -1, -1,
                snapshot.getElectionTerm(), snapshot.getElectionVotedFor(), snapshot.getServerConfiguration());
    }

    // Create a replicated log with the snapshot information
    // The replicated log can be used later on to retrieve this snapshot
    // when we need to install it on a peer

    context.setReplicatedLog(ReplicatedLogImpl.newInstance(snapshot, context));
    context.setLastApplied(snapshot.getLastAppliedIndex());
    context.setCommitIndex(snapshot.getLastAppliedIndex());
    context.getTermInformation().update(snapshot.getElectionTerm(), snapshot.getElectionVotedFor());

    Stopwatch timer = Stopwatch.createStarted();

    // Apply the snapshot to the actors state
    if (!(snapshot.getState() instanceof EmptyState)) {
        cohort.applyRecoverySnapshot(snapshot.getState());
    }

    if (snapshot.getServerConfiguration() != null) {
        context.updatePeerIds(snapshot.getServerConfiguration());
    }

    timer.stop();
    log.info("Recovery snapshot applied for {} in {}: snapshotIndex={}, snapshotTerm={}, journal-size={}",
            context.getId(), timer.toString(), replicatedLog().getSnapshotIndex(),
            replicatedLog().getSnapshotTerm(), replicatedLog().size());
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Test
public void testRaftActorForwardsToRaftActorRecoverySupport() {
    String persistenceId = factory.generateActorId("leader-");

    DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();

    config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));

    TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
            Collections.<String, String>emptyMap(), config), persistenceId);

    MockRaftActor mockRaftActor = mockActorRef.underlyingActor();

    // Wait for akka's recovery to complete so it doesn't interfere.
    mockRaftActor.waitForRecoveryComplete();

    RaftActorRecoverySupport mockSupport = mock(RaftActorRecoverySupport.class);
    mockRaftActor.setRaftActorRecoverySupport(mockSupport);

    Snapshot snapshot = Snapshot.create(ByteState.of(new byte[]{1}),
            Collections.<ReplicatedLogEntry>emptyList(), 3, 1, 3, 1, -1, null, null);
    SnapshotOffer snapshotOffer = new SnapshotOffer(new SnapshotMetadata("test", 6, 12345), snapshot);
    mockRaftActor.handleRecover(snapshotOffer);

    ReplicatedLogEntry logEntry = new SimpleReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("1", 5));
    mockRaftActor.handleRecover(logEntry);

    ApplyJournalEntries applyJournalEntries = new ApplyJournalEntries(2);
    mockRaftActor.handleRecover(applyJournalEntries);

    DeleteEntries deleteEntries = new DeleteEntries(1);
    mockRaftActor.handleRecover(deleteEntries);

    UpdateElectionTerm updateElectionTerm = new UpdateElectionTerm(5, "member2");
    mockRaftActor.handleRecover(updateElectionTerm);

    verify(mockSupport).handleRecoveryMessage(same(snapshotOffer), any(PersistentDataProvider.class));
    verify(mockSupport).handleRecoveryMessage(same(logEntry), any(PersistentDataProvider.class));
    verify(mockSupport).handleRecoveryMessage(same(applyJournalEntries), any(PersistentDataProvider.class));
    verify(mockSupport).handleRecoveryMessage(same(deleteEntries), any(PersistentDataProvider.class));
    verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm), any(PersistentDataProvider.class));
}
 

import akka.persistence.SnapshotOffer; //导入依赖的package包/类
@Test
public void testOnSnapshotOffer() {

    ReplicatedLog replicatedLog = context.getReplicatedLog();
    replicatedLog.append(new SimpleReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("1")));
    replicatedLog.append(new SimpleReplicatedLogEntry(2, 1, new MockRaftActorContext.MockPayload("2")));
    replicatedLog.append(new SimpleReplicatedLogEntry(3, 1, new MockRaftActorContext.MockPayload("3")));

    ReplicatedLogEntry unAppliedEntry1 = new SimpleReplicatedLogEntry(4, 1,
            new MockRaftActorContext.MockPayload("4", 4));

    ReplicatedLogEntry unAppliedEntry2 = new SimpleReplicatedLogEntry(5, 1,
            new MockRaftActorContext.MockPayload("5", 5));

    long lastAppliedDuringSnapshotCapture = 3;
    long lastIndexDuringSnapshotCapture = 5;
    long electionTerm = 2;
    String electionVotedFor = "member-2";

    MockSnapshotState snapshotState = new MockSnapshotState(Arrays.asList(new MockPayload("1")));
    Snapshot snapshot = Snapshot.create(snapshotState,
            Arrays.asList(unAppliedEntry1, unAppliedEntry2), lastIndexDuringSnapshotCapture, 1,
            lastAppliedDuringSnapshotCapture, 1, electionTerm, electionVotedFor, null);

    SnapshotMetadata metadata = new SnapshotMetadata("test", 6, 12345);
    SnapshotOffer snapshotOffer = new SnapshotOffer(metadata , snapshot);

    sendMessageToSupport(snapshotOffer);

    assertEquals("Journal log size", 2, context.getReplicatedLog().size());
    assertEquals("Journal data size", 9, context.getReplicatedLog().dataSize());
    assertEquals("Last index", lastIndexDuringSnapshotCapture, context.getReplicatedLog().lastIndex());
    assertEquals("Last applied", lastAppliedDuringSnapshotCapture, context.getLastApplied());
    assertEquals("Commit index", lastAppliedDuringSnapshotCapture, context.getCommitIndex());
    assertEquals("Snapshot term", 1, context.getReplicatedLog().getSnapshotTerm());
    assertEquals("Snapshot index", lastAppliedDuringSnapshotCapture, context.getReplicatedLog().getSnapshotIndex());
    assertEquals("Election term", electionTerm, context.getTermInformation().getCurrentTerm());
    assertEquals("Election votedFor", electionVotedFor, context.getTermInformation().getVotedFor());
    assertFalse("Dynamic server configuration", context.isDynamicServerConfigurationInUse());

    verify(mockCohort).applyRecoverySnapshot(snapshotState);
}