Do not expose the shutdown initiator in OnComplete

The sender (replyTo) is not used from callbacks or anywhere else, it should
really be just internal to the RaftActor's onShutdown path. Also clarify Shutdown
message.

Change-Id: Ibc64e345e54c101f22ff360afc1d71d9d6a281f7
Signed-off-by: Robert Varga <rovarga@cisco.com>

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 ff464b0c9ca6629a3c754c017e56eb7cfda6bd2f..43a954756ca36bce89f99782a052fd1e4419721f 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
@@ -274,15 +274,15 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         LOG.debug("{}: Initiating leader transfer", persistenceId());
 
         if(leadershipTransferInProgress == null) {
         LOG.debug("{}: Initiating leader transfer", persistenceId());
 
         if(leadershipTransferInProgress == null) {

-            leadershipTransferInProgress = new RaftActorLeadershipTransferCohort(this, getSender());
+            leadershipTransferInProgress = new RaftActorLeadershipTransferCohort(this);

             leadershipTransferInProgress.addOnComplete(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override
             leadershipTransferInProgress.addOnComplete(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override

-                public void onSuccess(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onSuccess(ActorRef raftActorRef) {

                     leadershipTransferInProgress = null;
                 }
 
                 @Override
                     leadershipTransferInProgress = null;
                 }
 
                 @Override

-                public void onFailure(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onFailure(ActorRef raftActorRef) {

                     leadershipTransferInProgress = null;
                 }
             });
                     leadershipTransferInProgress = null;
                 }
             });


