X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActor.java;h=7148986877330cccdbccf4b46856b7b987fb8c26;hb=6ef0b898f2117a4bb3a510c0df7af340f4fc8eca;hp=660a6463f0ecd0ae5ff3f6e2d7af8ff8fef7d323;hpb=8119659681a6814d257314178e759a6ef1b49766;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 660a6463f0..7148986877 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 @@ -6,27 +6,27 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.raft; +import static com.google.common.base.Verify.verify; +import static java.util.Objects.requireNonNull; + import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.PoisonPill; +import akka.actor.Status; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; -import com.google.common.base.Verify; -import com.google.common.collect.Lists; -import java.io.Serializable; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.TimeUnit; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; import org.apache.commons.lang3.time.DurationFormatUtils; +import org.eclipse.jdt.annotation.NonNull; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.DelegatingPersistentDataProvider; import org.opendaylight.controller.cluster.NonPersistentDataProvider; @@ -35,6 +35,7 @@ import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersisten import org.opendaylight.controller.cluster.notifications.LeaderStateChanged; import org.opendaylight.controller.cluster.notifications.RoleChanged; import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; +import org.opendaylight.controller.cluster.raft.base.messages.CheckConsensusReached; import org.opendaylight.controller.cluster.raft.base.messages.InitiateCaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning; import org.opendaylight.controller.cluster.raft.base.messages.Replicate; @@ -49,20 +50,22 @@ import org.opendaylight.controller.cluster.raft.client.messages.FollowerInfo; 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.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; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * RaftActor encapsulates a state machine that needs to be kept synchronized * in a cluster. It implements the RAFT algorithm as described in the paper * * In Search of an Understandable Consensus Algorithm - *
+ * + ** RaftActor has 3 states and each state has a certain behavior associated * with it. A Raft actor can behave as, *
* A RaftActor MUST be a Leader in order to accept requests from clients to * change the state of it's encapsulated state machine. Once a RaftActor becomes * a Leader it is also responsible for ensuring that all followers ultimately * have the same log and therefore the same state machine as itself. - *
- * + * + ** The current behavior of a RaftActor determines how election for leadership * is initiated and how peer RaftActors react to request for votes. - *
- * + * + ** Each RaftActor also needs to know the current election term. It uses this * information for a couple of things. One is to simply figure out who it * voted for in the last election. Another is to figure out if the message * it received to update it's state is stale. - *
- * + * + ** The RaftActor uses akka-persistence to store it's replicated log. * Furthermore through it's behaviors a Raft Actor determines - *
** This is to account for situations where a we know that a peer * exists but we do not know an address up-front. This may also be used in * situations where a known peer starts off in a different location and we * need to change it's address + * *
* Note that if the peerId does not match the list of peers passed to
* this actor during construction an IllegalStateException will be thrown.
- *
- * @param peerId
- * @param peerAddress
*/
- protected void setPeerAddress(String peerId, String peerAddress){
+ protected void setPeerAddress(final String peerId, final String peerAddress) {
context.setPeerAddress(peerId, peerAddress);
}
/**
* The applyState method will be called by the RaftActor when some data
- * needs to be applied to the actor's state
+ * needs to be applied to the actor's state.
*
* @param clientActor A reference to the client who sent this message. This
* is the same reference that was passed to persistData
@@ -742,8 +814,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
/**
* Returns the RaftActorRecoveryCohort to participate in persistence recovery.
*/
- @Nonnull
- protected abstract RaftActorRecoveryCohort getRaftActorRecoveryCohort();
+ protected abstract @NonNull RaftActorRecoveryCohort getRaftActorRecoveryCohort();
/**
* This method is called when recovery is complete.
@@ -751,10 +822,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
protected abstract void onRecoveryComplete();
/**
- * Returns the RaftActorSnapshotCohort to participate in persistence recovery.
+ * Returns the RaftActorSnapshotCohort to participate in snapshot captures.
*/
- @Nonnull
- protected abstract RaftActorSnapshotCohort getRaftActorSnapshotCohort();
+ protected abstract @NonNull RaftActorSnapshotCohort getRaftActorSnapshotCohort();
/**
* This method will be called by the RaftActor when the state of the
@@ -764,32 +834,50 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
protected abstract void onStateChanged();
/**
- * Notifier Actor for this RaftActor to notify when a role change happens
+ * Notifier Actor for this RaftActor to notify when a role change happens.
+ *
* @return ActorRef - ActorRef of the notifier or Optional.absent if none.
*/
protected abstract Optional
* The default implementation immediately runs the operation.
*
* @param operation the operation to run
*/
- protected void pauseLeader(Runnable operation) {
+ protected void pauseLeader(final Runnable operation) {
operation.run();
}
- protected void onLeaderChanged(String oldLeader, String newLeader) {
+ /**
+ * This method is invoked when the actions hooked to the leader becoming paused failed to execute and the leader
+ * should resume normal operations.
+ *
+ *
+ * Note this method can be invoked even before the operation supplied to {@link #pauseLeader(Runnable)} is invoked.
+ */
+ protected void unpauseLeader() {
- };
+ }
+
+ protected void onLeaderChanged(final String oldLeader, final String newLeader) {
+ }
- private String getLeaderAddress(){
- if(isLeader()){
+ private String getLeaderAddress() {
+ if (isLeader()) {
return getSelf().path().toString();
}
String leaderId = getLeaderId();
@@ -797,15 +885,12 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
return null;
}
String peerAddress = context.getPeerAddress(leaderId);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: getLeaderAddress leaderId = {} peerAddress = {}",
- persistenceId(), leaderId, peerAddress);
- }
+ LOG.debug("{}: getLeaderAddress leaderId = {} peerAddress = {}", persistenceId(), leaderId, peerAddress);
return peerAddress;
}
- protected boolean hasFollowers(){
+ protected boolean hasFollowers() {
return getRaftActorContext().hasFollowers();
}
@@ -817,7 +902,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
LOG.debug("Take a snapshot of current state. lastReplicatedLog is {} and replicatedToAllIndex is {}",
replicatedLog().last(), idx);
- snapshotManager.capture(replicatedLog().last(), idx);
+ snapshotManager.captureWithForcedTrim(replicatedLog().last(), idx);
}
}
@@ -829,13 +914,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
if (isLeader()) {
initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
@Override
- public void onSuccess(ActorRef raftActorRef) {
+ public void onSuccess(final ActorRef raftActorRef) {
LOG.debug("{}: leader transfer succeeded after change to non-voting", persistenceId());
ensureFollowerState();
}
@Override
- public void onFailure(ActorRef raftActorRef) {
+ public void onFailure(final ActorRef raftActorRef) {
LOG.debug("{}: leader transfer failed after change to non-voting", persistenceId());
ensureFollowerState();
}
@@ -847,72 +932,21 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
initializeBehavior();
}
}
- });
- }
- }
-
- /**
- * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.persisted.DeleteEntries}
- * whose type for fromIndex is long instead of int. This class was kept for backwards
- * compatibility with Helium.
- */
- // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility.
- @SuppressWarnings("serial")
- @Deprecated
- static class DeleteEntries implements Serializable {
- private final int fromIndex;
-
- public DeleteEntries(int fromIndex) {
- this.fromIndex = fromIndex;
- }
-
- public int getFromIndex() {
- return fromIndex;
- }
-
- private Object readResolve() {
- return org.opendaylight.controller.cluster.raft.persisted.DeleteEntries.createMigrated(fromIndex);
- }
- }
-
- /**
- * @deprecated Deprecated in favor of non-inner class {@link org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm}
- * which has serialVersionUID set. This class was kept for backwards compatibility with Helium.
- */
- // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility.
- @SuppressWarnings("serial")
- @Deprecated
- static class UpdateElectionTerm implements Serializable {
- private final long currentTerm;
- private final String votedFor;
-
- public UpdateElectionTerm(long currentTerm, String votedFor) {
- this.currentTerm = currentTerm;
- this.votedFor = votedFor;
- }
-
- public long getCurrentTerm() {
- return currentTerm;
- }
-
- public String getVotedFor() {
- return votedFor;
- }
-
- private Object readResolve() {
- return org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm.createMigrated(
- currentTerm, votedFor);
+ }, null, RaftActorLeadershipTransferCohort.USE_DEFAULT_LEADER_TIMEOUT);
}
}
/**
* A point-in-time capture of {@link RaftActorBehavior} state critical for transitioning between behaviors.
*/
- private static abstract class BehaviorState implements Immutable {
+ private abstract static class BehaviorState implements Immutable {
@Nullable abstract RaftActorBehavior getBehavior();
+
@Nullable abstract String getLastValidLeaderId();
+
@Nullable abstract String getLastLeaderId();
- @Nullable abstract short getLeaderPayloadVersion();
+
+ abstract short getLeaderPayloadVersion();
}
/**
@@ -928,7 +962,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
final RaftActorBehavior behavior) {
this.lastValidLeaderId = lastValidLeaderId;
this.lastLeaderId = lastLeaderId;
- this.behavior = Preconditions.checkNotNull(behavior);
+ this.behavior = requireNonNull(behavior);
this.leaderPayloadVersion = behavior.getLeaderPayloadVersion();
}
@@ -991,7 +1025,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
BehaviorState capture(final RaftActorBehavior behavior) {
if (behavior == null) {
- Verify.verify(lastValidLeaderId == null, "Null behavior with non-null last leader");
+ verify(lastValidLeaderId == null, "Null behavior with non-null last leader");
return NULL_BEHAVIOR_STATE;
}
@@ -1003,5 +1037,4 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
return new SimpleBehaviorState(lastValidLeaderId, lastLeaderId, behavior);
}
}
-
}