Merge changes Idfb4fbea,Ief82b050

[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / RaftActor.java

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 cb1b42aa1b1fb95ced067fc152e0760a3ce0576f..77ff47d0ad6ac5e53d1b375be4e4fbd28e0a8859 100644 (file)
--- 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
@@ -87,6 +87,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
     private static final long APPLY_STATE_DELAY_THRESHOLD_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(50L); // 50 millis
 
+    private static final Procedure<ApplyJournalEntries> APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK =
+            new Procedure<ApplyJournalEntries>() {
+                @Override
+                public void apply(ApplyJournalEntries param) throws Exception {
+                }
+            };
+
     protected final Logger LOG = LoggerFactory.getLogger(getClass());
 
     /**
@@ -99,7 +106,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
      * This context should NOT be passed directly to any other actor it is
      * only to be consumed by the RaftActorBehaviors
      */
-    private final RaftActorContext context;
+    private final RaftActorContextImpl context;
 
     /**
      * The in-memory journal
@@ -140,6 +147,19 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         super.preStart();
     }
 
+    @Override
+    public void postStop() {
+        if(currentBehavior != null) {
+            try {
+                currentBehavior.close();
+            } catch (Exception e) {
+                LOG.debug("{}: Error closing behavior {}", persistenceId(), currentBehavior.state());
+            }
+        }
+
+        super.postStop();
+    }
+
     @Override
     public void handleRecover(Object message) {
         if(persistence().isRecoveryApplicable()) {
@@ -306,11 +326,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             if(LOG.isDebugEnabled()) {
                 LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), applyEntries.getToIndex());
             }
-            persistence().persist(applyEntries, new Procedure<ApplyJournalEntries>() {
-                @Override
-                public void apply(ApplyJournalEntries param) throws Exception {
-                }
-            });
+
+            persistence().persist(applyEntries, APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK);
 
         } else if(message instanceof ApplySnapshot ) {
             Snapshot snapshot = ((ApplySnapshot) message).getSnapshot();
@@ -358,7 +375,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
                 context.getReplicatedLog().size());
 
         } else if (message instanceof CaptureSnapshot) {
-            LOG.info("{}: CaptureSnapshot received by actor", persistenceId());
+            LOG.debug("{}: CaptureSnapshot received by actor: {}", persistenceId(), message);
 
             if(captureSnapshot == null) {
                 captureSnapshot = (CaptureSnapshot)message;
@@ -515,6 +532,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         return context;
     }
 
+    protected void updateConfigParams(ConfigParams configParams) {
+        context.setConfigParams(configParams);
+    }
+
     /**
      * setPeerAddress sets the address of a known peer at a later time.
      * <p>
@@ -658,7 +679,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
     }
 
     private void handleCaptureSnapshotReply(byte[] snapshotBytes) {
-        LOG.info("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length);
+        LOG.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length);
 
         // create a snapshot object from the state provided and save it
         // when snapshot is saved async, SaveSnapshotSuccess is raised.
@@ -672,16 +693,27 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
 
         LOG.info("{}: Persisting of snapshot done:{}", persistenceId(), sn.getLogMessage());
 
-        long dataThreshold = Runtime.getRuntime().totalMemory() *
+        long dataThreshold = getTotalMemory() *
                 getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100;
         if (context.getReplicatedLog().dataSize() > dataThreshold) {
+
+            if(LOG.isDebugEnabled()) {
+                LOG.debug("{}: dataSize {} exceeds dataThreshold {} - doing snapshotPreCommit with index {}",
+                        persistenceId(), context.getReplicatedLog().dataSize(), dataThreshold,
+                        captureSnapshot.getLastAppliedIndex());
+            }
+
             // if memory is less, clear the log based on lastApplied.
             // this could/should only happen if one of the followers is down
             // as normally we keep removing from the log when its replicated to all.
             context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(),
                     captureSnapshot.getLastAppliedTerm());
 
-            getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
+            // Don't reset replicatedToAllIndex to -1 as this may prevent us from trimming the log after an
+            // install snapshot to a follower.
+            if(captureSnapshot.getReplicatedToAllIndex() >= 0) {
+                getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
+            }
         } else if(captureSnapshot.getReplicatedToAllIndex() != -1){
             // clear the log based on replicatedToAllIndex
             context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(),
@@ -698,9 +730,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         }
 
 
-        LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
-            "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(),
-            captureSnapshot.getLastAppliedTerm());
+        LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex: {} " +
+            "and term: {}", persistenceId(), replicatedLog.getSnapshotIndex(),
+            replicatedLog.getSnapshotTerm());
 
         if (isLeader() && captureSnapshot.isInstallSnapshotInitiated()) {
             // this would be call straight to the leader and won't initiate in serialization
@@ -712,6 +744,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         context.setSnapshotCaptureInitiated(false);
     }
 
+    protected long getTotalMemory() {
+        return Runtime.getRuntime().totalMemory();
+    }
+
     protected boolean hasFollowers(){
         return getRaftActorContext().getPeerAddresses().keySet().size() > 0;
     }
@@ -802,7 +838,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
                             dataSizeForCheck = dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER;
                         }
 
-                        long dataThreshold = Runtime.getRuntime().totalMemory() *
+                        long dataThreshold = getTotalMemory() *
                                 getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100;
 
                         // when a snaphsot is being taken, captureSnapshot != null