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