import akka.persistence.RecoveryCompleted;
import akka.persistence.SnapshotOffer;
import com.google.common.base.Stopwatch;
-import org.opendaylight.controller.cluster.DataPersistenceProvider;
-import org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries;
+import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
import org.slf4j.Logger;
* @author Thomas Pantelis
*/
class RaftActorRecoverySupport {
- private final DataPersistenceProvider persistence;
private final RaftActorContext context;
private final RaftActorBehavior currentBehavior;
private final RaftActorRecoveryCohort cohort;
private Stopwatch recoveryTimer;
private final Logger log;
- RaftActorRecoverySupport(DataPersistenceProvider persistence, RaftActorContext context,
- RaftActorBehavior currentBehavior, RaftActorRecoveryCohort cohort) {
- this.persistence = persistence;
+ RaftActorRecoverySupport(RaftActorContext context, RaftActorBehavior currentBehavior,
+ RaftActorRecoveryCohort cohort) {
this.context = context;
this.currentBehavior = currentBehavior;
this.cohort = cohort;
boolean handleRecoveryMessage(Object message) {
boolean recoveryComplete = false;
- if(persistence.isRecoveryApplicable()) {
+ if(context.getPersistenceProvider().isRecoveryApplicable()) {
if (message instanceof SnapshotOffer) {
onRecoveredSnapshot((SnapshotOffer) message);
} else if (message instanceof ReplicatedLogEntry) {
} else if (message instanceof org.opendaylight.controller.cluster.raft.RaftActor.DeleteEntries) {
// Handle this message for backwards compatibility with pre-Lithium versions.
replicatedLog().removeFrom(((org.opendaylight.controller.cluster.raft.RaftActor.DeleteEntries) message).getFromIndex());
+ } else if (message instanceof org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm) {
+ // Handle this message for backwards compatibility with pre-Lithium versions.
+ org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm update =
+ (org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm)message;
+ context.getTermInformation().update(update.getCurrentTerm(), update.getVotedFor());
} else if (message instanceof UpdateElectionTerm) {
context.getTermInformation().update(((UpdateElectionTerm) message).getCurrentTerm(),
((UpdateElectionTerm) message).getVotedFor());
// 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, persistence, currentBehavior));
+ context.setReplicatedLog(ReplicatedLogImpl.newInstance(snapshot, context, currentBehavior));
context.setLastApplied(snapshot.getLastAppliedIndex());
context.setCommitIndex(snapshot.getLastAppliedIndex());
long lastUnappliedIndex = context.getLastApplied() + 1;
if(log.isDebugEnabled()) {
+ // it can happen that lastUnappliedIndex > toIndex, if the AJE is in the persistent journal
+ // but the entry itself has made it to that state and recovered via the snapshot
log.debug("{}: Received apply journal entries for recovery, applying to state: {} to {}",
context.getId(), lastUnappliedIndex, toIndex);
}