X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActor.java;h=5f6f3ec24e573c4dedba6ba42a38dbbe28a51698;hp=e9e0d630d092eeffa1833441936c26cfeea40d17;hb=f1c3050779d7770ef6a12a67a1870765c3dfd9eb;hpb=766df8b58cc1decd7b40e0adbf41d3657fc21f2c

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java
index e9e0d630d0..5f6f3ec24e 100644
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java
@@ -12,7 +12,6 @@ package org.opendaylight.controller.cluster.raft;
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.japi.Procedure;
-import akka.persistence.SnapshotSelectionCriteria;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
 import com.google.common.base.Optional;
@@ -109,7 +108,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
      */
     private final RaftActorContextImpl context;
 
-    private final DelegatingPersistentDataProvider delegatingPersistenceProvider = new DelegatingPersistentDataProvider(null);
+    private final DelegatingPersistentDataProvider delegatingPersistenceProvider;
+
+    private final PersistentDataProvider persistentProvider;
 
     private RaftActorRecoverySupport raftRecovery;
 
@@ -119,11 +120,16 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
     private final SwitchBehaviorSupplier reusableSwitchBehaviorSupplier = new SwitchBehaviorSupplier();
 
+    private RaftActorServerConfigurationSupport serverConfigurationSupport;
+
     public RaftActor(String id, Map<String, String> peerAddresses,
          Optional<ConfigParams> configParams, short payloadVersion) {
 
+        persistentProvider = new PersistentDataProvider(this);
+        delegatingPersistenceProvider = new RaftActorDelegatingPersistentDataProvider(null, persistentProvider);
+
         context = new RaftActorContextImpl(this.getSelf(),
-            this.getContext(), id, new ElectionTermImpl(delegatingPersistenceProvider, id, LOG),
+            this.getContext(), id, new ElectionTermImpl(persistentProvider, id, LOG),
             -1, -1, peerAddresses,
             (configParams.isPresent() ? configParams.get(): new DefaultConfigParamsImpl()),
             delegatingPersistenceProvider, LOG);
@@ -140,6 +146,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         super.preStart();
 
         snapshotSupport = newRaftActorSnapshotMessageSupport();
+        serverConfigurationSupport = new RaftActorServerConfigurationSupport(getRaftActorContext());
     }
 
     @Override
@@ -161,22 +168,20 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             raftRecovery = newRaftActorRecoverySupport();
         }
 
-        boolean recoveryComplete = raftRecovery.handleRecoveryMessage(message);
+        boolean recoveryComplete = raftRecovery.handleRecoveryMessage(message, persistentProvider);
         if(recoveryComplete) {
-            if(!persistence().isRecoveryApplicable()) {
-                // Delete all the messages from the akka journal so that we do not end up with consistency issues
-                // Note I am not using the dataPersistenceProvider and directly using the akka api here
-                deleteMessages(lastSequenceNr());
-
-                // Delete all the akka snapshots as they will not be needed
-                deleteSnapshots(new SnapshotSelectionCriteria(scala.Long.MaxValue(), scala.Long.MaxValue()));
-            }
-
             onRecoveryComplete();
 
             initializeBehavior();
 
             raftRecovery = null;
+
+            if (context.getReplicatedLog().size() > 0) {
+                self().tell(new InitiateCaptureSnapshot(), self());
+                LOG.info("Snapshot capture initiated after recovery");
+            } else {
+                LOG.info("Snapshot capture NOT initiated after recovery, journal empty");
+            }
         }
     }
 
