X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActorSnapshotMessageSupport.java;h=2db595d743b727fdde09a66d57f8791c0545ac2f;hb=27966a20f38fa6771b56936137a3a857474a91ef;hp=21c8ffa68e44763a14f0d5865ad03d8104e25d60;hpb=fe45ad923c8cf83d730cf4d576c310967afabdf3;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java index 21c8ffa68e..2db595d743 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java @@ -7,13 +7,10 @@ */ package org.opendaylight.controller.cluster.raft; -import akka.actor.ActorRef; import akka.japi.Procedure; import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; -import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; -import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; import org.slf4j.Logger; @@ -26,28 +23,34 @@ import org.slf4j.Logger; class RaftActorSnapshotMessageSupport { static final String COMMIT_SNAPSHOT = "commit_snapshot"; - private final DataPersistenceProvider persistence; private final RaftActorContext context; private final RaftActorBehavior currentBehavior; private final RaftActorSnapshotCohort cohort; - private final ActorRef raftActorRef; private final Logger log; private final Procedure createSnapshotProcedure = new Procedure() { @Override - public void apply(Void notUsed) throws Exception { - cohort.createSnapshot(raftActorRef); + public void apply(Void notUsed) { + cohort.createSnapshot(context.getActor()); } }; - RaftActorSnapshotMessageSupport(DataPersistenceProvider persistence, RaftActorContext context, - RaftActorBehavior currentBehavior, RaftActorSnapshotCohort cohort, ActorRef raftActorRef) { - this.persistence = persistence; + private final Procedure applySnapshotProcedure = new Procedure() { + @Override + public void apply(byte[] state) { + cohort.applySnapshot(state); + } + }; + + RaftActorSnapshotMessageSupport(RaftActorContext context, RaftActorBehavior currentBehavior, + RaftActorSnapshotCohort cohort) { this.context = context; this.currentBehavior = currentBehavior; this.cohort = cohort; - this.raftActorRef = raftActorRef; this.log = context.getLogger(); + + context.getSnapshotManager().setCreateSnapshotCallable(createSnapshotProcedure); + context.getSnapshotManager().setApplySnapshotProcedure(applySnapshotProcedure); } boolean handleSnapshotMessage(Object message) { @@ -60,14 +63,11 @@ class RaftActorSnapshotMessageSupport { } else if (message instanceof SaveSnapshotFailure) { onSaveSnapshotFailure((SaveSnapshotFailure) message); return true; - } else if (message instanceof CaptureSnapshot) { - onCaptureSnapshot(message); - return true; } else if (message instanceof CaptureSnapshotReply) { onCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot()); return true; } else if (message.equals(COMMIT_SNAPSHOT)) { - context.getSnapshotManager().commit(persistence, -1); + context.getSnapshotManager().commit(-1, currentBehavior); return true; } else { return false; @@ -77,13 +77,7 @@ class RaftActorSnapshotMessageSupport { private void onCaptureSnapshotReply(byte[] snapshotBytes) { log.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", context.getId(), snapshotBytes.length); - context.getSnapshotManager().persist(persistence, snapshotBytes, currentBehavior, context.getTotalMemory()); - } - - private void onCaptureSnapshot(Object message) { - log.debug("{}: CaptureSnapshot received by actor: {}", context.getId(), message); - - context.getSnapshotManager().create(createSnapshotProcedure); + context.getSnapshotManager().persist(snapshotBytes, currentBehavior, context.getTotalMemory()); } private void onSaveSnapshotFailure(SaveSnapshotFailure saveSnapshotFailure) { @@ -98,21 +92,13 @@ class RaftActorSnapshotMessageSupport { long sequenceNumber = success.metadata().sequenceNr(); - context.getSnapshotManager().commit(persistence, sequenceNumber); + context.getSnapshotManager().commit(sequenceNumber, currentBehavior); } private void onApplySnapshot(Snapshot snapshot) { - if(log.isDebugEnabled()) { - log.debug("{}: ApplySnapshot called on Follower Actor " + - "snapshotIndex:{}, snapshotTerm:{}", context.getId(), snapshot.getLastAppliedIndex(), - snapshot.getLastAppliedTerm()); - } - - cohort.applySnapshot(snapshot.getState()); + log.info("{}: Applying snapshot on follower with snapshotIndex: {}, snapshotTerm: {}", context.getId(), + snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm()); - //clears the followers log, sets the snapshot index to ensure adjusted-index works - context.setReplicatedLog(ReplicatedLogImpl.newInstance(snapshot, context, persistence, - currentBehavior)); - context.setLastApplied(snapshot.getLastAppliedIndex()); + context.getSnapshotManager().apply(snapshot); } }