@@ -314,13 +314,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         if (context.hasFollowers()) {
             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override
         if (context.hasFollowers()) {
             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override

-                public void onSuccess(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onSuccess(ActorRef raftActorRef) {

                     LOG.debug("{}: leader transfer succeeded - sending PoisonPill", persistenceId());
                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
                 }
 
                 @Override
                     LOG.debug("{}: leader transfer succeeded - sending PoisonPill", persistenceId());
                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
                 }
 
                 @Override

-                public void onFailure(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onFailure(ActorRef raftActorRef) {

                     LOG.debug("{}: leader transfer failed - sending PoisonPill", persistenceId());
                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
                 }
                     LOG.debug("{}: leader transfer failed - sending PoisonPill", persistenceId());
                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
                 }


@@ -786,13 +786,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
         if (isLeader()) {
             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override
         if (isLeader()) {
             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
                 @Override

-                public void onSuccess(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onSuccess(ActorRef raftActorRef) {

                     LOG.debug("{}: leader transfer succeeded after change to non-voting", persistenceId());
                     ensureFollowerState();
                 }
 
                 @Override
                     LOG.debug("{}: leader transfer succeeded after change to non-voting", persistenceId());
                     ensureFollowerState();
                 }
 
                 @Override

-                public void onFailure(ActorRef raftActorRef, ActorRef replyTo) {
+                public void onFailure(ActorRef raftActorRef) {

                     LOG.debug("{}: leader transfer failed after change to non-voting", persistenceId());
                     ensureFollowerState();
                 }
                     LOG.debug("{}: leader transfer failed after change to non-voting", persistenceId());
                     ensureFollowerState();
                 }

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohort.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohort.java
index d5dfcf3943387149a179a731a6e51e04819c23cc..2916867ffdecd012033a12cd17cd1d5cb7ad9271 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohort.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohort.java
@@ -52,16 +52,14 @@ public class RaftActorLeadershipTransferCohort {
     private static final Logger LOG = LoggerFactory.getLogger(RaftActorLeadershipTransferCohort.class);
 
     private final RaftActor raftActor;
     private static final Logger LOG = LoggerFactory.getLogger(RaftActorLeadershipTransferCohort.class);
 
     private final RaftActor raftActor;

-    private final ActorRef replyTo;

     private Cancellable newLeaderTimer;
     private final List<OnComplete> onCompleteCallbacks = new ArrayList<>();
     private long newLeaderTimeoutInMillis = 2000;
     private final Stopwatch transferTimer = Stopwatch.createUnstarted();
     private boolean isTransferring;
 
     private Cancellable newLeaderTimer;
     private final List<OnComplete> onCompleteCallbacks = new ArrayList<>();
     private long newLeaderTimeoutInMillis = 2000;
     private final Stopwatch transferTimer = Stopwatch.createUnstarted();
     private boolean isTransferring;
 

-    RaftActorLeadershipTransferCohort(RaftActor raftActor, ActorRef replyTo) {
+    RaftActorLeadershipTransferCohort(RaftActor raftActor) {

         this.raftActor = raftActor;
         this.raftActor = raftActor;

-        this.replyTo = replyTo;

     }
 
     void init() {
     }
 
     void init() {


@@ -170,9 +168,9 @@ public class RaftActorLeadershipTransferCohort {
 
         for(OnComplete onComplete: onCompleteCallbacks) {
             if(success) {
 
         for(OnComplete onComplete: onCompleteCallbacks) {
             if(success) {

-                onComplete.onSuccess(raftActor.self(), replyTo);
+                onComplete.onSuccess(raftActor.self());

             } else {
             } else {

-                onComplete.onFailure(raftActor.self(), replyTo);
+                onComplete.onFailure(raftActor.self());

             }
         }
     }
             }
         }
     }


@@ -191,7 +189,7 @@ public class RaftActorLeadershipTransferCohort {
     }
 
     interface OnComplete {
     }
 
     interface OnComplete {

-        void onSuccess(ActorRef raftActorRef, ActorRef replyTo);
-        void onFailure(ActorRef raftActorRef, ActorRef replyTo);
+        void onSuccess(ActorRef raftActorRef);
+        void onFailure(ActorRef raftActorRef);

     }
 }
     }
 }

diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/client/messages/Shutdown.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/client/messages/Shutdown.java
index 754bf30c23e831d1e9861ba77523a5faa30fe2a6..5b76685cb731c5cf9e3b9145d0585d5e67184efb 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/client/messages/Shutdown.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/client/messages/Shutdown.java
@@ -11,7 +11,8 @@ import java.io.Serializable;
 
 /**
  * Message sent to a raft actor to shutdown gracefully. If it's the leader it will transfer leadership to a
 
 /**
  * Message sent to a raft actor to shutdown gracefully. If it's the leader it will transfer leadership to a

- * follower. As its last act, the actor self-destructs via a PoisonPill.
+ * follower. As its last act, the actor self-destructs via a PoisonPill. This message should only be used with
+ * Patterns.gracefulStop().

  *
  * @author Thomas Pantelis
  */
  *
  * @author Thomas Pantelis
  */

diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohortTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohortTest.java
index 45c797cbde7fafeb9a75b5f02d4ec0c9ec91af1f..754e3e436be62e6d32fc2af72a2cc5dd96d4491d 100644 (file)
--- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohortTest.java
+++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohortTest.java
@@ -41,7 +41,7 @@ public class RaftActorLeadershipTransferCohortTest extends AbstractActorTest {
         mockRaftActor = factory.<MockRaftActor>createTestActor(MockRaftActor.builder().id(persistenceId).config(
                 config).pauseLeaderFunction(pauseLeaderFunction).props().withDispatcher(Dispatchers.DefaultDispatcherId()),
                 persistenceId).underlyingActor();
         mockRaftActor = factory.<MockRaftActor>createTestActor(MockRaftActor.builder().id(persistenceId).config(
                 config).pauseLeaderFunction(pauseLeaderFunction).props().withDispatcher(Dispatchers.DefaultDispatcherId()),
                 persistenceId).underlyingActor();

-        cohort = new RaftActorLeadershipTransferCohort(mockRaftActor, null);
+        cohort = new RaftActorLeadershipTransferCohort(mockRaftActor);

         cohort.addOnComplete(onComplete);
         mockRaftActor.waitForInitializeBehaviorComplete();
     }
         cohort.addOnComplete(onComplete);
         mockRaftActor.waitForInitializeBehaviorComplete();
     }


@@ -52,15 +52,15 @@ public class RaftActorLeadershipTransferCohortTest extends AbstractActorTest {
         cohort.setNewLeaderTimeoutInMillis(20000);
 
         cohort.onNewLeader("new-leader");
         cohort.setNewLeaderTimeoutInMillis(20000);
 
         cohort.onNewLeader("new-leader");

-        verify(onComplete, never()).onSuccess(mockRaftActor.self(), null);
+        verify(onComplete, never()).onSuccess(mockRaftActor.self());

 
         cohort.transferComplete();
 
         cohort.onNewLeader(null);
 
         cohort.transferComplete();
 
         cohort.onNewLeader(null);

-        verify(onComplete, never()).onSuccess(mockRaftActor.self(), null);
+        verify(onComplete, never()).onSuccess(mockRaftActor.self());

 
         cohort.onNewLeader("new-leader");
 
         cohort.onNewLeader("new-leader");

-        verify(onComplete).onSuccess(mockRaftActor.self(), null);
+        verify(onComplete).onSuccess(mockRaftActor.self());

     }
 
     @Test
     }
 
     @Test


@@ -68,7 +68,7 @@ public class RaftActorLeadershipTransferCohortTest extends AbstractActorTest {
         setup("testNewLeaderTimeout");
         cohort.setNewLeaderTimeoutInMillis(200);
         cohort.transferComplete();
         setup("testNewLeaderTimeout");
         cohort.setNewLeaderTimeoutInMillis(200);
         cohort.transferComplete();

-        verify(onComplete, timeout(3000)).onSuccess(mockRaftActor.self(), null);
+        verify(onComplete, timeout(3000)).onSuccess(mockRaftActor.self());

     }
 
     @Test
     }
 
     @Test


@@ -76,14 +76,14 @@ public class RaftActorLeadershipTransferCohortTest extends AbstractActorTest {
         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
         setup("testNotLeaderOnDoTransfer");
         cohort.doTransfer();
         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
         setup("testNotLeaderOnDoTransfer");
         cohort.doTransfer();

-        verify(onComplete).onSuccess(mockRaftActor.self(), null);
+        verify(onComplete).onSuccess(mockRaftActor.self());

     }
 
     @Test
     public void testAbortTransfer() {
         setup("testAbortTransfer");
         cohort.abortTransfer();
     }
 
     @Test
     public void testAbortTransfer() {
         setup("testAbortTransfer");
         cohort.abortTransfer();

-        verify(onComplete).onFailure(mockRaftActor.self(), null);
+        verify(onComplete).onFailure(mockRaftActor.self());

     }
 
     @Test
     }
 
     @Test


@@ -97,6 +97,6 @@ public class RaftActorLeadershipTransferCohortTest extends AbstractActorTest {
 
         setup("testPauseLeaderTimeout");
         cohort.init();
 
         setup("testPauseLeaderTimeout");
         cohort.init();

-        verify(onComplete, timeout(2000)).onFailure(mockRaftActor.self(), null);
+        verify(onComplete, timeout(2000)).onFailure(mockRaftActor.self());

     }
 }
     }
 }