import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActor;
import org.opendaylight.controller.cluster.notifications.RoleChanged;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
private static final long APPLY_STATE_DELAY_THRESHOLD_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(50L); // 50 millis
+ private static final Procedure<ApplyJournalEntries> APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK =
+ new Procedure<ApplyJournalEntries>() {
+ @Override
+ public void apply(ApplyJournalEntries param) throws Exception {
+ }
+ };
+
protected final Logger LOG = LoggerFactory.getLogger(getClass());
/**
* This context should NOT be passed directly to any other actor it is
* only to be consumed by the RaftActorBehaviors
*/
- private final RaftActorContext context;
+ private final RaftActorContextImpl context;
/**
* The in-memory journal
super.preStart();
}
+ @Override
+ public void postStop() {
+ if(currentBehavior != null) {
+ try {
+ currentBehavior.close();
+ } catch (Exception e) {
+ LOG.debug("{}: Error closing behavior {}", persistenceId(), currentBehavior.state());
+ }
+ }
+
+ super.postStop();
+ }
+
@Override
public void handleRecover(Object message) {
if(persistence().isRecoveryApplicable()) {
} else if (message instanceof ReplicatedLogEntry) {
onRecoveredJournalLogEntry((ReplicatedLogEntry) message);
} else if (message instanceof ApplyLogEntries) {
- onRecoveredApplyLogEntries((ApplyLogEntries) message);
+ // Handle this message for backwards compatibility with pre-Lithium versions.
+ onRecoveredApplyLogEntries(((ApplyLogEntries) message).getToIndex());
+ } else if (message instanceof ApplyJournalEntries) {
+ onRecoveredApplyLogEntries(((ApplyJournalEntries) message).getToIndex());
} else if (message instanceof DeleteEntries) {
replicatedLog.removeFrom(((DeleteEntries) message).getFromIndex());
} else if (message instanceof UpdateElectionTerm) {
replicatedLog.append(logEntry);
}
- private void onRecoveredApplyLogEntries(ApplyLogEntries ale) {
+ private void onRecoveredApplyLogEntries(long toIndex) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Received ApplyLogEntries for recovery, applying to state: {} to {}",
- persistenceId(), context.getLastApplied() + 1, ale.getToIndex());
+ persistenceId(), context.getLastApplied() + 1, toIndex);
}
- for (long i = context.getLastApplied() + 1; i <= ale.getToIndex(); i++) {
+ for (long i = context.getLastApplied() + 1; i <= toIndex; i++) {
batchRecoveredLogEntry(replicatedLog.get(i));
}
- context.setLastApplied(ale.getToIndex());
- context.setCommitIndex(ale.getToIndex());
+ context.setLastApplied(toIndex);
+ context.setCommitIndex(toIndex);
}
private void batchRecoveredLogEntry(ReplicatedLogEntry logEntry) {
applyState(applyState.getClientActor(), applyState.getIdentifier(),
applyState.getReplicatedLogEntry().getData());
- } else if (message instanceof ApplyLogEntries){
- ApplyLogEntries ale = (ApplyLogEntries) message;
+ } else if (message instanceof ApplyJournalEntries){
+ ApplyJournalEntries applyEntries = (ApplyJournalEntries) message;
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), ale.getToIndex());
+ LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), applyEntries.getToIndex());
}
- persistence().persist(new ApplyLogEntries(ale.getToIndex()), new Procedure<ApplyLogEntries>() {
- @Override
- public void apply(ApplyLogEntries param) throws Exception {
- }
- });
+
+ persistence().persist(applyEntries, APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK);
} else if(message instanceof ApplySnapshot ) {
Snapshot snapshot = ((ApplySnapshot) message).getSnapshot();
context.getReplicatedLog().size());
} else if (message instanceof CaptureSnapshot) {
- LOG.info("{}: CaptureSnapshot received by actor", persistenceId());
+ LOG.debug("{}: CaptureSnapshot received by actor: {}", persistenceId(), message);
if(captureSnapshot == null) {
captureSnapshot = (CaptureSnapshot)message;
// Apply the state immediately
applyState(clientActor, identifier, data);
- // Send a ApplyLogEntries message so that we write the fact that we applied
+ // Send a ApplyJournalEntries message so that we write the fact that we applied
// the state to durable storage
- self().tell(new ApplyLogEntries((int) replicatedLogEntry.getIndex()), self());
+ self().tell(new ApplyJournalEntries(replicatedLogEntry.getIndex()), self());
// Check if the "real" snapshot capture has been initiated. If no then do the fake snapshot
if(!context.isSnapshotCaptureInitiated()){
return context;
}
+ protected void updateConfigParams(ConfigParams configParams) {
+ context.setConfigParams(configParams);
+ }
+
/**
* setPeerAddress sets the address of a known peer at a later time.
* <p>
}
private void handleCaptureSnapshotReply(byte[] snapshotBytes) {
- LOG.info("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length);
+ LOG.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length);
// create a snapshot object from the state provided and save it
// when snapshot is saved async, SaveSnapshotSuccess is raised.
long dataThreshold = Runtime.getRuntime().totalMemory() *
getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100;
if (context.getReplicatedLog().dataSize() > dataThreshold) {
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("{}: dataSize {} exceeds dataThreshold {} - doing snapshotPreCommit with index {}",
+ persistenceId(), context.getReplicatedLog().dataSize(), dataThreshold,
+ captureSnapshot.getLastAppliedIndex());
+ }
+
// if memory is less, clear the log based on lastApplied.
// this could/should only happen if one of the followers is down
// as normally we keep removing from the log when its replicated to all.
context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(),
captureSnapshot.getLastAppliedTerm());
- getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
+ // Don't reset replicatedToAllIndex to -1 as this may prevent us from trimming the log after an
+ // install snapshot to a follower.
+ if(captureSnapshot.getReplicatedToAllIndex() >= 0) {
+ getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex());
+ }
} else if(captureSnapshot.getReplicatedToAllIndex() != -1){
// clear the log based on replicatedToAllIndex
context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(),
}
- LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " +
- "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(),
- captureSnapshot.getLastAppliedTerm());
+ LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex: {} " +
+ "and term: {}", persistenceId(), replicatedLog.getSnapshotIndex(),
+ replicatedLog.getSnapshotTerm());
if (isLeader() && captureSnapshot.isInstallSnapshotInitiated()) {
// this would be call straight to the leader and won't initiate in serialization
Code Review / controller.git / blobdiff