+ DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
+
+ TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
+ Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+
+ MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
+
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,0, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,1, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,2, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,3, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,4, mock(Payload.class)));
+
+ ByteString snapshotBytes = fromObject(Arrays.asList(
+ new MockRaftActorContext.MockPayload("A"),
+ new MockRaftActorContext.MockPayload("B"),
+ new MockRaftActorContext.MockPayload("C"),
+ new MockRaftActorContext.MockPayload("D")));
+
+ RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
+ mockRaftActor.setCurrentBehavior(new Follower(raftActorContext));
+
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1));
+
+ verify(mockRaftActor.delegate).createSnapshot();
+
+ mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
+
+ mockRaftActor.onReceiveCommand(new SaveSnapshotSuccess(new SnapshotMetadata("foo", 100, 100)));
+
+ verify(dataPersistenceProvider).deleteSnapshots(any(SnapshotSelectionCriteria.class));
+
+ verify(dataPersistenceProvider).deleteMessages(100);
+
+ assertEquals(2, mockRaftActor.getReplicatedLog().size());
+
+ assertNotNull(mockRaftActor.getReplicatedLog().get(3));
+ assertNotNull(mockRaftActor.getReplicatedLog().get(4));
+
+ // Index 2 will not be in the log because it was removed due to snapshotting
+ assertNull(mockRaftActor.getReplicatedLog().get(2));
+
+ mockActorRef.tell(PoisonPill.getInstance(), getRef());
+
+ }
+ };
+ }
+
+ @Test
+ public void testApplyState() throws Exception {
+
+ new JavaTestKit(getSystem()) {
+ {
+ String persistenceId = "testApplyState";
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+
+ DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
+
+ TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
+ Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+
+ MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
+
+ ReplicatedLogEntry entry = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
+ new MockRaftActorContext.MockPayload("F"));
+
+ mockRaftActor.onReceiveCommand(new ApplyState(mockActorRef, "apply-state", entry));
+
+ verify(mockRaftActor.delegate).applyState(eq(mockActorRef), eq("apply-state"), anyObject());
+
+ mockActorRef.tell(PoisonPill.getInstance(), getRef());
+
+ }
+ };
+ }
+
+ @Test
+ public void testApplySnapshot() throws Exception {
+ new JavaTestKit(getSystem()) {
+ {
+ String persistenceId = "testApplySnapshot";
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+
+ DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
+
+ TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
+ Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
+
+ MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
+
+ ReplicatedLog oldReplicatedLog = mockRaftActor.getReplicatedLog();
+
+ oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,0,mock(Payload.class)));
+ oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,1,mock(Payload.class)));
+ oldReplicatedLog.append(
+ new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
+ mock(Payload.class)));
+
+ ByteString snapshotBytes = fromObject(Arrays.asList(
+ new MockRaftActorContext.MockPayload("A"),
+ new MockRaftActorContext.MockPayload("B"),
+ new MockRaftActorContext.MockPayload("C"),
+ new MockRaftActorContext.MockPayload("D")));
+
+ Snapshot snapshot = mock(Snapshot.class);
+
+ doReturn(snapshotBytes.toByteArray()).when(snapshot).getState();
+
+ doReturn(3L).when(snapshot).getLastAppliedIndex();
+
+ mockRaftActor.onReceiveCommand(new ApplySnapshot(snapshot));
+
+ verify(mockRaftActor.delegate).applySnapshot(eq(snapshotBytes));
+
+ assertTrue("The replicatedLog should have changed",
+ oldReplicatedLog != mockRaftActor.getReplicatedLog());
+
+ assertEquals("lastApplied should be same as in the snapshot",
+ (Long) 3L, mockRaftActor.getLastApplied());
+
+ assertEquals(0, mockRaftActor.getReplicatedLog().size());
+
+ mockActorRef.tell(PoisonPill.getInstance(), getRef());
+
+ }
+ };
+ }
+
+ @Test
+ public void testSaveSnapshotFailure() throws Exception {
+ new JavaTestKit(getSystem()) {
+ {
+ String persistenceId = "testSaveSnapshotFailure";
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+