() {
@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
applyState(clientActor, identifier, data);
// 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());
context.getSnapshotManager().trimLog(context.getLastApplied(), currentBehavior);
} else if (clientActor != null) {
// Send message for replication
currentBehavior.handleMessage(getSelf(),
new Replicate(clientActor, identifier,
replicatedLogEntry)
);
}
}
}); }
protected String getId() {
return context.getId();
}
/**
* Derived actors can call the isLeader method to check if the current
* RaftActor is the Leader or not
*
* @return true it this RaftActor is a Leader false otherwise
*/
protected boolean isLeader() {
return context.getId().equals(currentBehavior.getLeaderId());
}
/**
* Derived actor can call getLeader if they need a reference to the Leader.
* This would be useful for example in forwarding a request to an actor
* which is the leader
*
* @return A reference to the leader if known, null otherwise
*/
protected ActorSelection getLeader(){
String leaderAddress = getLeaderAddress();
if(leaderAddress == null){
return null;
}
return context.actorSelection(leaderAddress);
}
/**
*
* @return the current leader's id
*/
protected String getLeaderId(){
return currentBehavior.getLeaderId();
}
protected RaftState getRaftState() {
return currentBehavior.state();
}
protected ReplicatedLogEntry getLastLogEntry() {
return replicatedLog.last();
}
protected Long getCurrentTerm(){
return context.getTermInformation().getCurrentTerm();
}
protected Long getCommitIndex(){
return context.getCommitIndex();
}
protected Long getLastApplied(){
return context.getLastApplied();
}
protected RaftActorContext getRaftActorContext() {
return context;
}
protected void updateConfigParams(ConfigParams configParams) {
context.setConfigParams(configParams);
}
/**
* 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){
context.setPeerAddress(peerId, peerAddress);
}
protected void commitSnapshot(long sequenceNumber) {
context.getSnapshotManager().commit(persistence(), sequenceNumber);
}
/**
* The applyState method will be called by the RaftActor when some data
* 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
* by the derived actor. clientActor may be null when
* the RaftActor is behaving as a follower or during
* recovery.
* @param identifier The identifier of the persisted data. This is also
* the same identifier that was passed to persistData by
* the derived actor. identifier may be null when
* the RaftActor is behaving as a follower or during
* recovery
* @param data A piece of data that was persisted by the persistData call.
* This should NEVER be null.
*/
protected abstract void applyState(ActorRef clientActor, String identifier,
Object data);
/**
* This method is called during recovery at the start of a batch of state entries. Derived
* classes should perform any initialization needed to start a batch.
*/
protected abstract void startLogRecoveryBatch(int maxBatchSize);
/**
* This method is called during recovery to append state data to the current batch. This method
* is called 1 or more times after {@link #startLogRecoveryBatch}.
*
* @param data the state data
*/
protected abstract void appendRecoveredLogEntry(Payload data);
/**
* This method is called during recovery to reconstruct the state of the actor.
*
* @param snapshotBytes A snapshot of the state of the actor
*/
protected abstract void applyRecoverySnapshot(byte[] snapshotBytes);
/**
* This method is called during recovery at the end of a batch to apply the current batched
* log entries. This method is called after {@link #appendRecoveredLogEntry}.
*/
protected abstract void applyCurrentLogRecoveryBatch();
/**
* This method is called when recovery is complete.
*/
protected abstract void onRecoveryComplete();
/**
* 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
*/
protected abstract void createSnapshot();
/**
* This method can 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 snapshotBytes A snapshot of the state of the actor
*/
protected abstract void applySnapshot(byte[] snapshotBytes);
/**
* This method will be called by the RaftActor when the state of the
* RaftActor changes. The derived actor can then use methods like
* isLeader or getLeader to do something useful
*/
protected abstract void onStateChanged();
protected abstract DataPersistenceProvider persistence();
/**
* 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();
protected void onLeaderChanged(String oldLeader, String newLeader){};
private void trimPersistentData(long sequenceNumber) {
// Trim akka snapshots
// FIXME : Not sure how exactly the SnapshotSelectionCriteria is applied
// For now guessing that it is ANDed.
persistence().deleteSnapshots(new SnapshotSelectionCriteria(
sequenceNumber - context.getConfigParams().getSnapshotBatchCount(), 43200000));
// Trim akka journal
persistence().deleteMessages(sequenceNumber);
}
private String getLeaderAddress(){
if(isLeader()){
return getSelf().path().toString();
}
String leaderId = currentBehavior.getLeaderId();
if (leaderId == null) {
return null;
}
String peerAddress = context.getPeerAddress(leaderId);
if(LOG.isDebugEnabled()) {
LOG.debug("{}: getLeaderAddress leaderId = {} peerAddress = {}",
persistenceId(), leaderId, peerAddress);
}
return peerAddress;
}
private void handleCaptureSnapshotReply(byte[] snapshotBytes) {
LOG.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length);
context.getSnapshotManager().persist(persistence(), snapshotBytes, currentBehavior, getTotalMemory());
}
protected long getTotalMemory() {
return Runtime.getRuntime().totalMemory();
}
protected boolean hasFollowers(){
return getRaftActorContext().getPeerAddresses().keySet().size() > 0;
}
private class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
private static final int DATA_SIZE_DIVIDER = 5;
private long dataSizeSinceLastSnapshot = 0L;
public ReplicatedLogImpl(Snapshot snapshot) {
super(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();
persistence().persist(new DeleteEntries(adjustedIndex), new Procedure() {
@Override
public void apply(DeleteEntries param)
throws Exception {
//FIXME : Doing nothing for now
dataSize = 0;
for (ReplicatedLogEntry entry : journal) {
dataSize += entry.size();
}
}
});
}
@Override public void appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry) {
appendAndPersist(replicatedLogEntry, null);
}
public void appendAndPersist(
final ReplicatedLogEntry replicatedLogEntry,
final Procedure callback) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Append log entry and persist {} ", persistenceId(), 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.
persistence().persist(replicatedLogEntry,
new Procedure() {
@Override
public void apply(ReplicatedLogEntry evt) throws Exception {
int logEntrySize = replicatedLogEntry.size();
dataSize += logEntrySize;
long dataSizeForCheck = dataSize;
dataSizeSinceLastSnapshot += logEntrySize;
if (!hasFollowers()) {
// When we do not have followers we do not maintain an in-memory log
// due to this the journalSize will never become anything close to the
// snapshot batch count. In fact will mostly be 1.
// Similarly since the journal's dataSize depends on the entries in the
// journal the journal's dataSize will never reach a value close to the
// memory threshold.
// By maintaining the dataSize outside the journal we are tracking essentially
// what we have written to the disk however since we no longer are in
// need of doing a snapshot just for the sake of freeing up memory we adjust
// the real size of data by the DATA_SIZE_DIVIDER so that we do not snapshot as often
// as if we were maintaining a real snapshot
dataSizeForCheck = dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER;
}
long journalSize = replicatedLogEntry.getIndex() + 1;
long dataThreshold = getTotalMemory() *
context.getConfigParams().getSnapshotDataThresholdPercentage() / 100;
if ((journalSize % context.getConfigParams().getSnapshotBatchCount() == 0
|| dataSizeForCheck > dataThreshold)) {
boolean started = context.getSnapshotManager().capture(replicatedLogEntry,
currentBehavior.getReplicatedToAllIndex());
if(started){
dataSizeSinceLastSnapshot = 0;
}
}
if (callback != null){
callback.apply(replicatedLogEntry);
}
}
}
);
}
}
static class DeleteEntries implements Serializable {
private static final long serialVersionUID = 1L;
private final int fromIndex;
public DeleteEntries(int fromIndex) {
this.fromIndex = fromIndex;
}
public int getFromIndex() {
return fromIndex;
}
}
private class ElectionTermImpl implements ElectionTerm {
/**
* Identifier of the actor whose election term information this is
*/
private long currentTerm = 0;
private String votedFor = null;
@Override
public long getCurrentTerm() {
return currentTerm;
}
@Override
public String getVotedFor() {
return votedFor;
}
@Override public void update(long currentTerm, String votedFor) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Set currentTerm={}, votedFor={}", persistenceId(), currentTerm, votedFor);
}
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
persistence().persist(new UpdateElectionTerm(this.currentTerm, this.votedFor), new Procedure(){
@Override public void apply(UpdateElectionTerm param)
throws Exception {
}
});
}
}
static class UpdateElectionTerm implements Serializable {
private static final long serialVersionUID = 1L;
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;
}
}
protected class NonPersistentRaftDataProvider extends NonPersistentDataProvider {
public NonPersistentRaftDataProvider(){
}
/**
* 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.
*
* @param o
*/
@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(COMMIT_SNAPSHOT, self());
}
}
private class CreateSnapshotProcedure implements Procedure {
@Override
public void apply(Void aVoid) throws Exception {
createSnapshot();
}
}
@VisibleForTesting
void setCurrentBehavior(AbstractRaftActorBehavior behavior) {
currentBehavior = behavior;
}
protected RaftActorBehavior getCurrentBehavior() {
return currentBehavior;
}
private static class BehaviorStateHolder {
private RaftActorBehavior behavior;
private String leaderId;
void init(RaftActorBehavior behavior) {
this.behavior = behavior;
this.leaderId = behavior != null ? behavior.getLeaderId() : null;
}
RaftActorBehavior getBehavior() {
return behavior;
}
String getLeaderId() {
return leaderId;
}
}
}