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%2FRaftActorLeadershipTransferCohort.java;h=0a22abb9657ff5813ce27a34736e78e5b7e3bc3a;hp=b83bfd370948dc18a3826633568e2a78eeb650ac;hb=dac16f0d464eff3325b3800a803e81b303964e4b;hpb=135129e0cca66040fd512fab740d59b2ab1f8382 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 b83bfd3709..0a22abb965 100644 --- 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 @@ -16,6 +16,7 @@ import com.google.common.base.Stopwatch; import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning; import org.opendaylight.controller.cluster.raft.behaviors.Leader; import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; @@ -26,6 +27,7 @@ import scala.concurrent.duration.FiniteDuration; /** * A raft actor support class that participates in leadership transfer. An instance is created upon * initialization of leadership transfer. + * *

* The transfer process is as follows: *

    @@ -35,13 +37,14 @@ import scala.concurrent.duration.FiniteDuration; * their local RoleChangeNotifiers. *
  1. Call {@link RaftActor#pauseLeader} passing this RaftActorLeadershipTransferCohort * instance. This allows derived classes to perform work prior to transferring leadership.
    2. - *
  2. When the pause is complete, the {@link #run} method is called which in turn calls - * {@link Leader#transferLeadership}.
    3. + *
  3. When the pause is complete, the run method is called which in turn calls + * {@link Leader#transferLeadership(RaftActorLeadershipTransferCohort)}.
    4. *
  4. The Leader calls {@link #transferComplete} on successful completion.
    5. *
  5. Wait a short period of time for the new leader to be elected to give the derived class a chance to * possibly complete work that was suspended while we were transferring.
    6. *
  6. On notification of the new leader from the RaftActor or on time out, notify {@link OnComplete} callbacks.
    7. *
+ * *

* NOTE: All methods on this class must be called on the actor's thread dispatcher as they may access/modify * internal state. @@ -51,17 +54,31 @@ import scala.concurrent.duration.FiniteDuration; public class RaftActorLeadershipTransferCohort { private static final Logger LOG = LoggerFactory.getLogger(RaftActorLeadershipTransferCohort.class); - private final RaftActor raftActor; - private final ActorRef replyTo; - private Cancellable newLeaderTimer; + static final long USE_DEFAULT_LEADER_TIMEOUT = -1; + private final List onCompleteCallbacks = new ArrayList<>(); - private long newLeaderTimeoutInMillis = 2000; private final Stopwatch transferTimer = Stopwatch.createUnstarted(); + private final RaftActor raftActor; + private final String requestedFollowerId; + + private long newLeaderTimeoutInMillis = 2000; + private Cancellable newLeaderTimer; private boolean isTransferring; - RaftActorLeadershipTransferCohort(RaftActor raftActor, ActorRef replyTo) { + RaftActorLeadershipTransferCohort(final RaftActor raftActor) { + this(raftActor, null); + } + + RaftActorLeadershipTransferCohort(final RaftActor raftActor, @Nullable final String requestedFollowerId) { this.raftActor = raftActor; - this.replyTo = replyTo; + this.requestedFollowerId = requestedFollowerId; + + // We'll wait an election timeout period for a new leader to be elected plus some cushion to take into + // account the variance. + final long electionTimeout = raftActor.getRaftActorContext().getConfigParams() + .getElectionTimeOutInterval().toMillis(); + final int variance = raftActor.getRaftActorContext().getConfigParams().getElectionTimeVariance(); + newLeaderTimeoutInMillis = 2 * (electionTimeout + variance); } void init() { @@ -71,29 +88,29 @@ public class RaftActorLeadershipTransferCohort { transferTimer.start(); Optional roleChangeNotifier = raftActor.getRoleChangeNotifier(); - if(roleChangeNotifier.isPresent()) { + if (roleChangeNotifier.isPresent()) { roleChangeNotifier.get().tell(raftActor.newLeaderStateChanged(context.getId(), null, currentBehavior.getLeaderPayloadVersion()), raftActor.self()); } - LeaderTransitioning leaderTransitioning = new LeaderTransitioning(); - for(String peerId: context.getPeerIds()) { + for (String peerId: context.getPeerIds()) { ActorSelection followerActor = context.getPeerActorSelection(peerId); - if(followerActor != null) { - followerActor.tell(leaderTransitioning, context.getActor()); + if (followerActor != null) { + followerActor.tell(new LeaderTransitioning(context.getId()), context.getActor()); } } raftActor.pauseLeader(new TimedRunnable(context.getConfigParams().getElectionTimeOutInterval(), raftActor) { @Override protected void doRun() { + LOG.debug("{}: pauseLeader successfully completed - doing transfer", raftActor.persistenceId()); doTransfer(); } @Override protected void doCancel() { - LOG.debug("{}: pauseLeader timed out - aborting transfer", raftActor.persistenceId()); - abortTransfer(); + LOG.debug("{}: pauseLeader timed out - continuing with transfer", raftActor.persistenceId()); + doTransfer(); } }); } @@ -105,7 +122,7 @@ public class RaftActorLeadershipTransferCohort { void doTransfer() { RaftActorBehavior behavior = raftActor.getCurrentBehavior(); // Sanity check... - if(behavior instanceof Leader) { + if (behavior instanceof Leader) { isTransferring = true; ((Leader)behavior).transferLeadership(this); } else { @@ -134,51 +151,46 @@ public class RaftActorLeadershipTransferCohort { // and convert to follower due to higher term. We should then get an AppendEntries heart // beat with the new leader id. - // Add a timer in case we don't get a leader change - 2 sec should be plenty of time if a new - // leader is elected. Note: the Runnable is sent as a message to the raftActor which executes it - // safely run on the actor's thread dispatcher. + // Add a timer in case we don't get a leader change. Note: the Runnable is sent as a message to the raftActor + // which executes it safely run on the actor's thread dispatcher. FiniteDuration timeout = FiniteDuration.create(newLeaderTimeoutInMillis, TimeUnit.MILLISECONDS); newLeaderTimer = raftActor.getContext().system().scheduler().scheduleOnce(timeout, raftActor.self(), - new Runnable() { - @Override - public void run() { - LOG.debug("{}: leader not elected in time", raftActor.persistenceId()); - finish(true); - } - }, raftActor.getContext().system().dispatcher(), raftActor.self()); + (Runnable) () -> { + LOG.debug("{}: leader not elected in time", raftActor.persistenceId()); + finish(true); + }, raftActor.getContext().system().dispatcher(), raftActor.self()); } - void onNewLeader(String newLeader) { - if(newLeader != null && newLeaderTimer != null) { + void onNewLeader(final String newLeader) { + if (newLeader != null && newLeaderTimer != null) { LOG.debug("{}: leader changed to {}", raftActor.persistenceId(), newLeader); newLeaderTimer.cancel(); finish(true); } } - private void finish(boolean success) { + private void finish(final boolean success) { isTransferring = false; - if(transferTimer.isRunning()) { + if (transferTimer.isRunning()) { transferTimer.stop(); - if(success) { + if (success) { LOG.info("{}: Successfully transferred leadership to {} in {}", raftActor.persistenceId(), - raftActor.getLeaderId(), transferTimer.toString()); + raftActor.getLeaderId(), transferTimer); } else { - LOG.warn("{}: Failed to transfer leadership in {}", raftActor.persistenceId(), - transferTimer.toString()); + LOG.warn("{}: Failed to transfer leadership in {}", raftActor.persistenceId(), transferTimer); } } - for(OnComplete onComplete: onCompleteCallbacks) { - if(success) { - onComplete.onSuccess(raftActor.self(), replyTo); + for (OnComplete onComplete: onCompleteCallbacks) { + if (success) { + onComplete.onSuccess(raftActor.self()); } else { - onComplete.onFailure(raftActor.self(), replyTo); + onComplete.onFailure(raftActor.self()); } } } - void addOnComplete(OnComplete onComplete) { + void addOnComplete(final OnComplete onComplete) { onCompleteCallbacks.add(onComplete); } @@ -186,13 +198,19 @@ public class RaftActorLeadershipTransferCohort { return isTransferring; } - @VisibleForTesting - void setNewLeaderTimeoutInMillis(long newLeaderTimeoutInMillis) { - this.newLeaderTimeoutInMillis = newLeaderTimeoutInMillis; + void setNewLeaderTimeoutInMillis(final long newLeaderTimeoutInMillis) { + if (newLeaderTimeoutInMillis != USE_DEFAULT_LEADER_TIMEOUT) { + this.newLeaderTimeoutInMillis = newLeaderTimeoutInMillis; + } + } + + public Optional getRequestedFollowerId() { + return Optional.fromNullable(requestedFollowerId); } interface OnComplete { - void onSuccess(ActorRef raftActorRef, ActorRef replyTo); - void onFailure(ActorRef raftActorRef, ActorRef replyTo); + void onSuccess(ActorRef raftActorRef); + + void onFailure(ActorRef raftActorRef); } }