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%2FRaftActor.java;h=d10cfae21960750fcdf6af5cdbae14656f14ada6;hb=537c44438b1e8854b719903ac2a4a0d5e12d6606;hp=6fd0693db22c77ae75ec68f7110ff0d427d6fcba;hpb=e84f63ee098fff5b02cbce1281ca0d1208f966fa;p=controller.git

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 6fd0693db2..d10cfae219 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
@@ -15,6 +15,8 @@ import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.PoisonPill;
 import akka.actor.Status;
+import akka.persistence.JournalProtocol;
+import akka.persistence.SnapshotProtocol;
 import com.google.common.annotations.VisibleForTesting;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
@@ -51,12 +53,12 @@ import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
 import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
+import org.opendaylight.controller.cluster.raft.messages.Payload;
 import org.opendaylight.controller.cluster.raft.messages.RequestLeadership;
 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
 import org.opendaylight.controller.cluster.raft.persisted.NoopPayload;
 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
-import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
 import org.opendaylight.yangtools.concepts.Identifier;
 import org.opendaylight.yangtools.concepts.Immutable;
 
@@ -101,8 +103,7 @@ import org.opendaylight.yangtools.concepts.Immutable;
  * </ul>
  */
 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 long APPLY_STATE_DELAY_THRESHOLD_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(50);
 
     /**
      * This context should NOT be passed directly to any other actor it is
@@ -131,9 +132,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         persistentProvider = new PersistentDataProvider(this);
         delegatingPersistenceProvider = new RaftActorDelegatingPersistentDataProvider(null, persistentProvider);
 
-        context = new RaftActorContextImpl(this.getSelf(),
-            this.getContext(), id, new ElectionTermImpl(persistentProvider, id, LOG),
-            -1, -1, peerAddresses,
+        context = new RaftActorContextImpl(getSelf(), getContext(), id,
+            new ElectionTermImpl(persistentProvider, id, LOG), -1, -1, peerAddresses,
             configParams.isPresent() ? configParams.get() : new DefaultConfigParamsImpl(),
             delegatingPersistenceProvider, this::handleApplyState, LOG, this::executeInSelf);
 
@@ -227,9 +227,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         if (snapshotSupport.handleSnapshotMessage(message, getSender())) {
             return;
         }
-        if (message instanceof ApplyState) {
-            ApplyState applyState = (ApplyState) message;
-
+        if (message instanceof ApplyState applyState) {
             if (!hasFollowers()) {
                 // for single node, the capture should happen after the apply state
                 // as we delete messages from the persistent journal which have made it to the snapshot
@@ -241,35 +239,38 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             }
 
             possiblyHandleBehaviorMessage(message);
-        } else if (message instanceof ApplyJournalEntries) {
-            ApplyJournalEntries applyEntries = (ApplyJournalEntries) message;
+        } else if (message instanceof ApplyJournalEntries applyEntries) {
             LOG.debug("{}: Persisting ApplyJournalEntries with index={}", persistenceId(), applyEntries.getToIndex());
 
             persistence().persistAsync(applyEntries, NoopProcedure.instance());
-
         } else if (message instanceof FindLeader) {
-            getSender().tell(
-                new FindLeaderReply(getLeaderAddress()),
-                getSelf()
-            );
+            getSender().tell(new FindLeaderReply(getLeaderAddress()), getSelf());
         } else if (message instanceof GetOnDemandRaftState) {
             onGetOnDemandRaftStats();
         } else if (message instanceof InitiateCaptureSnapshot) {
             captureSnapshot();
-        } else if (message instanceof SwitchBehavior) {
-            switchBehavior((SwitchBehavior) message);
-        } else if (message instanceof LeaderTransitioning) {
-            onLeaderTransitioning((LeaderTransitioning)message);
+        } else if (message instanceof SwitchBehavior switchBehavior) {
+            switchBehavior(switchBehavior);
+        } else if (message instanceof LeaderTransitioning leaderTransitioning) {
+            onLeaderTransitioning(leaderTransitioning);
         } else if (message instanceof Shutdown) {
             onShutDown();
-        } else if (message instanceof Runnable) {
-            ((Runnable)message).run();
-        } else if (message instanceof NoopPayload) {
-            persistData(null, null, (NoopPayload) message, false);
-        } else if (message instanceof RequestLeadership) {
-            onRequestLeadership((RequestLeadership) message);
+        } else if (message instanceof Runnable runnable) {
+            runnable.run();
+        } else if (message instanceof NoopPayload noopPayload) {
+            persistData(null, null, noopPayload, false);
+        } else if (message instanceof RequestLeadership requestLeadership) {
+            onRequestLeadership(requestLeadership);
         } else if (!possiblyHandleBehaviorMessage(message)) {
-            handleNonRaftCommand(message);
+            if (message instanceof JournalProtocol.Response response
+                && delegatingPersistenceProvider.handleJournalResponse(response)) {
+                LOG.debug("{}: handled a journal response", persistenceId());
+            } else if (message instanceof SnapshotProtocol.Response response
+                && delegatingPersistenceProvider.handleSnapshotResponse(response)) {
+                LOG.debug("{}: handled a snapshot response", persistenceId());
+            } else {
+                handleNonRaftCommand(message);
+            }
         }
     }
 
@@ -479,11 +480,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
             builder.lastLogTerm(lastLogEntry.getTerm());
         }
 
-        if (getCurrentBehavior() instanceof AbstractLeader) {
-            AbstractLeader leader = (AbstractLeader)getCurrentBehavior();
+        if (getCurrentBehavior() instanceof AbstractLeader leader) {
             Collection<String> followerIds = leader.getFollowerIds();
             List<FollowerInfo> followerInfoList = new ArrayList<>(followerIds.size());
-            for (String id: followerIds) {
+            for (String id : followerIds) {
                 final FollowerLogInformation info = leader.getFollower(id);
                 followerInfoList.add(new FollowerInfo(id, info.getNextIndex(), info.getMatchIndex(),
                         info.isFollowerActive(), DurationFormatUtils.formatDurationHMS(