() {
+ @Override
+ public void apply(ReplicatedLogEntry replicatedLogEntry) throws Exception {
+ if (!hasFollowers()){
+ // Increment the Commit Index and the Last Applied values
+ raftContext.setCommitIndex(replicatedLogEntry.getIndex());
+ raftContext.setLastApplied(replicatedLogEntry.getIndex());
+
+ // Apply the state immediately.
+ self().tell(new ApplyState(clientActor, identifier, replicatedLogEntry), self());
+
+ // Send a ApplyJournalEntries message so that we write the fact that we applied
+ // the state to durable storage
+ self().tell(new ApplyJournalEntries(replicatedLogEntry.getIndex()), self());
+
+ } else if (clientActor != null) {
+ context.getReplicatedLog().captureSnapshotIfReady(replicatedLogEntry);
+
+ // Send message for replication
+ currentBehavior.handleMessage(getSelf(),
+ new Replicate(clientActor, identifier, replicatedLogEntry));
+ }
+ }
+ });
+ }
+
+ private ReplicatedLog replicatedLog() {
+ return context.getReplicatedLog();
}
protected String getId() {
return context.getId();
}
+ @VisibleForTesting
+ void setCurrentBehavior(RaftActorBehavior behavior) {
+ currentBehavior.setDelegate(behavior);
+ }
+
+ protected RaftActorBehavior getCurrentBehavior() {
+ return currentBehavior.getDelegate();
+ }
+
/**
* Derived actors can call the isLeader method to check if the current
* RaftActor is the Leader or not
@@ -376,7 +456,7 @@ public abstract class RaftActor extends UntypedPersistentActor {
}
protected ReplicatedLogEntry getLastLogEntry() {
- return replicatedLog.last();
+ return replicatedLog().last();
}
protected Long getCurrentTerm(){
@@ -391,6 +471,49 @@ public abstract class RaftActor extends UntypedPersistentActor {
return context.getLastApplied();
}
+ protected RaftActorContext getRaftActorContext() {
+ return context;
+ }
+
+ protected void updateConfigParams(ConfigParams configParams) {
+ context.setConfigParams(configParams);
+ }
+
+ public final DataPersistenceProvider persistence() {
+ return delegatingPersistenceProvider.getDelegate();
+ }
+
+ public void setPersistence(DataPersistenceProvider provider) {
+ delegatingPersistenceProvider.setDelegate(provider);
+ }
+
+ protected void setPersistence(boolean persistent) {
+ if(persistent) {
+ setPersistence(new PersistentDataProvider(this));
+ } else {
+ setPersistence(new NonPersistentDataProvider() {
+ /**
+ * The way snapshotting works is,
+ *
+ * - RaftActor calls createSnapshot on the Shard
+ *
- Shard sends a CaptureSnapshotReply and RaftActor then calls saveSnapshot
+ *
- When saveSnapshot is invoked on the akka-persistence API it uses the SnapshotStore to save
+ * the snapshot. The SnapshotStore sends SaveSnapshotSuccess or SaveSnapshotFailure. When the
+ * RaftActor gets SaveSnapshot success it commits the snapshot to the in-memory journal. This
+ * commitSnapshot is mimicking what is done in SaveSnapshotSuccess.
+ *
+ */
+ @Override
+ public void saveSnapshot(Object o) {
+ // Make saving Snapshot successful
+ // Committing the snapshot here would end up calling commit in the creating state which would
+ // be a state violation. That's why now we send a message to commit the snapshot.
+ self().tell(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT, self());
+ }
+ });
+ }
+ }
+
/**
* setPeerAddress sets the address of a known peer at a later time.
*
@@ -409,8 +532,6 @@ public abstract class RaftActor extends UntypedPersistentActor {
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
@@ -432,27 +553,21 @@ public abstract class RaftActor extends UntypedPersistentActor {
Object data);
/**
- * This method will be called by the RaftActor when a snapshot needs to be
- * created. The derived actor should respond with its current state.
- *
- * During recovery the state that is returned by the derived actor will
- * be passed back to it by calling the applySnapshot method
- *
- * @return The current state of the actor
+ * Returns the RaftActorRecoveryCohort to participate in persistence recovery.
*/
- protected abstract void createSnapshot();
+ @Nonnull
+ protected abstract RaftActorRecoveryCohort getRaftActorRecoveryCohort();
/**
- * This method will be called by the RaftActor during recovery to
- * reconstruct the state of the actor.
- *
- * This method may also be called at any other point during normal
- * operations when the derived actor is out of sync with it's peers
- * and the only way to bring it in sync is by applying a snapshot
- *
- * @param snapshot A snapshot of the state of the actor
+ * This method is called when recovery is complete.
*/
- protected abstract void applySnapshot(ByteString snapshot);
+ protected abstract void onRecoveryComplete();
+
+ /**
+ * Returns the RaftActorSnapshotCohort to participate in persistence recovery.
+ */
+ @Nonnull
+ protected abstract RaftActorSnapshotCohort getRaftActorSnapshotCohort();
/**
* This method will be called by the RaftActor when the state of the
@@ -461,50 +576,13 @@ public abstract class RaftActor extends UntypedPersistentActor {
*/
protected abstract void onStateChanged();
- protected void onLeaderChanged(String oldLeader, String newLeader){};
-
- private RaftActorBehavior switchBehavior(RaftState state) {
- if (currentBehavior != null) {
- if (currentBehavior.state() == state) {
- return currentBehavior;
- }
- LOG.info("Switching from state " + currentBehavior.state() + " to "
- + state);
-
- try {
- currentBehavior.close();
- } catch (Exception e) {
- LOG.error(e,
- "Failed to close behavior : " + currentBehavior.state());
- }
-
- } else {
- LOG.info("Switching behavior to " + state);
- }
- RaftActorBehavior behavior = null;
- if (state == RaftState.Candidate) {
- behavior = new Candidate(context);
- } else if (state == RaftState.Follower) {
- behavior = new Follower(context);
- } else {
- behavior = new Leader(context);
- }
-
-
-
- return behavior;
- }
-
- private void trimPersistentData(long sequenceNumber) {
- // Trim akka snapshots
- // FIXME : Not sure how exactly the SnapshotSelectionCriteria is applied
- // For now guessing that it is ANDed.
- deleteSnapshots(new SnapshotSelectionCriteria(
- sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
+ /**
+ * 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 getRoleChangeNotifier();
- // Trim akka journal
- deleteMessages(sequenceNumber);
- }
+ protected void onLeaderChanged(String oldLeader, String newLeader){};
private String getLeaderAddress(){
if(isLeader()){
@@ -516,140 +594,39 @@ public abstract class RaftActor extends UntypedPersistentActor {
}
String peerAddress = context.getPeerAddress(leaderId);
if(LOG.isDebugEnabled()) {
- LOG.debug("getLeaderAddress leaderId = " + leaderId + " peerAddress = "
- + peerAddress);
+ LOG.debug("{}: getLeaderAddress leaderId = {} peerAddress = {}",
+ persistenceId(), leaderId, peerAddress);
}
return peerAddress;
}
- private void handleCaptureSnapshotReply(ByteString stateInBytes) {
- // create a snapshot object from the state provided and save it
- // when snapshot is saved async, SaveSnapshotSuccess is raised.
-
- Snapshot sn = Snapshot.create(stateInBytes.toByteArray(),
- context.getReplicatedLog().getFrom(captureSnapshot.getLastAppliedIndex() + 1),
- captureSnapshot.getLastIndex(), captureSnapshot.getLastTerm(),
- captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm());
-
- saveSnapshot(sn);
-
- LOG.info("Persisting of snapshot done:{}", sn.getLogMessage());
-
- //be greedy and remove entries from in-mem journal which are in the snapshot
- // and update snapshotIndex and snapshotTerm without waiting for the success,
-
- context.getReplicatedLog().snapshotPreCommit(stateInBytes,
- captureSnapshot.getLastAppliedIndex(),
- captureSnapshot.getLastAppliedTerm());
-
- LOG.info("Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
- "and term:{}", captureSnapshot.getLastAppliedIndex(),
- captureSnapshot.getLastAppliedTerm());
-
- captureSnapshot = null;
- hasSnapshotCaptureInitiated = false;
+ protected boolean hasFollowers(){
+ return getRaftActorContext().hasFollowers();
}
+ private void captureSnapshot() {
+ SnapshotManager snapshotManager = context.getSnapshotManager();
- private class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
-
- public ReplicatedLogImpl(Snapshot snapshot) {
- super(ByteString.copyFrom(snapshot.getState()),
- snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm(),
- snapshot.getUnAppliedEntries());
- }
-
- public ReplicatedLogImpl() {
- super();
- }
-
- @Override public void removeFromAndPersist(long logEntryIndex) {
- int adjustedIndex = adjustedIndex(logEntryIndex);
-
- if (adjustedIndex < 0) {
- return;
- }
-
- // FIXME: Maybe this should be done after the command is saved
- journal.subList(adjustedIndex , journal.size()).clear();
-
- persist(new DeleteEntries(adjustedIndex), new Procedure(){
-
- @Override public void apply(DeleteEntries param)
- throws Exception {
- //FIXME : Doing nothing for now
- }
- });
- }
-
- @Override public void appendAndPersist(
- final ReplicatedLogEntry replicatedLogEntry) {
- appendAndPersist(null, null, replicatedLogEntry);
- }
+ if(!snapshotManager.isCapturing()) {
+ LOG.debug("Take a snapshot of current state. lastReplicatedLog is {} and replicatedToAllIndex is {}",
+ replicatedLog().last(), currentBehavior.getReplicatedToAllIndex());
- public void appendAndPersist(final ActorRef clientActor,
- final String identifier,
- final ReplicatedLogEntry replicatedLogEntry) {
- context.getLogger().debug(
- "Append log entry and persist {} ", replicatedLogEntry);
- // FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
- journal.add(replicatedLogEntry);
-
- // When persisting events with persist it is guaranteed that the
- // persistent actor will not receive further commands between the
- // persist call and the execution(s) of the associated event
- // handler. This also holds for multiple persist calls in context
- // of a single command.
- persist(replicatedLogEntry,
- new Procedure() {
- public void apply(ReplicatedLogEntry evt) throws Exception {
- // when a snaphsot is being taken, captureSnapshot != null
- if (hasSnapshotCaptureInitiated == false &&
- journal.size() % context.getConfigParams().getSnapshotBatchCount() == 0) {
-
- LOG.info("Initiating Snapshot Capture..");
- long lastAppliedIndex = -1;
- long lastAppliedTerm = -1;
-
- ReplicatedLogEntry lastAppliedEntry = get(context.getLastApplied());
- if (lastAppliedEntry != null) {
- lastAppliedIndex = lastAppliedEntry.getIndex();
- lastAppliedTerm = lastAppliedEntry.getTerm();
- }
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("Snapshot Capture logSize: {}", journal.size());
- LOG.debug("Snapshot Capture lastApplied:{} ",
- context.getLastApplied());
- LOG.debug("Snapshot Capture lastAppliedIndex:{}", lastAppliedIndex);
- LOG.debug("Snapshot Capture lastAppliedTerm:{}", lastAppliedTerm);
- }
-
- // send a CaptureSnapshot to self to make the expensive operation async.
- getSelf().tell(new CaptureSnapshot(
- lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm),
- null);
- hasSnapshotCaptureInitiated = true;
- }
- // Send message for replication
- if (clientActor != null) {
- currentBehavior.handleMessage(getSelf(),
- new Replicate(clientActor, identifier,
- replicatedLogEntry)
- );
- }
- }
- }
- );
+ snapshotManager.capture(replicatedLog().last(), currentBehavior.getReplicatedToAllIndex());
}
-
}
- private static class DeleteEntries implements Serializable {
+ /**
+ * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.base.messages.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;
}
@@ -659,13 +636,21 @@ public abstract class RaftActor extends UntypedPersistentActor {
}
}
+ /**
+ * @deprecated Deprecated in favor of non-inner class {@link org.opendaylight.controller.cluster.raft.base.messages.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;
- private class ElectionTermImpl implements ElectionTerm {
- /**
- * Identifier of the actor whose election term information this is
- */
- private long currentTerm = 0;
- private String votedFor = null;
+ public UpdateElectionTerm(long currentTerm, String votedFor) {
+ this.currentTerm = currentTerm;
+ this.votedFor = votedFor;
+ }
public long getCurrentTerm() {
return currentTerm;
@@ -674,45 +659,52 @@ public abstract class RaftActor extends UntypedPersistentActor {
public String getVotedFor() {
return votedFor;
}
+ }
- @Override public void update(long currentTerm, String votedFor) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Set currentTerm={}, votedFor={}", currentTerm, votedFor);
+ private static class BehaviorStateHolder {
+ private RaftActorBehavior behavior;
+ private String lastValidLeaderId;
+ private short leaderPayloadVersion;
+
+ void init(RaftActorBehavior behavior) {
+ this.behavior = behavior;
+ this.leaderPayloadVersion = behavior != null ? behavior.getLeaderPayloadVersion() : -1;
+
+ String behaviorLeaderId = behavior != null ? behavior.getLeaderId() : null;
+ if(behaviorLeaderId != null) {
+ this.lastValidLeaderId = behaviorLeaderId;
}
- this.currentTerm = currentTerm;
- this.votedFor = votedFor;
}
- @Override
- public void updateAndPersist(long currentTerm, String votedFor){
- update(currentTerm, votedFor);
- // FIXME : Maybe first persist then update the state
- persist(new UpdateElectionTerm(this.currentTerm, this.votedFor), new Procedure(){
+ RaftActorBehavior getBehavior() {
+ return behavior;
+ }
- @Override public void apply(UpdateElectionTerm param)
- throws Exception {
+ String getLastValidLeaderId() {
+ return lastValidLeaderId;
+ }
- }
- });
+ short getLeaderPayloadVersion() {
+ return leaderPayloadVersion;
}
}
- private static class UpdateElectionTerm implements Serializable {
- private final long currentTerm;
- private final String votedFor;
+ private class SwitchBehaviorSupplier implements Supplier {
+ private Object message;
+ private ActorRef sender;
- public UpdateElectionTerm(long currentTerm, String votedFor) {
- this.currentTerm = currentTerm;
- this.votedFor = votedFor;
- }
-
- public long getCurrentTerm() {
- return currentTerm;
+ public SwitchBehaviorSupplier handleMessage(ActorRef sender, Object message){
+ this.sender = sender;
+ this.message = message;
+ return this;
}
- public String getVotedFor() {
- return votedFor;
+ @Override
+ public RaftActorBehavior get() {
+ if(this.message instanceof SwitchBehavior){
+ return ((SwitchBehavior) message).getNewState().createBehavior(getRaftActorContext());
+ }
+ return currentBehavior.handleMessage(sender, message);
}
}
-
}