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