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

   1 /*
   2  * Copyright (c) 2015 Cisco 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
   9 package org.opendaylight.controller.cluster.raft;
  10
  11 import static org.junit.Assert.assertArrayEquals;
  12 import static org.junit.Assert.assertEquals;
  13 import static org.junit.Assert.assertFalse;
  14 import static org.junit.Assert.assertTrue;
  15 import static org.mockito.Matchers.any;
  16 import static org.mockito.Matchers.anyLong;
  17 import static org.mockito.Matchers.anyObject;
  18 import static org.mockito.Mockito.doNothing;
  19 import static org.mockito.Mockito.doReturn;
  20 import static org.mockito.Mockito.doThrow;
  21 import static org.mockito.Mockito.mock;
  22 import static org.mockito.Mockito.never;
  23 import static org.mockito.Mockito.reset;
  24 import static org.mockito.Mockito.times;
  25 import static org.mockito.Mockito.verify;
  26
  27 import akka.actor.ActorRef;
  28 import akka.persistence.SnapshotSelectionCriteria;
  29 import java.io.OutputStream;
  30 import java.util.Arrays;
  31 import java.util.Optional;
  32 import java.util.function.Consumer;
  33 import org.junit.After;
  34 import org.junit.Before;
  35 import org.junit.Test;
  36 import org.mockito.ArgumentCaptor;
  37 import org.mockito.Mock;
  38 import org.mockito.MockitoAnnotations;
  39 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  40 import org.opendaylight.controller.cluster.io.FileBackedOutputStreamFactory;
  41 import org.opendaylight.controller.cluster.raft.SnapshotManager.LastAppliedTermInformationReader;
  42 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  43 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  44 import org.opendaylight.controller.cluster.raft.base.messages.SnapshotComplete;
  45 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  46 import org.opendaylight.controller.cluster.raft.persisted.ByteState;
  47 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  48 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  49 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  50 import org.slf4j.LoggerFactory;
  51
  52 public class SnapshotManagerTest extends AbstractActorTest {
  53
  54     @Mock
  55     private RaftActorContext mockRaftActorContext;
  56
  57     @Mock
  58     private ConfigParams mockConfigParams;
  59
  60     @Mock
  61     private ReplicatedLog mockReplicatedLog;
  62
  63     @Mock
  64     private DataPersistenceProvider mockDataPersistenceProvider;
  65
  66     @Mock
  67     private RaftActorBehavior mockRaftActorBehavior;
  68
  69     @Mock
  70     private Consumer<Optional<OutputStream>> mockProcedure;
  71
  72     @Mock
  73     private ElectionTerm mockElectionTerm;
  74
  75     private SnapshotManager snapshotManager;
  76
  77     private TestActorFactory factory;
  78
  79     private ActorRef actorRef;
  80
  81     @Before
  82     public void setUp() {
  83         MockitoAnnotations.initMocks(this);
  84
  85         doReturn(false).when(mockRaftActorContext).hasFollowers();
  86         doReturn(mockConfigParams).when(mockRaftActorContext).getConfigParams();
  87         doReturn(10L).when(mockConfigParams).getSnapshotBatchCount();
  88         doReturn(70).when(mockConfigParams).getSnapshotDataThresholdPercentage();
  89         doReturn(mockReplicatedLog).when(mockRaftActorContext).getReplicatedLog();
  90         doReturn("123").when(mockRaftActorContext).getId();
  91         doReturn(mockDataPersistenceProvider).when(mockRaftActorContext).getPersistenceProvider();
  92         doReturn(mockRaftActorBehavior).when(mockRaftActorContext).getCurrentBehavior();
  93         doReturn("123").when(mockRaftActorBehavior).getLeaderId();
  94
  95         doReturn(mockElectionTerm).when(mockRaftActorContext).getTermInformation();
  96         doReturn(5L).when(mockElectionTerm).getCurrentTerm();
  97         doReturn("member5").when(mockElectionTerm).getVotedFor();
  98
  99         doReturn(new FileBackedOutputStreamFactory(10000000, "target"))
 100                 .when(mockRaftActorContext).getFileBackedOutputStreamFactory();
 101
 102         snapshotManager = new SnapshotManager(mockRaftActorContext, LoggerFactory.getLogger(this.getClass()));
 103         factory = new TestActorFactory(getSystem());
 104
 105         actorRef = factory.createActor(MessageCollectorActor.props(), factory.generateActorId("test-"));
 106         doReturn(actorRef).when(mockRaftActorContext).getActor();
 107
 108         snapshotManager.setCreateSnapshotConsumer(mockProcedure);
 109     }
 110
 111     @After
 112     public void tearDown() {
 113         factory.close();
 114     }
 115
 116     @Test
 117     public void testConstruction() {
 118         assertEquals(false, snapshotManager.isCapturing());
 119     }
 120
 121     @SuppressWarnings({ "unchecked", "rawtypes" })
 122     @Test
 123     public void testCaptureToInstall() {
 124
 125         // Force capturing toInstall = true
 126         snapshotManager.captureToInstall(new SimpleReplicatedLogEntry(0, 1,
 127                 new MockRaftActorContext.MockPayload()), 0, "follower-1");
 128
 129         assertEquals(true, snapshotManager.isCapturing());
 130
 131         ArgumentCaptor<Optional> outputStream = ArgumentCaptor.forClass(Optional.class);
 132         verify(mockProcedure).accept(outputStream.capture());
 133         assertEquals("isPresent", true, outputStream.getValue().isPresent());
 134
 135         CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
 136
 137         // LastIndex and LastTerm are picked up from the lastLogEntry
 138         assertEquals(0L, captureSnapshot.getLastIndex());
 139         assertEquals(1L, captureSnapshot.getLastTerm());
 140
 141         // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
 142         assertEquals(0L, captureSnapshot.getLastAppliedIndex());
 143         assertEquals(1L, captureSnapshot.getLastAppliedTerm());
 144
 145         //
 146         assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
 147         assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
 148         MessageCollectorActor.clearMessages(actorRef);
 149     }
 150
 151     @SuppressWarnings({ "rawtypes", "unchecked" })
 152     @Test
 153     public void testCapture() {
 154         boolean capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 155                 new MockRaftActorContext.MockPayload()), 9);
 156
 157         assertTrue(capture);
 158
 159         assertEquals(true, snapshotManager.isCapturing());
 160
 161         ArgumentCaptor<Optional> outputStream = ArgumentCaptor.forClass(Optional.class);
 162         verify(mockProcedure).accept(outputStream.capture());
 163         assertEquals("isPresent", false, outputStream.getValue().isPresent());
 164
 165         CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
 166
 167         // LastIndex and LastTerm are picked up from the lastLogEntry
 168         assertEquals(9L, captureSnapshot.getLastIndex());
 169         assertEquals(1L, captureSnapshot.getLastTerm());
 170
 171         // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
 172         assertEquals(9L, captureSnapshot.getLastAppliedIndex());
 173         assertEquals(1L, captureSnapshot.getLastAppliedTerm());
 174
 175         //
 176         assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
 177         assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
 178
 179         MessageCollectorActor.clearMessages(actorRef);
 180     }
 181
 182     @SuppressWarnings({ "unchecked", "rawtypes" })
 183     @Test
 184     public void testCaptureWithNullLastLogEntry() {
 185         boolean capture = snapshotManager.capture(null, 1);
 186
 187         assertTrue(capture);
 188
 189         assertEquals(true, snapshotManager.isCapturing());
 190
 191         ArgumentCaptor<Optional> outputStream = ArgumentCaptor.forClass(Optional.class);
 192         verify(mockProcedure).accept(outputStream.capture());
 193         assertEquals("isPresent", false, outputStream.getValue().isPresent());
 194
 195         CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
 196
 197         // LastIndex and LastTerm are picked up from the lastLogEntry
 198         assertEquals(-1L, captureSnapshot.getLastIndex());
 199         assertEquals(-1L, captureSnapshot.getLastTerm());
 200
 201         // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
 202         assertEquals(-1L, captureSnapshot.getLastAppliedIndex());
 203         assertEquals(-1L, captureSnapshot.getLastAppliedTerm());
 204
 205         //
 206         assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
 207         assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
 208         MessageCollectorActor.clearMessages(actorRef);
 209     }
 210
 211     @Test
 212     public void testCaptureWithCreateProcedureError() {
 213         doThrow(new RuntimeException("mock")).when(mockProcedure).accept(anyObject());
 214
 215         boolean capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 216                 new MockRaftActorContext.MockPayload()), 9);
 217
 218         assertFalse(capture);
 219
 220         assertEquals(false, snapshotManager.isCapturing());
 221
 222         verify(mockProcedure).accept(anyObject());
 223     }
 224
 225     @SuppressWarnings("unchecked")
 226     @Test
 227     public void testIllegalCapture() {
 228         boolean capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 229                 new MockRaftActorContext.MockPayload()), 9);
 230
 231         assertTrue(capture);
 232
 233         verify(mockProcedure).accept(anyObject());
 234
 235         reset(mockProcedure);
 236
 237         // This will not cause snapshot capture to start again
 238         capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 239                 new MockRaftActorContext.MockPayload()), 9);
 240
 241         assertFalse(capture);
 242
 243         verify(mockProcedure, never()).accept(anyObject());
 244     }
 245
 246     @Test
 247     public void testPersistWhenReplicatedToAllIndexMinusOne() {
 248         doReturn(7L).when(mockReplicatedLog).getSnapshotIndex();
 249         doReturn(1L).when(mockReplicatedLog).getSnapshotTerm();
 250
 251         doReturn(true).when(mockRaftActorContext).hasFollowers();
 252
 253         doReturn(8L).when(mockRaftActorContext).getLastApplied();
 254
 255         ReplicatedLogEntry lastLogEntry = new SimpleReplicatedLogEntry(9L, 3L, new MockRaftActorContext.MockPayload());
 256
 257         ReplicatedLogEntry lastAppliedEntry = new SimpleReplicatedLogEntry(
 258                 8L, 2L, new MockRaftActorContext.MockPayload());
 259
 260         doReturn(lastAppliedEntry).when(mockReplicatedLog).get(8L);
 261         doReturn(Arrays.asList(lastLogEntry)).when(mockReplicatedLog).getFrom(9L);
 262
 263         // when replicatedToAllIndex = -1
 264         snapshotManager.capture(lastLogEntry, -1);
 265
 266         ByteState snapshotState = ByteState.of(new byte[] {1,2,3,4,5,6,7,8,9,10});
 267         snapshotManager.persist(snapshotState, Optional.empty(), Runtime.getRuntime().totalMemory());
 268
 269         ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
 270         verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
 271
 272         Snapshot snapshot = snapshotArgumentCaptor.getValue();
 273
 274         assertEquals("getLastTerm", 3L, snapshot.getLastTerm());
 275         assertEquals("getLastIndex", 9L, snapshot.getLastIndex());
 276         assertEquals("getLastAppliedTerm", 2L, snapshot.getLastAppliedTerm());
 277         assertEquals("getLastAppliedIndex", 8L, snapshot.getLastAppliedIndex());
 278         assertEquals("getState", snapshotState, snapshot.getState());
 279         assertEquals("getUnAppliedEntries", Arrays.asList(lastLogEntry), snapshot.getUnAppliedEntries());
 280         assertEquals("electionTerm", mockElectionTerm.getCurrentTerm(), snapshot.getElectionTerm());
 281         assertEquals("electionVotedFor", mockElectionTerm.getVotedFor(), snapshot.getElectionVotedFor());
 282
 283         verify(mockReplicatedLog).snapshotPreCommit(7L, 1L);
 284     }
 285
 286     @Test
 287     public void testPersistWhenReplicatedToAllIndexNotMinus() {
 288         doReturn(45L).when(mockReplicatedLog).getSnapshotIndex();
 289         doReturn(6L).when(mockReplicatedLog).getSnapshotTerm();
 290         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 291         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(9);
 292         doReturn(6L).when(replicatedLogEntry).getTerm();
 293         doReturn(9L).when(replicatedLogEntry).getIndex();
 294
 295         // when replicatedToAllIndex != -1
 296         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), 9);
 297
 298         ByteState snapshotState = ByteState.of(new byte[] {1,2,3,4,5,6,7,8,9,10});
 299         snapshotManager.persist(snapshotState, Optional.empty(), Runtime.getRuntime().totalMemory());
 300
 301         ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
 302         verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
 303
 304         Snapshot snapshot = snapshotArgumentCaptor.getValue();
 305
 306         assertEquals("getLastTerm", 6L, snapshot.getLastTerm());
 307         assertEquals("getLastIndex", 9L, snapshot.getLastIndex());
 308         assertEquals("getLastAppliedTerm", 6L, snapshot.getLastAppliedTerm());
 309         assertEquals("getLastAppliedIndex", 9L, snapshot.getLastAppliedIndex());
 310         assertEquals("getState", snapshotState, snapshot.getState());
 311         assertEquals("getUnAppliedEntries size", 0, snapshot.getUnAppliedEntries().size());
 312
 313         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 314
 315         verify(mockRaftActorBehavior).setReplicatedToAllIndex(9);
 316     }
 317
 318     @Test
 319     public void testPersistWhenReplicatedLogDataSizeGreaterThanThreshold() {
 320         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 321
 322         // when replicatedToAllIndex = -1
 323         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 324
 325         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 326
 327         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 328
 329         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 330
 331         verify(mockRaftActorBehavior, never()).setReplicatedToAllIndex(anyLong());
 332     }
 333
 334     @Test
 335     public void testPersistWhenReplicatedLogSizeExceedsSnapshotBatchCount() {
 336         doReturn(10L).when(mockReplicatedLog).size(); // matches snapshotBatchCount
 337         doReturn(100).when(mockReplicatedLog).dataSize();
 338
 339         doReturn(5L).when(mockReplicatedLog).getSnapshotIndex();
 340         doReturn(5L).when(mockReplicatedLog).getSnapshotTerm();
 341
 342         long replicatedToAllIndex = 1;
 343         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 344         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(replicatedToAllIndex);
 345         doReturn(6L).when(replicatedLogEntry).getTerm();
 346         doReturn(replicatedToAllIndex).when(replicatedLogEntry).getIndex();
 347
 348         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6,
 349                 new MockRaftActorContext.MockPayload()), replicatedToAllIndex);
 350
 351         snapshotManager.persist(ByteState.empty(), Optional.empty(), 2000000L);
 352
 353         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 354
 355         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 356
 357         verify(mockRaftActorBehavior).setReplicatedToAllIndex(replicatedToAllIndex);
 358     }
 359
 360     @SuppressWarnings({ "rawtypes", "unchecked" })
 361     @Test
 362     public void testPersistSendInstallSnapshot() throws Exception {
 363         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 364         doNothing().when(mockProcedure).accept(anyObject());
 365
 366         // when replicatedToAllIndex = -1
 367         boolean capture = snapshotManager.captureToInstall(new SimpleReplicatedLogEntry(9, 6,
 368                 new MockRaftActorContext.MockPayload()), -1, "follower-1");
 369
 370         assertTrue(capture);
 371
 372         ByteState snapshotState = ByteState.of(new byte[] {1,2,3,4,5,6,7,8,9,10});
 373
 374         ArgumentCaptor<Optional> installSnapshotStreamCapture = ArgumentCaptor.forClass(Optional.class);
 375         verify(mockProcedure).accept(installSnapshotStreamCapture.capture());
 376
 377         Optional<OutputStream> installSnapshotStream = installSnapshotStreamCapture.getValue();
 378         assertEquals("isPresent", true, installSnapshotStream.isPresent());
 379
 380         installSnapshotStream.get().write(snapshotState.getBytes());
 381
 382         snapshotManager.persist(snapshotState, installSnapshotStream, Runtime.getRuntime().totalMemory());
 383
 384         assertEquals(true, snapshotManager.isCapturing());
 385
 386         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 387
 388         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 389
 390         ArgumentCaptor<SendInstallSnapshot> sendInstallSnapshotArgumentCaptor
 391                 = ArgumentCaptor.forClass(SendInstallSnapshot.class);
 392
 393         verify(mockRaftActorBehavior).handleMessage(any(ActorRef.class), sendInstallSnapshotArgumentCaptor.capture());
 394
 395         SendInstallSnapshot sendInstallSnapshot = sendInstallSnapshotArgumentCaptor.getValue();
 396
 397         assertEquals("state", snapshotState, sendInstallSnapshot.getSnapshot().getState());
 398         assertArrayEquals("state", snapshotState.getBytes(), sendInstallSnapshot.getSnapshotBytes().read());
 399     }
 400
 401     @Test
 402     public void testCallingPersistWithoutCaptureWillDoNothing() {
 403         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 404
 405         verify(mockDataPersistenceProvider, never()).saveSnapshot(any(Snapshot.class));
 406
 407         verify(mockReplicatedLog, never()).snapshotPreCommit(9L, 6L);
 408
 409         verify(mockRaftActorBehavior, never()).handleMessage(any(ActorRef.class), any(SendInstallSnapshot.class));
 410     }
 411
 412     @Test
 413     public void testCallingPersistTwiceWillDoNoHarm() {
 414         doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
 415
 416         // when replicatedToAllIndex = -1
 417         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 418
 419         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 420
 421         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 422
 423         verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
 424
 425         verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
 426     }
 427
 428     @Test
 429     public void testCommit() {
 430         doReturn(50L).when(mockDataPersistenceProvider).getLastSequenceNumber();
 431
 432         // when replicatedToAllIndex = -1
 433         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 434
 435         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 436
 437         assertEquals(true, snapshotManager.isCapturing());
 438
 439         snapshotManager.commit(100L, 1234L);
 440
 441         assertEquals(false, snapshotManager.isCapturing());
 442
 443         verify(mockReplicatedLog).snapshotCommit();
 444
 445         verify(mockDataPersistenceProvider).deleteMessages(50L);
 446
 447         ArgumentCaptor<SnapshotSelectionCriteria> criteriaCaptor =
 448                 ArgumentCaptor.forClass(SnapshotSelectionCriteria.class);
 449
 450         verify(mockDataPersistenceProvider).deleteSnapshots(criteriaCaptor.capture());
 451
 452         assertEquals(100L, criteriaCaptor.getValue().maxSequenceNr());
 453         assertEquals(1233L, criteriaCaptor.getValue().maxTimestamp());
 454
 455         MessageCollectorActor.expectFirstMatching(actorRef, SnapshotComplete.class);
 456     }
 457
 458     @Test
 459     public void testCommitBeforePersist() {
 460         // when replicatedToAllIndex = -1
 461         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 462
 463         snapshotManager.commit(100L, 0);
 464
 465         verify(mockReplicatedLog, never()).snapshotCommit();
 466
 467         verify(mockDataPersistenceProvider, never()).deleteMessages(100L);
 468
 469         verify(mockDataPersistenceProvider, never()).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 470
 471     }
 472
 473     @Test
 474     public void testCommitBeforeCapture() {
 475         snapshotManager.commit(100L, 0);
 476
 477         verify(mockReplicatedLog, never()).snapshotCommit();
 478
 479         verify(mockDataPersistenceProvider, never()).deleteMessages(anyLong());
 480
 481         verify(mockDataPersistenceProvider, never()).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 482
 483     }
 484
 485     @Test
 486     public void testCallingCommitMultipleTimesCausesNoHarm() {
 487         doReturn(50L).when(mockDataPersistenceProvider).getLastSequenceNumber();
 488
 489         // when replicatedToAllIndex = -1
 490         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 491
 492         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 493
 494         snapshotManager.commit(100L, 0);
 495
 496         snapshotManager.commit(100L, 0);
 497
 498         verify(mockReplicatedLog, times(1)).snapshotCommit();
 499
 500         verify(mockDataPersistenceProvider, times(1)).deleteMessages(50L);
 501
 502         verify(mockDataPersistenceProvider, times(1)).deleteSnapshots(any(SnapshotSelectionCriteria.class));
 503     }
 504
 505     @Test
 506     public void testRollback() {
 507         // when replicatedToAllIndex = -1
 508         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 509
 510         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 511
 512         snapshotManager.rollback();
 513
 514         verify(mockReplicatedLog).snapshotRollback();
 515
 516         MessageCollectorActor.expectFirstMatching(actorRef, SnapshotComplete.class);
 517     }
 518
 519
 520     @Test
 521     public void testRollbackBeforePersist() {
 522         // when replicatedToAllIndex = -1
 523         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 524
 525         snapshotManager.rollback();
 526
 527         verify(mockReplicatedLog, never()).snapshotRollback();
 528     }
 529
 530     @Test
 531     public void testRollbackBeforeCapture() {
 532         snapshotManager.rollback();
 533
 534         verify(mockReplicatedLog, never()).snapshotRollback();
 535     }
 536
 537     @Test
 538     public void testCallingRollbackMultipleTimesCausesNoHarm() {
 539         // when replicatedToAllIndex = -1
 540         snapshotManager.capture(new SimpleReplicatedLogEntry(9, 6, new MockRaftActorContext.MockPayload()), -1);
 541
 542         snapshotManager.persist(ByteState.empty(), Optional.empty(), Runtime.getRuntime().totalMemory());
 543
 544         snapshotManager.rollback();
 545
 546         snapshotManager.rollback();
 547
 548         verify(mockReplicatedLog, times(1)).snapshotRollback();
 549     }
 550
 551     @Test
 552     public void testTrimLogWhenTrimIndexLessThanLastApplied() {
 553         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 554
 555         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 556         doReturn(true).when(mockReplicatedLog).isPresent(10);
 557         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(10);
 558         doReturn(5L).when(replicatedLogEntry).getTerm();
 559
 560         long retIndex = snapshotManager.trimLog(10);
 561         assertEquals("return index", 10L, retIndex);
 562
 563         verify(mockReplicatedLog).snapshotPreCommit(10, 5);
 564         verify(mockReplicatedLog).snapshotCommit();
 565
 566         verify(mockRaftActorBehavior, never()).setReplicatedToAllIndex(anyLong());
 567     }
 568
 569     @Test
 570     public void testTrimLogWhenLastAppliedNotSet() {
 571         doReturn(-1L).when(mockRaftActorContext).getLastApplied();
 572
 573         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 574         doReturn(true).when(mockReplicatedLog).isPresent(10);
 575         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(10);
 576         doReturn(5L).when(replicatedLogEntry).getTerm();
 577
 578         long retIndex = snapshotManager.trimLog(10);
 579         assertEquals("return index", -1L, retIndex);
 580
 581         verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
 582         verify(mockReplicatedLog, never()).snapshotCommit();
 583
 584         verify(mockRaftActorBehavior, never()).setReplicatedToAllIndex(anyLong());
 585     }
 586
 587     @Test
 588     public void testTrimLogWhenLastAppliedZero() {
 589         doReturn(0L).when(mockRaftActorContext).getLastApplied();
 590
 591         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 592         doReturn(true).when(mockReplicatedLog).isPresent(10);
 593         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(10);
 594         doReturn(5L).when(replicatedLogEntry).getTerm();
 595
 596         long retIndex = snapshotManager.trimLog(10);
 597         assertEquals("return index", -1L, retIndex);
 598
 599         verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
 600         verify(mockReplicatedLog, never()).snapshotCommit();
 601
 602         verify(mockRaftActorBehavior, never()).setReplicatedToAllIndex(anyLong());
 603     }
 604
 605     @Test
 606     public void testTrimLogWhenTrimIndexNotPresent() {
 607         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 608
 609         doReturn(false).when(mockReplicatedLog).isPresent(10);
 610
 611         long retIndex = snapshotManager.trimLog(10);
 612         assertEquals("return index", -1L, retIndex);
 613
 614         verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
 615         verify(mockReplicatedLog, never()).snapshotCommit();
 616
 617         // Trim index is greater than replicatedToAllIndex so should update it.
 618         verify(mockRaftActorBehavior).setReplicatedToAllIndex(10L);
 619     }
 620
 621     @Test
 622     public void testTrimLogAfterCapture() {
 623         boolean capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 624                 new MockRaftActorContext.MockPayload()), 9);
 625
 626         assertTrue(capture);
 627
 628         assertEquals(true, snapshotManager.isCapturing());
 629
 630         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 631         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 632         doReturn(true).when(mockReplicatedLog).isPresent(10);
 633         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(10);
 634         doReturn(5L).when(replicatedLogEntry).getTerm();
 635
 636         snapshotManager.trimLog(10);
 637
 638         verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
 639         verify(mockReplicatedLog, never()).snapshotCommit();
 640
 641     }
 642
 643     @Test
 644     public void testTrimLogAfterCaptureToInstall() {
 645         boolean capture = snapshotManager.capture(new SimpleReplicatedLogEntry(9, 1,
 646                 new MockRaftActorContext.MockPayload()), 9);
 647
 648         assertTrue(capture);
 649
 650         assertEquals(true, snapshotManager.isCapturing());
 651
 652         ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
 653         doReturn(20L).when(mockRaftActorContext).getLastApplied();
 654         doReturn(true).when(mockReplicatedLog).isPresent(10);
 655         doReturn(replicatedLogEntry).when(mockReplicatedLog).get(10);
 656         doReturn(5L).when(replicatedLogEntry).getTerm();
 657
 658         snapshotManager.trimLog(10);
 659
 660         verify(mockReplicatedLog, never()).snapshotPreCommit(10, 5);
 661         verify(mockReplicatedLog, never()).snapshotCommit();
 662
 663     }
 664
 665     @Test
 666     public void testLastAppliedTermInformationReader() {
 667
 668         LastAppliedTermInformationReader reader = new LastAppliedTermInformationReader();
 669
 670         doReturn(4L).when(mockReplicatedLog).getSnapshotTerm();
 671         doReturn(7L).when(mockReplicatedLog).getSnapshotIndex();
 672
 673         ReplicatedLogEntry lastLogEntry = new SimpleReplicatedLogEntry(9L, 6L,
 674                 new MockRaftActorContext.MockPayload());
 675
 676         // No followers and valid lastLogEntry
 677         reader.init(mockReplicatedLog, 1L, lastLogEntry, false);
 678
 679         assertEquals("getTerm", 6L, reader.getTerm());
 680         assertEquals("getIndex", 9L, reader.getIndex());
 681
 682         // No followers and null lastLogEntry
 683         reader.init(mockReplicatedLog, 1L, null, false);
 684
 685         assertEquals("getTerm", -1L, reader.getTerm());
 686         assertEquals("getIndex", -1L, reader.getIndex());
 687
 688         // Followers and valid originalIndex entry
 689         doReturn(new SimpleReplicatedLogEntry(8L, 5L,
 690                 new MockRaftActorContext.MockPayload())).when(mockReplicatedLog).get(8L);
 691         reader.init(mockReplicatedLog, 8L, lastLogEntry, true);
 692
 693         assertEquals("getTerm", 5L, reader.getTerm());
 694         assertEquals("getIndex", 8L, reader.getIndex());
 695
 696         // Followers and null originalIndex entry and valid snapshot index
 697         reader.init(mockReplicatedLog, 7L, lastLogEntry, true);
 698
 699         assertEquals("getTerm", 4L, reader.getTerm());
 700         assertEquals("getIndex", 7L, reader.getIndex());
 701
 702         // Followers and null originalIndex entry and invalid snapshot index
 703         doReturn(-1L).when(mockReplicatedLog).getSnapshotIndex();
 704         reader.init(mockReplicatedLog, 7L, lastLogEntry, true);
 705
 706         assertEquals("getTerm", -1L, reader.getTerm());
 707         assertEquals("getIndex", -1L, reader.getIndex());
 708     }
 709 }