+
+ 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(ReplicatedLogEntry repLogEntry) {
+ return repLogEntry.getData() instanceof ServerConfigurationPayload;
+ }
+
+ private static boolean isPersistentPayload(ReplicatedLogEntry repLogEntry) {
+ return repLogEntry.getData() instanceof PersistentPayload;
+ }
+
+ private static boolean isMigratedPayload(ReplicatedLogEntry repLogEntry) {
+ return isMigratedSerializable(repLogEntry.getData());
+ }
+
+ private static boolean isMigratedSerializable(Object message) {
+ return message instanceof MigratedSerializable && ((MigratedSerializable)message).isMigrated();