- log.info("Recovery completed" + recoveryTime + " - Switching actor to Follower - " +
- "Persistence Id = " + context.getId() +
- " Last index in log = {}, snapshotIndex = {}, snapshotTerm = {}, " +
- "journal-size = {}", replicatedLog().lastIndex(), replicatedLog().getSnapshotIndex(),
- replicatedLog().getSnapshotTerm(), replicatedLog().size());
+ 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();
+ }
+ }
+
+ private static boolean isServerConfigurationPayload(final ReplicatedLogEntry repLogEntry) {
+ return repLogEntry.getData() instanceof ServerConfigurationPayload;
+ }
+
+ private static boolean isPersistentPayload(final ReplicatedLogEntry repLogEntry) {
+ return repLogEntry.getData() instanceof PersistentPayload;
+ }
+
+ private static boolean isMigratedPayload(final ReplicatedLogEntry repLogEntry) {
+ return isMigratedSerializable(repLogEntry.getData());