roleChangeNotifier = getRoleChangeNotifier();
- if(!Objects.equals(lastLeaderId, currentBehavior.getLeaderId()) ||
- oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) {
- if(roleChangeNotifier.isPresent()) {
+ if (!Objects.equals(lastLeaderId, currentBehavior.getLeaderId())
+ || oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) {
+ if (roleChangeNotifier.isPresent()) {
roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId(),
currentBehavior.getLeaderPayloadVersion()), getSelf());
}
onLeaderChanged(lastValidLeaderId, currentBehavior.getLeaderId());
- if(leadershipTransferInProgress != null) {
+ if (leadershipTransferInProgress != null) {
leadershipTransferInProgress.onNewLeader(currentBehavior.getLeaderId());
}
serverConfigurationSupport.onNewLeader(currentBehavior.getLeaderId());
}
- if (roleChangeNotifier.isPresent() &&
- (oldBehavior == null || oldBehavior.state() != currentBehavior.state())) {
+ if (roleChangeNotifier.isPresent()
+ && (oldBehavior == null || oldBehavior.state() != currentBehavior.state())) {
roleChangeNotifier.get().tell(new RoleChanged(getId(), oldBehaviorStateName ,
currentBehavior.state().name()), getSelf());
}
@@ -508,53 +508,64 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
public long snapshotSequenceNr() {
// When we do a snapshot capture, we also capture and save the sequence-number of the persistent journal,
// so that we can delete the persistent journal based on the saved sequence-number
- // However , when akka replays the journal during recovery, it replays it from the sequence number when the snapshot
- // was saved and not the number we saved.
- // We would want to override it , by asking akka to use the last-sequence number known to us.
+ // However , when akka replays the journal during recovery, it replays it from the sequence number when the
+ // snapshot was saved and not the number we saved. We would want to override it , by asking akka to use the
+ // last-sequence number known to us.
return context.getSnapshotManager().getLastSequenceNumber();
}
/**
- * When a derived RaftActor needs to persist something it must call
- * persistData.
+ * Persists the given Payload in the journal and replicates to any followers. After successful completion,
+ * {@link #applyState(ActorRef, Identifier, Object)} is notified.
*
- * @param clientActor
- * @param identifier
- * @param data
+ * @param clientActor optional ActorRef that is provided via the applyState callback
+ * @param identifier the payload identifier
+ * @param data the payload data to persist
+ * @param batchHint if true, an attempt is made to delay immediate replication and batch the payload with
+ * subsequent payloads for efficiency. Otherwise the payload is immediately replicated.
*/
- protected final void persistData(final ActorRef clientActor, final Identifier identifier, final Payload data) {
-
- ReplicatedLogEntry replicatedLogEntry = new ReplicatedLogImplEntry(
+ protected final void persistData(final ActorRef clientActor, final Identifier identifier, final Payload data,
+ final boolean batchHint) {
+ ReplicatedLogEntry replicatedLogEntry = new SimpleReplicatedLogEntry(
context.getReplicatedLog().lastIndex() + 1,
context.getTermInformation().getCurrentTerm(), data);
+ replicatedLogEntry.setPersistencePending(true);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Persist data {}", persistenceId(), replicatedLogEntry);
- }
+ LOG.debug("{}: Persist data {}", persistenceId(), replicatedLogEntry);
final RaftActorContext raftContext = getRaftActorContext();
- replicatedLog().appendAndPersist(replicatedLogEntry, replicatedLogEntry1 -> {
- if (!hasFollowers()){
+ boolean wasAppended = replicatedLog().appendAndPersist(replicatedLogEntry, persistedLogEntry -> {
+ // Clear the persistence pending flag in the log entry.
+ persistedLogEntry.setPersistencePending(false);
+
+ if (!hasFollowers()) {
// Increment the Commit Index and the Last Applied values
- raftContext.setCommitIndex(replicatedLogEntry1.getIndex());
- raftContext.setLastApplied(replicatedLogEntry1.getIndex());
+ raftContext.setCommitIndex(persistedLogEntry.getIndex());
+ raftContext.setLastApplied(persistedLogEntry.getIndex());
// Apply the state immediately.
- self().tell(new ApplyState(clientActor, identifier, replicatedLogEntry1), self());
+ self().tell(new ApplyState(clientActor, identifier, persistedLogEntry), self());
// Send a ApplyJournalEntries message so that we write the fact that we applied
// the state to durable storage
- self().tell(new ApplyJournalEntries(replicatedLogEntry1.getIndex()), self());
+ self().tell(new ApplyJournalEntries(persistedLogEntry.getIndex()), self());
} else {
- context.getReplicatedLog().captureSnapshotIfReady(replicatedLogEntry1);
+ context.getReplicatedLog().captureSnapshotIfReady(replicatedLogEntry);
- // Send message for replication
- getCurrentBehavior().handleMessage(getSelf(),
- new Replicate(clientActor, identifier, replicatedLogEntry1));
+ // Local persistence is complete so send the CheckConsensusReached message to the behavior (which
+ // normally should still be the leader) to check if consensus has now been reached in conjunction with
+ // follower replication.
+ getCurrentBehavior().handleMessage(getSelf(), CheckConsensusReached.INSTANCE);
}
- });
+ }, true);
+
+ if (wasAppended && hasFollowers()) {
+ // Send log entry for replication.
+ getCurrentBehavior().handleMessage(getSelf(), new Replicate(clientActor, identifier, replicatedLogEntry,
+ !batchHint));
+ }
}
private ReplicatedLog replicatedLog() {
@@ -576,7 +587,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
/**
* Derived actors can call the isLeader method to check if the current
- * RaftActor is the Leader or not
+ * RaftActor is the Leader or not.
*
* @return true it this RaftActor is a Leader false otherwise
*/
@@ -585,8 +596,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
}
protected final boolean isLeaderActive() {
- return getRaftState() != RaftState.IsolatedLeader && getRaftState() != RaftState.PreLeader &&
- !shuttingDown && !isLeadershipTransferInProgress();
+ return getRaftState() != RaftState.IsolatedLeader && getRaftState() != RaftState.PreLeader
+ && !shuttingDown && !isLeadershipTransferInProgress();
}
private boolean isLeadershipTransferInProgress() {
@@ -600,10 +611,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
*
* @return A reference to the leader if known, null otherwise
*/
- public ActorSelection getLeader(){
+ public ActorSelection getLeader() {
String leaderAddress = getLeaderAddress();
- if(leaderAddress == null){
+ if (leaderAddress == null) {
return null;
}
@@ -611,10 +622,11 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
}
/**
+ * Returns the id of the current leader.
*
* @return the current leader's id
*/
- protected final String getLeaderId(){
+ protected final String getLeaderId() {
return getCurrentBehavior().getLeaderId();
}
@@ -623,7 +635,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
return getCurrentBehavior().state();
}
- protected Long getCurrentTerm(){
+ protected Long getCurrentTerm() {
return context.getTermInformation().getCurrentTerm();
}
@@ -634,10 +646,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
protected void updateConfigParams(ConfigParams configParams) {
// obtain the RaftPolicy for oldConfigParams and the updated one.
- String oldRaftPolicy = context.getConfigParams().
- getCustomRaftPolicyImplementationClass();
- String newRaftPolicy = configParams.
- getCustomRaftPolicyImplementationClass();
+ String oldRaftPolicy = context.getConfigParams().getCustomRaftPolicyImplementationClass();
+ String newRaftPolicy = configParams.getCustomRaftPolicyImplementationClass();
LOG.debug("{}: RaftPolicy used with prev.config {}, RaftPolicy used with newConfig {}", persistenceId(),
oldRaftPolicy, newRaftPolicy);
@@ -651,7 +661,8 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
String previousLeaderId = behavior.getLeaderId();
short previousLeaderPayloadVersion = behavior.getLeaderPayloadVersion();
- LOG.debug("{}: Re-initializing to Follower with previous leaderId {}", persistenceId(), previousLeaderId);
+ LOG.debug("{}: Re-initializing to Follower with previous leaderId {}", persistenceId(),
+ previousLeaderId);
changeCurrentBehavior(new Follower(context, previousLeaderId, previousLeaderPayloadVersion));
} else {
@@ -670,14 +681,14 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
protected void setPersistence(boolean persistent) {
DataPersistenceProvider currentPersistence = persistence();
- if(persistent && (currentPersistence == null || !currentPersistence.isRecoveryApplicable())) {
+ if (persistent && (currentPersistence == null || !currentPersistence.isRecoveryApplicable())) {
setPersistence(new PersistentDataProvider(this));
- if(getCurrentBehavior() != null) {
+ if (getCurrentBehavior() != null) {
LOG.info("{}: Persistence has been enabled - capturing snapshot", persistenceId());
captureSnapshot();
}
- } else if(!persistent && (currentPersistence == null || currentPersistence.isRecoveryApplicable())) {
+ } else if (!persistent && (currentPersistence == null || currentPersistence.isRecoveryApplicable())) {
setPersistence(new NonPersistentDataProvider() {
/**
* The way snapshotting works is,
@@ -691,7 +702,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
*
*/
@Override
- public void saveSnapshot(Object o) {
+ public void saveSnapshot(Object object) {
// 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.
@@ -703,25 +714,24 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
/**
* setPeerAddress sets the address of a known peer at a later time.
+ *
*
* 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(String peerId, 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
@@ -763,7 +773,8 @@ 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 getRoleChangeNotifier();
@@ -774,6 +785,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
* work prior to performing the operation. On completion of any work, the run method must be called on the
* given Runnable to proceed with the given operation. Important: the run method must be called on
* this actor's thread dispatcher as as it modifies internal state.
+ *
*
* The default implementation immediately runs the operation.
*
@@ -784,11 +796,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
}
protected void onLeaderChanged(String oldLeader, String newLeader) {
-
}
- private String getLeaderAddress(){
- if(isLeader()){
+ private String getLeaderAddress() {
+ if (isLeader()) {
return getSelf().path().toString();
}
String leaderId = getLeaderId();
@@ -796,15 +807,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();
}
@@ -853,10 +861,13 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor {
/**
* 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();
}