opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.raft;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12
  13 import akka.actor.ActorRef;
  14 import akka.persistence.SaveSnapshotSuccess;
  15 import com.google.common.collect.ImmutableMap;
  16 import com.google.common.util.concurrent.Uninterruptibles;
  17 import java.util.Arrays;
  18 import java.util.HashSet;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Set;
  22 import java.util.concurrent.TimeUnit;
  23 import javax.annotation.Nullable;
  24 import org.junit.Assert;
  25 import org.junit.Test;
  26 import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
  27 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
  28 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  29 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  30 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
  31 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  32 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  33 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  34 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  35 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  36 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
  37 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
  38 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
  39 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  40 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
  41 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  42 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  43 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  44
  45 /**
  46  * Tests replication and snapshots end-to-end using real RaftActors and behavior communication with a
  47  * lagging follower.
  48  *
  49  * @author Thomas Pantelis
  50  */
  51 public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends AbstractRaftActorIntegrationTest {
  52
  53     private void setup() {
  54         leaderId = factory.generateActorId("leader");
  55         follower1Id = factory.generateActorId("follower");
  56         follower2Id = factory.generateActorId("follower");
  57
  58         // Setup the persistent journal for the leader - just an election term and no journal/snapshots.
  59         InMemoryJournal.addEntry(leaderId, 1, new UpdateElectionTerm(initialTerm, leaderId));
  60
  61         // Create the leader and 2 follower actors.
  62         follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
  63                 follower2Id, testActorPath(follower2Id)), newFollowerConfigParams());
  64
  65         follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
  66                 follower1Id, testActorPath(follower1Id)), newFollowerConfigParams());
  67
  68         Map<String, String> leaderPeerAddresses = ImmutableMap.<String, String>builder()
  69                 .put(follower1Id, follower1Actor.path().toString())
  70                 .put(follower2Id, follower2Actor.path().toString()).build();
  71
  72         leaderConfigParams = newLeaderConfigParams();
  73         leaderActor = newTestRaftActor(leaderId, leaderPeerAddresses, leaderConfigParams);
  74
  75         waitUntilLeader(leaderActor);
  76
  77         leaderContext = leaderActor.underlyingActor().getRaftActorContext();
  78         leader = leaderActor.underlyingActor().getCurrentBehavior();
  79
  80         follower1Context = follower1Actor.underlyingActor().getRaftActorContext();
  81         follower1 = follower1Actor.underlyingActor().getCurrentBehavior();
  82
  83         follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
  84         follower2 = follower2Actor.underlyingActor().getCurrentBehavior();
  85
  86         currentTerm = leaderContext.getTermInformation().getCurrentTerm();
  87         assertEquals("Current term > " + initialTerm, true, currentTerm > initialTerm);
  88
  89         leaderCollectorActor = leaderActor.underlyingActor().collectorActor();
  90         follower1CollectorActor = follower1Actor.underlyingActor().collectorActor();
  91         follower2CollectorActor = follower2Actor.underlyingActor().collectorActor();
  92
  93         testLog.info("Leader created and elected");
  94     }
  95
  96     /**
  97      * Send 2 payload instances with follower 2 lagging then resume the follower and verifies it gets
  98      * caught up via AppendEntries.
  99      */
 100     @Test
 101     public void testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries() throws Exception {
 102         testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries starting: sending 2 new payloads");
 103
 104         setup();
 105
 106         // Simulate lagging by dropping AppendEntries messages in follower 2.
 107         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 108
 109         // Send the payloads.
 110         MockPayload payload0 = sendPayloadData(leaderActor, "zero");
 111         MockPayload payload1 = sendPayloadData(leaderActor, "one");
 112
 113         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 114         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 2);
 115         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload0.toString(), currentTerm, 0, payload0);
 116         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload1.toString(), currentTerm, 1, payload1);
 117
 118         // Verify follower 1 applies each log entry.
 119         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 2);
 120         verifyApplyState(applyStates.get(0), null, null, currentTerm, 0, payload0);
 121         verifyApplyState(applyStates.get(1), null, null, currentTerm, 1, payload1);
 122
 123         // Ensure there's at least 1 more heartbeat.
 124         MessageCollectorActor.clearMessages(leaderCollectorActor);
 125         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 126
 127         // The leader should not have performed fake snapshots to trim the log because the entries have not
 128         // been replicated to follower 2.
 129         assertEquals("Leader snapshot term", -1, leaderContext.getReplicatedLog().getSnapshotTerm());
 130         assertEquals("Leader snapshot index", -1, leaderContext.getReplicatedLog().getSnapshotIndex());
 131         assertEquals("Leader journal log size", 2, leaderContext.getReplicatedLog().size());
 132         assertEquals("Leader journal last index", 1, leaderContext.getReplicatedLog().lastIndex());
 133         assertEquals("Leader commit index", 1, leaderContext.getCommitIndex());
 134         assertEquals("Leader last applied", 1, leaderContext.getLastApplied());
 135         assertEquals("Leader replicatedToAllIndex", -1, leader.getReplicatedToAllIndex());
 136
 137         testLog.info(
 138             "testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries: new entries applied - resuming follower {}",
 139             follower2Id);
 140
 141         // Now stop dropping AppendEntries in follower 2.
 142         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 143
 144         // Verify follower 2 applies each log entry.
 145         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 2);
 146         verifyApplyState(applyStates.get(0), null, null, currentTerm, 0, payload0);
 147         verifyApplyState(applyStates.get(1), null, null, currentTerm, 1, payload1);
 148
 149         // Ensure there's at least 1 more heartbeat.
 150         MessageCollectorActor.clearMessages(leaderCollectorActor);
 151         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 152
 153         // The leader should now have performed fake snapshots to trim the log.
 154         verifyLeadersTrimmedLog(1);
 155
 156         // Even though follower 2 lagged behind, the leader should not have tried to install a snapshot
 157         // to catch it up because no snapshotting was done so the follower's next index was present in the log.
 158         InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor,
 159                 InstallSnapshot.class);
 160         Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot);
 161
 162         testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries complete");
 163     }
 164
 165     /**
 166      * Send payloads to trigger a leader snapshot due to snapshotBatchCount reached with follower 2
 167      * lagging but not enough for the leader to trim its log from the last applied index. Follower 2's log
 168      * will be behind by several entries and, when it is resumed, it should be caught up via AppendEntries
 169      * sent by the leader.
 170      */
 171     @Test
 172     public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries() throws Exception {
 173         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries starting");
 174
 175         setup();
 176
 177         sendInitialPayloadsReplicatedToAllFollowers("zero", "one");
 178
 179         // Configure follower 2 to drop messages and lag.
 180         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 181
 182         // Send the first payload and verify it gets applied by the leader and follower 1.
 183         MockPayload payload2 = sendPayloadData(leaderActor, "two");
 184
 185         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 186         verifyApplyState(applyState, leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 187
 188         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 189         verifyApplyState(applyState, null, null, currentTerm, 2, payload2);
 190
 191         expSnapshotState.add(payload2);
 192
 193         MessageCollectorActor.clearMessages(leaderCollectorActor);
 194         MessageCollectorActor.clearMessages(follower1CollectorActor);
 195
 196         // Send another payload - this should cause a snapshot due to snapshotBatchCount reached.
 197         MockPayload payload3 = sendPayloadData(leaderActor, "three");
 198
 199         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 200
 201         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries: sending 2 more payloads");
 202
 203         // Send 2 more payloads - not enough to trigger another snapshot.
 204         MockPayload payload4 = sendPayloadData(leaderActor, "four");
 205         MockPayload payload5 = sendPayloadData(leaderActor, "five");
 206
 207         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 208         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 3);
 209         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3);
 210         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 211         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload5.toString(), currentTerm, 5, payload5);
 212
 213         // Verify follower 1 applies each log entry.
 214         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3);
 215         verifyApplyState(applyStates.get(0), null, null, currentTerm, 3, payload3);
 216         verifyApplyState(applyStates.get(1), null, null, currentTerm, 4, payload4);
 217         verifyApplyState(applyStates.get(2), null, null, currentTerm, 5, payload5);
 218
 219         // The snapshot should have caused the leader to advanced the snapshot index to the
 220         // last previously applied index (1) that was replicated to all followers at the time of capture.
 221         // Note: since the log size (3) did not exceed the snapshot batch count (4), the leader should not
 222         // have trimmed the log to the last index actually applied (5).
 223         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
 224         assertEquals("Leader snapshot index", 1, leaderContext.getReplicatedLog().getSnapshotIndex());
 225         assertEquals("Leader journal log size", 4, leaderContext.getReplicatedLog().size());
 226         assertEquals("Leader journal last index", 5, leaderContext.getReplicatedLog().lastIndex());
 227         assertEquals("Leader commit index", 5, leaderContext.getCommitIndex());
 228         assertEquals("Leader last applied", 5, leaderContext.getLastApplied());
 229         assertEquals("Leader replicatedToAllIndex", 1, leader.getReplicatedToAllIndex());
 230
 231         // Now stop dropping AppendEntries in follower 2.
 232         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 233
 234         // Verify follower 2 applies each log entry. The leader should not install a snapshot b/c
 235         // follower 2's next index (3) is still present in the log.
 236         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 4);
 237         verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2);
 238         verifyApplyState(applyStates.get(1), null, null, currentTerm, 3, payload3);
 239         verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4);
 240         verifyApplyState(applyStates.get(3), null, null, currentTerm, 5, payload5);
 241
 242         // Verify the leader did not try to install a snapshot to catch up follower 2.
 243         InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor,
 244                 InstallSnapshot.class);
 245         Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot);
 246
 247         // Ensure there's at least 1 more heartbeat.
 248         MessageCollectorActor.clearMessages(leaderCollectorActor);
 249         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 250
 251         // The leader should now have performed fake snapshots to advance the snapshot index and to trim
 252         // the log. In addition replicatedToAllIndex should've advanced.
 253         verifyLeadersTrimmedLog(5);
 254
 255         // Verify the leader's persisted snapshot.
 256         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 257         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 258         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 2, currentTerm, 3);
 259         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 260         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 261         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3);
 262
 263         // Verify follower 1's log and snapshot indexes.
 264         MessageCollectorActor.clearMessages(follower1CollectorActor);
 265         MessageCollectorActor.expectFirstMatching(follower1CollectorActor, AppendEntries.class);
 266         verifyFollowersTrimmedLog(1, follower1Actor, 5);
 267
 268         // Verify follower 2's log and snapshot indexes.
 269         MessageCollectorActor.clearMessages(follower2CollectorActor);
 270         MessageCollectorActor.expectFirstMatching(follower2CollectorActor, AppendEntries.class);
 271         verifyFollowersTrimmedLog(2, follower2Actor, 5);
 272
 273         MessageCollectorActor.clearMessages(leaderCollectorActor);
 274         MessageCollectorActor.clearMessages(follower1CollectorActor);
 275         MessageCollectorActor.clearMessages(follower2CollectorActor);
 276
 277         expSnapshotState.add(payload3);
 278         expSnapshotState.add(payload4);
 279         expSnapshotState.add(payload5);
 280
 281         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries complete");
 282     }
 283
 284     /**
 285      * Send payloads to trigger a leader snapshot due to snapshotBatchCount reached with follower 2
 286      * lagging where the leader trims its log from the last applied index. Follower 2's log
 287      * will be behind by several entries and, when it is resumed, it should be caught up via a snapshot
 288      * installed by the leader.
 289      */
 290     @Test
 291     public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot() throws Exception {
 292         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot starting");
 293
 294         setup();
 295
 296         sendInitialPayloadsReplicatedToAllFollowers("zero", "one");
 297
 298         // Configure follower 2 to drop messages and lag.
 299         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 300
 301         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 302         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 303                 TimeUnit.MILLISECONDS);
 304
 305         // Send 5 payloads - the second should cause a leader snapshot.
 306         final MockPayload payload2 = sendPayloadData(leaderActor, "two");
 307         final MockPayload payload3 = sendPayloadData(leaderActor, "three");
 308         final MockPayload payload4 = sendPayloadData(leaderActor, "four");
 309         final MockPayload payload5 = sendPayloadData(leaderActor, "five");
 310         final MockPayload payload6 = sendPayloadData(leaderActor, "six");
 311
 312         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 313
 314         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 315         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 5);
 316         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 317         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 318         verifyApplyState(applyStates.get(4), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
 319
 320         MessageCollectorActor.clearMessages(leaderCollectorActor);
 321
 322         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot: "
 323                 + "sending 1 more payload to trigger second snapshot");
 324
 325         // Send another payload to trigger a second leader snapshot.
 326         MockPayload payload7 = sendPayloadData(leaderActor, "seven");
 327
 328         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 329
 330         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 331         verifyApplyState(applyState, leaderCollectorActor, payload7.toString(), currentTerm, 7, payload7);
 332
 333         // Verify follower 1 applies each log entry.
 334         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 6);
 335         verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2);
 336         verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4);
 337         verifyApplyState(applyStates.get(5), null, null, currentTerm, 7, payload7);
 338
 339         // The snapshot should have caused the leader to advanced the snapshot index to the leader's last
 340         // applied index (6) since the log size should have exceed the snapshot batch count (4).
 341         // replicatedToAllIndex should remain at 1 since follower 2 is lagging.
 342         verifyLeadersTrimmedLog(7, 1);
 343
 344         expSnapshotState.add(payload2);
 345         expSnapshotState.add(payload3);
 346         expSnapshotState.add(payload4);
 347         expSnapshotState.add(payload5);
 348         expSnapshotState.add(payload6);
 349
 350         MessageCollectorActor.clearMessages(leaderCollectorActor);
 351         MessageCollectorActor.clearMessages(follower1CollectorActor);
 352
 353         // Send a server config change to test that the install snapshot includes the server config.
 354
 355         ServerConfigurationPayload serverConfig = new ServerConfigurationPayload(Arrays.asList(
 356                 new ServerInfo(leaderId, true),
 357                 new ServerInfo(follower1Id, false),
 358                 new ServerInfo(follower2Id, false)));
 359         leaderContext.updatePeerIds(serverConfig);
 360         ((AbstractLeader)leader).updateMinReplicaCount();
 361         leaderActor.tell(serverConfig, ActorRef.noSender());
 362
 363         applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 364         verifyApplyState(applyState, leaderCollectorActor, "serverConfig", currentTerm, 8, serverConfig);
 365
 366         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 367         verifyApplyState(applyState, null, null, currentTerm, 8, serverConfig);
 368
 369         // Verify the leader's persisted snapshot.
 370         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 371         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 372         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 6, currentTerm, 7);
 373         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 374         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 375         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 7, payload7);
 376
 377         expSnapshotState.add(payload7);
 378
 379         verifyInstallSnapshotToLaggingFollower(8, serverConfig);
 380
 381         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot complete");
 382     }
 383
 384     /**
 385      * Send payloads with follower 2 lagging with the last payload having a large enough size to trigger a
 386      * leader snapshot such that the leader trims its log from the last applied index.. Follower 2's log will
 387      * be behind by several entries and, when it is resumed, it should be caught up via a snapshot installed
 388      * by the leader.
 389      */
 390     @Test
 391     public void testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower() throws Exception {
 392         testLog.info("testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower starting");
 393
 394         snapshotBatchCount = 5;
 395         setup();
 396
 397         sendInitialPayloadsReplicatedToAllFollowers("zero");
 398
 399         leaderActor.underlyingActor().setMockTotalMemory(1000);
 400
 401         // We'll expect a ReplicatedLogImplEntry message and an ApplyJournalEntries message added to the journal.
 402         InMemoryJournal.addWriteMessagesCompleteLatch(leaderId, 2);
 403
 404         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 405
 406         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 407         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 408                 TimeUnit.MILLISECONDS);
 409
 410         // Send a payload with a large relative size but not enough to trigger a snapshot.
 411         MockPayload payload1 = sendPayloadData(leaderActor, "one", 500);
 412
 413         // Verify the leader got consensus and applies the first log entry even though follower 2 didn't respond.
 414         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 1);
 415         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload1.toString(), currentTerm, 1, payload1);
 416
 417         // Wait for all the ReplicatedLogImplEntry and ApplyJournalEntries messages to be added to the journal
 418         // before the snapshot so the snapshot sequence # will be higher to ensure the snapshot gets
 419         // purged from the snapshot store after subsequent snapshots.
 420         InMemoryJournal.waitForWriteMessagesComplete(leaderId);
 421
 422         // Verify a snapshot is not triggered.
 423         CaptureSnapshot captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor,
 424                 CaptureSnapshot.class);
 425         Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot);
 426
 427         expSnapshotState.add(payload1);
 428
 429         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 430         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 431                 TimeUnit.MILLISECONDS);
 432
 433         // Send another payload with a large enough relative size in combination with the last payload
 434         // that exceeds the memory threshold (70% * 1000 = 700) - this should do a snapshot.
 435         MockPayload payload2 = sendPayloadData(leaderActor, "two", 201);
 436
 437         // Verify the leader applies the last log entry.
 438         applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 2);
 439         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 440
 441         // Verify follower 1 applies each log entry.
 442         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 2);
 443         verifyApplyState(applyStates.get(0), null, null, currentTerm, 1, payload1);
 444         verifyApplyState(applyStates.get(1), null, null, currentTerm, 2, payload2);
 445
 446         // A snapshot should've occurred - wait for it to complete.
 447         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 448
 449         // Because the snapshot was triggered by exceeding the memory threshold the leader should've advanced
 450         // the snapshot index to the last applied index and trimmed the log even though the entries weren't
 451         // replicated to all followers.
 452         verifyLeadersTrimmedLog(2, 0);
 453
 454         // Verify the leader's persisted snapshot.
 455         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 456         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 457         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 1, currentTerm, 2);
 458         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 459         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 460         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 2, payload2);
 461
 462         expSnapshotState.add(payload2);
 463
 464         verifyInstallSnapshotToLaggingFollower(2L, null);
 465
 466         // Sends a payload with index 3.
 467         verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot();
 468
 469         // Sends 3 payloads with indexes 4, 5 and 6.
 470         long leadersSnapshotIndexOnRecovery = verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot();
 471
 472         // Recover the leader from persistence and verify.
 473         long leadersLastIndexOnRecovery = 6;
 474
 475         long leadersFirstJournalEntryIndexOnRecovery = leadersSnapshotIndexOnRecovery + 1;
 476
 477         verifyLeaderRecoveryAfterReinstatement(leadersLastIndexOnRecovery, leadersSnapshotIndexOnRecovery,
 478                 leadersFirstJournalEntryIndexOnRecovery);
 479
 480         testLog.info("testLeaderSnapshotTriggeredByMemoryThresholdExceeded ending");
 481     }
 482
 483     /**
 484      * Send another payload to verify another snapshot is not done since the last snapshot trimmed the
 485      * first log entry so the memory threshold should not be exceeded.
 486      */
 487     private void verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot() throws Exception {
 488         ApplyState applyState;
 489         CaptureSnapshot captureSnapshot;
 490
 491         MockPayload payload3 = sendPayloadData(leaderActor, "three");
 492
 493         // Verify the leader applies the state.
 494         applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 495         verifyApplyState(applyState, leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3);
 496
 497         captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor, CaptureSnapshot.class);
 498         Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot);
 499
 500         // Verify the follower 1 applies the state.
 501         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 502         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
 503
 504         // Verify the follower 2 applies the state.
 505         applyState = MessageCollectorActor.expectFirstMatching(follower2CollectorActor, ApplyState.class);
 506         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
 507
 508         // Verify the leader's state.
 509         verifyLeadersTrimmedLog(3);
 510
 511         // Verify follower 1's state.
 512         verifyFollowersTrimmedLog(1, follower1Actor, 3);
 513
 514         // Verify follower 2's state.
 515         verifyFollowersTrimmedLog(2, follower2Actor, 3);
 516
 517         // Revert back to JVM total memory.
 518         leaderActor.underlyingActor().setMockTotalMemory(0);
 519
 520         MessageCollectorActor.clearMessages(leaderCollectorActor);
 521         MessageCollectorActor.clearMessages(follower1CollectorActor);
 522         MessageCollectorActor.clearMessages(follower2CollectorActor);
 523
 524         expSnapshotState.add(payload3);
 525     }
 526
 527     /**
 528      * Resume the lagging follower 2 and verify it receives an install snapshot from the leader.
 529      */
 530     private void verifyInstallSnapshotToLaggingFollower(long lastAppliedIndex,
 531             @Nullable ServerConfigurationPayload expServerConfig) throws Exception {
 532         testLog.info("verifyInstallSnapshotToLaggingFollower starting");
 533
 534         MessageCollectorActor.clearMessages(leaderCollectorActor);
 535
 536         // Now stop dropping AppendEntries in follower 2.
 537         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 538
 539
 540         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 541
 542         // Verify the leader's persisted snapshot. The previous snapshot (currently) won't be deleted from
 543         // the snapshot store because the second snapshot was initiated by the follower install snapshot and
 544         // not because the batch count was reached so the persisted journal sequence number wasn't advanced
 545         // far enough to cause the previous snapshot to be deleted. This is because
 546         // RaftActor#trimPersistentData subtracts the snapshotBatchCount from the snapshot's sequence number.
 547         // This is OK - the next snapshot should delete it. In production, even if the system restarted
 548         // before another snapshot, they would both get applied which wouldn't hurt anything.
 549         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 550         Assert.assertTrue("Expected at least 1 persisted snapshots", persistedSnapshots.size() > 0);
 551         Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1);
 552         verifySnapshot("Persisted", persistedSnapshot, currentTerm, lastAppliedIndex, currentTerm, lastAppliedIndex);
 553         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshot.getUnAppliedEntries();
 554         assertEquals("Persisted Snapshot getUnAppliedEntries size", 0, unAppliedEntry.size());
 555
 556         int snapshotSize = persistedSnapshot.getState().length;
 557         final int expTotalChunks = snapshotSize / SNAPSHOT_CHUNK_SIZE
 558                 + (snapshotSize % SNAPSHOT_CHUNK_SIZE > 0 ? 1 : 0);
 559
 560         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor,
 561                 InstallSnapshot.class);
 562         assertEquals("InstallSnapshot getTerm", currentTerm, installSnapshot.getTerm());
 563         assertEquals("InstallSnapshot getLeaderId", leaderId, installSnapshot.getLeaderId());
 564         assertEquals("InstallSnapshot getChunkIndex", 1, installSnapshot.getChunkIndex());
 565         assertEquals("InstallSnapshot getTotalChunks", expTotalChunks, installSnapshot.getTotalChunks());
 566         assertEquals("InstallSnapshot getLastIncludedTerm", currentTerm, installSnapshot.getLastIncludedTerm());
 567         assertEquals("InstallSnapshot getLastIncludedIndex", lastAppliedIndex, installSnapshot.getLastIncludedIndex());
 568         //assertArrayEquals("InstallSnapshot getData", snapshot, installSnapshot.getData().toByteArray());
 569
 570         List<InstallSnapshotReply> installSnapshotReplies = MessageCollectorActor.expectMatching(
 571                 leaderCollectorActor, InstallSnapshotReply.class, expTotalChunks);
 572         int index = 1;
 573         for (InstallSnapshotReply installSnapshotReply: installSnapshotReplies) {
 574             assertEquals("InstallSnapshotReply getTerm", currentTerm, installSnapshotReply.getTerm());
 575             assertEquals("InstallSnapshotReply getChunkIndex", index++, installSnapshotReply.getChunkIndex());
 576             assertEquals("InstallSnapshotReply getFollowerId", follower2Id, installSnapshotReply.getFollowerId());
 577             assertEquals("InstallSnapshotReply isSuccess", true, installSnapshotReply.isSuccess());
 578         }
 579
 580         // Verify follower 2 applies the snapshot.
 581         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor,
 582                 ApplySnapshot.class);
 583         verifySnapshot("Follower 2", applySnapshot.getSnapshot(), currentTerm, lastAppliedIndex, currentTerm,
 584                 lastAppliedIndex);
 585         assertEquals("Persisted Snapshot getUnAppliedEntries size", 0,
 586                 applySnapshot.getSnapshot().getUnAppliedEntries().size());
 587
 588         // Wait for the snapshot to complete.
 589         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 590
 591         // Ensure there's at least 1 more heartbeat.
 592         MessageCollectorActor.clearMessages(leaderCollectorActor);
 593         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 594
 595         // The leader should now have performed fake snapshots to advance the snapshot index and to trim
 596         // the log. In addition replicatedToAllIndex should've advanced.
 597         verifyLeadersTrimmedLog(lastAppliedIndex);
 598
 599         if (expServerConfig != null) {
 600             Set<ServerInfo> expServerInfo = new HashSet<>(expServerConfig.getServerConfig());
 601             assertEquals("Leader snapshot server config", expServerInfo,
 602                     new HashSet<>(persistedSnapshot.getServerConfiguration().getServerConfig()));
 603
 604             assertEquals("Follower 2 snapshot server config", expServerInfo,
 605                     new HashSet<>(applySnapshot.getSnapshot().getServerConfiguration().getServerConfig()));
 606
 607             ServerConfigurationPayload follower2ServerConfig = follower2Context.getPeerServerInfo(true);
 608             assertNotNull("Follower 2 server config is null", follower2ServerConfig);
 609
 610             assertEquals("Follower 2 server config", expServerInfo,
 611                     new HashSet<>(follower2ServerConfig.getServerConfig()));
 612         }
 613
 614         MessageCollectorActor.clearMessages(leaderCollectorActor);
 615         MessageCollectorActor.clearMessages(follower1CollectorActor);
 616         MessageCollectorActor.clearMessages(follower2CollectorActor);
 617
 618         testLog.info("verifyInstallSnapshotToLaggingFollower complete");
 619     }
 620
 621     /**
 622      * Do another round of payloads and snapshot to verify replicatedToAllIndex gets back on track and
 623      * snapshots works as expected after doing a follower snapshot. In this step we don't lag a follower.
 624      */
 625     private long verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot() throws Exception {
 626         testLog.info(
 627                 "verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot starting: replicatedToAllIndex: {}",
 628                 leader.getReplicatedToAllIndex());
 629
 630         // Send another payload - a snapshot should occur.
 631         MockPayload payload4 = sendPayloadData(leaderActor, "four");
 632
 633         // Wait for the snapshot to complete.
 634         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 635
 636         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 637         verifyApplyState(applyState, leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 638
 639         // Verify the leader's last persisted snapshot (previous ones may not be purged yet).
 640         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 641         Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1);
 642         // The last (fourth) payload may or may not have been applied when the snapshot is captured depending on the
 643         // timing when the async persistence completes.
 644         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshot.getUnAppliedEntries();
 645         long leadersSnapshotIndex;
 646         if (unAppliedEntry.isEmpty()) {
 647             leadersSnapshotIndex = 4;
 648             expSnapshotState.add(payload4);
 649             verifySnapshot("Persisted", persistedSnapshot, currentTerm, 4, currentTerm, 4);
 650         } else {
 651             leadersSnapshotIndex = 3;
 652             verifySnapshot("Persisted", persistedSnapshot, currentTerm, 3, currentTerm, 4);
 653             assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 654             verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 4, payload4);
 655             expSnapshotState.add(payload4);
 656         }
 657
 658         // Send a couple more payloads.
 659         MockPayload payload5 = sendPayloadData(leaderActor, "five");
 660         MockPayload payload6 = sendPayloadData(leaderActor, "six");
 661
 662         // Verify the leader applies the 2 log entries.
 663         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 3);
 664         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload5.toString(), currentTerm, 5, payload5);
 665         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
 666
 667         // Verify the leader applies a log entry for at least the last entry index.
 668         verifyApplyJournalEntries(leaderCollectorActor, 6);
 669
 670         // Ensure there's at least 1 more heartbeat to trim the log.
 671         MessageCollectorActor.clearMessages(leaderCollectorActor);
 672         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 673
 674         // Verify the leader's final state.
 675         verifyLeadersTrimmedLog(6);
 676
 677         InMemoryJournal.dumpJournal(leaderId);
 678
 679         // Verify the leaders's persisted journal log - it should only contain the last 2 ReplicatedLogEntries
 680         // added after the snapshot as the persisted journal should've been purged to the snapshot
 681         // sequence number.
 682         verifyPersistedJournal(leaderId, Arrays.asList(new SimpleReplicatedLogEntry(5, currentTerm, payload5),
 683                 new SimpleReplicatedLogEntry(6, currentTerm, payload6)));
 684
 685         // Verify the leaders's persisted journal contains an ApplyJournalEntries for at least the last entry index.
 686         List<ApplyJournalEntries> persistedApplyJournalEntries =
 687                 InMemoryJournal.get(leaderId, ApplyJournalEntries.class);
 688         boolean found = false;
 689         for (ApplyJournalEntries entry: persistedApplyJournalEntries) {
 690             if (entry.getToIndex() == 6) {
 691                 found = true;
 692                 break;
 693             }
 694         }
 695
 696         Assert.assertTrue(String.format("ApplyJournalEntries with index %d not found in leader's persisted journal", 6),
 697                 found);
 698
 699         // Verify follower 1 applies the 3 log entries.
 700         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3);
 701         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
 702         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
 703         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
 704
 705         // Verify follower 1's log state.
 706         verifyFollowersTrimmedLog(1, follower1Actor, 6);
 707
 708         // Verify follower 2 applies the 3 log entries.
 709         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 3);
 710         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
 711         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
 712         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
 713
 714         // Verify follower 2's log state.
 715         verifyFollowersTrimmedLog(2, follower2Actor, 6);
 716
 717         expSnapshotState.add(payload5);
 718         expSnapshotState.add(payload6);
 719
 720         testLog.info("verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot ending");
 721
 722         return leadersSnapshotIndex;
 723     }
 724
 725     /**
 726      * Kill the leader actor, reinstate it and verify the recovered journal.
 727      */
 728     private void verifyLeaderRecoveryAfterReinstatement(long lastIndex, long snapshotIndex,
 729             long firstJournalEntryIndex) {
 730         testLog.info("verifyLeaderRecoveryAfterReinstatement starting: lastIndex: {}, snapshotIndex: {}, "
 731             + "firstJournalEntryIndex: {}", lastIndex, snapshotIndex, firstJournalEntryIndex);
 732
 733         killActor(leaderActor);
 734
 735         leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams);
 736         TestRaftActor testRaftActor = leaderActor.underlyingActor();
 737
 738         testRaftActor.startDropMessages(RequestVoteReply.class);
 739
 740         leaderContext = testRaftActor.getRaftActorContext();
 741
 742         testRaftActor.waitForRecoveryComplete();
 743
 744         int logSize = (int) (expSnapshotState.size() - firstJournalEntryIndex);
 745         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
 746         assertEquals("Leader snapshot index", snapshotIndex, leaderContext.getReplicatedLog().getSnapshotIndex());
 747         assertEquals("Leader journal log size", logSize, leaderContext.getReplicatedLog().size());
 748         assertEquals("Leader journal last index", lastIndex, leaderContext.getReplicatedLog().lastIndex());
 749         assertEquals("Leader commit index", lastIndex, leaderContext.getCommitIndex());
 750         assertEquals("Leader last applied", lastIndex, leaderContext.getLastApplied());
 751
 752         for (long i = firstJournalEntryIndex; i < expSnapshotState.size(); i++) {
 753             verifyReplicatedLogEntry(leaderContext.getReplicatedLog().get(i), currentTerm, i,
 754                     expSnapshotState.get((int) i));
 755         }
 756
 757         assertEquals("Leader applied state", expSnapshotState, testRaftActor.getState());
 758
 759         testLog.info("verifyLeaderRecoveryAfterReinstatement ending");
 760     }
 761
 762     private void sendInitialPayloadsReplicatedToAllFollowers(String... data) {
 763
 764         // Send the payloads.
 765         for (String d: data) {
 766             expSnapshotState.add(sendPayloadData(leaderActor, d));
 767         }
 768
 769         int numEntries = data.length;
 770
 771         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 772         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor,
 773                 ApplyState.class, numEntries);
 774         for (int i = 0; i < expSnapshotState.size(); i++) {
 775             MockPayload payload = expSnapshotState.get(i);
 776             verifyApplyState(applyStates.get(i), leaderCollectorActor, payload.toString(), currentTerm, i, payload);
 777         }
 778
 779         // Verify follower 1 applies each log entry.
 780         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, numEntries);
 781         for (int i = 0; i < expSnapshotState.size(); i++) {
 782             MockPayload payload = expSnapshotState.get(i);
 783             verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload);
 784         }
 785
 786         // Verify follower 2 applies each log entry.
 787         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, numEntries);
 788         for (int i = 0; i < expSnapshotState.size(); i++) {
 789             MockPayload payload = expSnapshotState.get(i);
 790             verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload);
 791         }
 792
 793         // Ensure there's at least 1 more heartbeat.
 794         MessageCollectorActor.clearMessages(leaderCollectorActor);
 795         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 796
 797         // The leader should have performed fake snapshots to trim the log to the last index replicated to
 798         // all followers.
 799         verifyLeadersTrimmedLog(numEntries - 1);
 800
 801         MessageCollectorActor.clearMessages(leaderCollectorActor);
 802         MessageCollectorActor.clearMessages(follower1CollectorActor);
 803         MessageCollectorActor.clearMessages(follower2CollectorActor);
 804     }
 805 }