+ } else {
+ recoveryTime = "";
+ }
+
+ if (recoverySnapshotTimer != null) {
+ recoverySnapshotTimer.stop();
+ recoverySnapshotTimer = null;
+ }
+
+ log.info("{}: Recovery completed {} - Switching actor to Follower - last log index = {}, last log term = {}, "
+ + "snapshot index = {}, snapshot term = {}, journal size = {}", context.getId(), recoveryTime,
+ replicatedLog().lastIndex(), replicatedLog().lastTerm(), replicatedLog().getSnapshotIndex(),
+ replicatedLog().getSnapshotTerm(), replicatedLog().size());
+
+ if (dataRecoveredWithPersistenceDisabled
+ || hasMigratedDataRecovered && !context.getPersistenceProvider().isRecoveryApplicable()) {
+ if (hasMigratedDataRecovered) {
+ log.info("{}: Saving snapshot after recovery due to migrated messages", context.getId());
+ } else {
+ log.info("{}: Saving snapshot after recovery due to data persistence disabled", context.getId());
+ }
+
+ // Either data persistence is disabled and we recovered some data entries (ie we must have just
+ // transitioned to disabled or a persistence backup was restored) or we recovered migrated
+ // messages. Either way, we persist a snapshot and delete all the messages from the akka journal
+ // to clean out unwanted messages.
+
+ Snapshot snapshot = Snapshot.create(
+ EmptyState.INSTANCE, Collections.<ReplicatedLogEntry>emptyList(),
+ -1, -1, -1, -1,
+ context.getTermInformation().getCurrentTerm(), context.getTermInformation().getVotedFor(),
+ context.getPeerServerInfo(true));
+
+ persistentProvider.saveSnapshot(snapshot);
+
+ persistentProvider.deleteMessages(persistentProvider.getLastSequenceNumber());
+ } else if (hasMigratedDataRecovered) {
+ log.info("{}: Snapshot capture initiated after recovery due to migrated messages", context.getId());
+
+ context.getSnapshotManager().capture(replicatedLog().last(), -1);
+ } else {
+ possiblyRestoreFromSnapshot();