package org.opendaylight.controller.cluster.raft.behaviors;
import akka.actor.ActorRef;
+import akka.japi.Procedure;
import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
* </ul>
*/
public class Follower extends AbstractRaftActorBehavior {
-
- private SnapshotTracker snapshotTracker = null;
+ private static final int SYNC_THRESHOLD = 10;
private final SyncStatusTracker initialSyncStatusTracker;
- private static final int SYNC_THRESHOLD = 10;
+ private final Procedure<ReplicatedLogEntry> appendAndPersistCallback = new Procedure<ReplicatedLogEntry>() {
+ @Override
+ public void apply(ReplicatedLogEntry logEntry) {
+ context.getReplicatedLog().captureSnapshotIfReady(logEntry);
+ }
+ };
+
+ private SnapshotTracker snapshotTracker = null;
public Follower(RaftActorContext context) {
this(context, null, (short)-1);
LOG.debug("{}: Append entry to log {}", logName(), entry.getData());
- context.getReplicatedLog().appendAndPersist(entry);
+ context.getReplicatedLog().appendAndPersist(entry, appendAndPersistCallback);
if(entry.getData() instanceof ServerConfigurationPayload) {
context.updatePeerIds((ServerConfigurationPayload)entry.getData());
// Create the leader and 2 follower actors and verify initial syncing of the followers after leader
// persistence recovery.
+ DefaultConfigParamsImpl followerConfigParams = newFollowerConfigParams();
+ followerConfigParams.setSnapshotBatchCount(snapshotBatchCount);
follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
- follower2Id, testActorPath(follower2Id)), newFollowerConfigParams());
+ follower2Id, testActorPath(follower2Id)), followerConfigParams);
follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
- follower1Id, testActorPath(follower1Id)), newFollowerConfigParams());
+ follower1Id, testActorPath(follower1Id)), followerConfigParams);
peerAddresses = ImmutableMap.<String, String>builder().
put(follower1Id, follower1Actor.path().toString()).
assertEquals("Leader last applied", 3, leaderContext.getLastApplied());
assertEquals("Leader replicatedToAllIndex", 2, leader.getReplicatedToAllIndex());
+ // The followers should also snapshot so verify.
+
+ MessageCollectorActor.expectFirstMatching(follower1CollectorActor, SaveSnapshotSuccess.class);
+ persistedSnapshots = InMemorySnapshotStore.getSnapshots(follower1Id, Snapshot.class);
+ assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
+ verifySnapshot("Persisted", persistedSnapshots.get(0), initialTerm, 2, currentTerm, 3);
+ unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
+ assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
+ verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3);
+
+ MessageCollectorActor.expectFirstMatching(follower2CollectorActor, SaveSnapshotSuccess.class);
+
MessageCollectorActor.clearMessages(leaderCollectorActor);
MessageCollectorActor.clearMessages(follower1CollectorActor);
MessageCollectorActor.clearMessages(follower2CollectorActor);