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()) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), applyEntries.getToIndex());
}
- persistence().persist(applyEntries, new Procedure<ApplyJournalEntries>() {
- @Override
- public void apply(ApplyJournalEntries 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;
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