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