@@ -196,7 +201,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
     @Override
     public void handleCommand(final Object message) {
-        if (message instanceof ApplyState){
+        if(serverConfigurationSupport.handleMessage(message, this, getSender())) {
+            return;
+        } else if (message instanceof ApplyState){
             ApplyState applyState = (ApplyState) message;
 
             long elapsedTime = (System.nanoTime() - applyState.getStartTime());
@@ -243,7 +250,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             captureSnapshot();
         } else if(message instanceof SwitchBehavior){
             switchBehavior(((SwitchBehavior) message));
-        } else if(!snapshotSupport.handleSnapshotMessage(message)) {
+        } else if(!snapshotSupport.handleSnapshotMessage(message, getSender())) {
             switchBehavior(reusableSwitchBehaviorSupplier.handleMessage(getSender(), message));
         }
     }
@@ -276,6 +283,11 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
     private void onGetOnDemandRaftStats() {
         // Debugging message to retrieve raft stats.
 
+        Map<String, String> peerAddresses = new HashMap<>();
+        for(String peerId: context.getPeerIds()) {
+            peerAddresses.put(peerId, context.getPeerAddress(peerId));
+        }
+
         OnDemandRaftState.Builder builder = OnDemandRaftState.builder()
                 .commitIndex(context.getCommitIndex())
                 .currentTerm(context.getTermInformation().getCurrentTerm())
@@ -291,7 +303,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
                 .snapshotIndex(replicatedLog().getSnapshotIndex())
                 .snapshotTerm(replicatedLog().getSnapshotTerm())
                 .votedFor(context.getTermInformation().getVotedFor())
-                .peerAddresses(new HashMap<>(context.getPeerAddresses()));
+                .peerAddresses(peerAddresses);
 
         ReplicatedLogEntry lastLogEntry = getLastLogEntry();
         if (lastLogEntry != null) {
@@ -323,19 +335,19 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             onStateChanged();
         }
 
-        String oldBehaviorLeaderId = oldBehavior == null ? null : oldBehaviorState.getLeaderId();
+        String lastValidLeaderId = oldBehavior == null ? null : oldBehaviorState.getLastValidLeaderId();
         String oldBehaviorStateName = oldBehavior == null ? null : oldBehavior.state().name();
 
         // it can happen that the state has not changed but the leader has changed.
         Optional<ActorRef> roleChangeNotifier = getRoleChangeNotifier();
-        if(!Objects.equal(oldBehaviorLeaderId, currentBehavior.getLeaderId()) ||
+        if(!Objects.equal(lastValidLeaderId, currentBehavior.getLeaderId()) ||
            oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) {
             if(roleChangeNotifier.isPresent()) {
                 roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId(),
                         currentBehavior.getLeaderPayloadVersion()), getSelf());
             }
 
-            onLeaderChanged(oldBehaviorLeaderId, currentBehavior.getLeaderId());
+            onLeaderChanged(lastValidLeaderId, currentBehavior.getLeaderId());
         }
 
         if (roleChangeNotifier.isPresent() &&
@@ -382,7 +394,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
         replicatedLog().appendAndPersist(replicatedLogEntry, new Procedure<ReplicatedLogEntry>() {
             @Override
-            public void apply(ReplicatedLogEntry replicatedLogEntry) throws Exception {
+            public void apply(ReplicatedLogEntry replicatedLogEntry) {
                 if (!hasFollowers()){
                     // Increment the Commit Index and the Last Applied values
                     raftContext.setCommitIndex(replicatedLogEntry.getIndex());
@@ -483,7 +495,20 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
     }
 
     protected void updateConfigParams(ConfigParams configParams) {
+
+        // obtain the RaftPolicy for oldConfigParams and the updated one.
+        String oldRaftPolicy = context.getConfigParams().
+            getCustomRaftPolicyImplementationClass();
+        String newRaftPolicy = configParams.
+            getCustomRaftPolicyImplementationClass();
+
+        LOG.debug ("RaftPolicy used with prev.config {}, RaftPolicy used with newConfig {}",
+            oldRaftPolicy, newRaftPolicy);
         context.setConfigParams(configParams);
+        if (!Objects.equal(oldRaftPolicy, newRaftPolicy)) {
+            //RaftPolicy is modifed for the Actor. Re-initialize its current behaviour
+            initializeBehavior();
+        }
     }
 
     public final DataPersistenceProvider persistence() {
@@ -670,21 +695,25 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
     private static class BehaviorStateHolder {
         private RaftActorBehavior behavior;
-        private String leaderId;
+        private String lastValidLeaderId;
         private short leaderPayloadVersion;
 
         void init(RaftActorBehavior behavior) {
             this.behavior = behavior;
-            this.leaderId = behavior != null ? behavior.getLeaderId() : null;
             this.leaderPayloadVersion = behavior != null ? behavior.getLeaderPayloadVersion() : -1;
+
+            String behaviorLeaderId = behavior != null ? behavior.getLeaderId() : null;
+            if(behaviorLeaderId != null) {
+                this.lastValidLeaderId = behaviorLeaderId;
+            }
         }
 
         RaftActorBehavior getBehavior() {
             return behavior;
         }
 
-        String getLeaderId() {
-            return leaderId;
+        String getLastValidLeaderId() {
+            return lastValidLeaderId;
         }
 
         short getLeaderPayloadVersion() {