import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyLong;
-import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import akka.actor.ActorRef;
-import akka.japi.Procedure;
import akka.persistence.SnapshotSelectionCriteria;
import akka.testkit.TestActorRef;
import java.util.Arrays;
private RaftActorBehavior mockRaftActorBehavior;
@Mock
- private Procedure<Void> mockProcedure;
+ private Runnable mockProcedure;
@Mock
private ElectionTerm mockElectionTerm;
private TestActorRef<MessageCollectorActor> actorRef;
@Before
- public void setUp() throws Exception {
+ public void setUp(){
MockitoAnnotations.initMocks(this);
doReturn(false).when(mockRaftActorContext).hasFollowers();
doReturn(mockReplicatedLog).when(mockRaftActorContext).getReplicatedLog();
doReturn("123").when(mockRaftActorContext).getId();
doReturn(mockDataPersistenceProvider).when(mockRaftActorContext).getPersistenceProvider();
+ doReturn(mockRaftActorBehavior).when(mockRaftActorContext).getCurrentBehavior();
doReturn("123").when(mockRaftActorBehavior).getLeaderId();
- doReturn(null).when(mockRaftActorBehavior).handleMessage(any(ActorRef.class), any(Object.class));
- doReturn(0L).when(mockRaftActorBehavior).getReplicatedToAllIndex();
doReturn(mockElectionTerm).when(mockRaftActorContext).getTermInformation();
doReturn(5L).when(mockElectionTerm).getCurrentTerm();
actorRef = factory.createTestActor(MessageCollectorActor.props(), factory.generateActorId("test-"));
doReturn(actorRef).when(mockRaftActorContext).getActor();
- doReturn(0L).when(mockRaftActorContext).getLastApplied();
- doReturn(null).when(mockRaftActorContext).getPeerServerInfo(any(boolean.class));
- doReturn(null).when(mockReplicatedLog).get(any(long.class));
- doReturn(null).when(mockReplicatedLog).getFrom(any(long.class));
- doReturn(0).when(mockReplicatedLog).dataSize();
- doReturn(0L).when(mockReplicatedLog).size();
- doReturn(0L).when(mockReplicatedLog).getSnapshotIndex();
- doReturn(0L).when(mockReplicatedLog).getSnapshotTerm();
- doNothing().when(mockReplicatedLog).snapshotPreCommit(any(long.class), any(long.class));
- doNothing().when(mockReplicatedLog).snapshotRollback();
- doNothing().when(mockReplicatedLog).snapshotCommit();
- doNothing().when(mockRaftActorBehavior).setReplicatedToAllIndex(any(long.class));
-
- doReturn(0L).when(mockDataPersistenceProvider).getLastSequenceNumber();
- doNothing().when(mockDataPersistenceProvider).saveSnapshot(any(Object.class));
- doNothing().when(mockDataPersistenceProvider).persist(any(Object.class), any(Procedure.class));
- doNothing().when(mockDataPersistenceProvider).deleteSnapshots(any(SnapshotSelectionCriteria.class));
- doNothing().when(mockDataPersistenceProvider).deleteMessages(any(long.class));
-
- doNothing().when(mockProcedure).apply(null);
-
- snapshotManager.setCreateSnapshotCallable(mockProcedure);
+
+ snapshotManager.setCreateSnapshotRunnable(mockProcedure);
}
@After
assertEquals(true, snapshotManager.isCapturing());
- verify(mockProcedure).apply(null);
+ verify(mockProcedure).run();
CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
assertEquals(true, snapshotManager.isCapturing());
- verify(mockProcedure).apply(null);
+ verify(mockProcedure).run();
CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
assertEquals(true, snapshotManager.isCapturing());
- verify(mockProcedure).apply(null);
+ verify(mockProcedure).run();
CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
@Test
public void testCaptureWithCreateProcedureError () throws Exception {
- doThrow(new Exception("mock")).when(mockProcedure).apply(null);
+ doThrow(new RuntimeException("mock")).when(mockProcedure).run();
boolean capture = snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
new MockRaftActorContext.MockPayload()), 9);
assertEquals(false, snapshotManager.isCapturing());
- verify(mockProcedure).apply(null);
+ verify(mockProcedure).run();
}
@Test
assertTrue(capture);
- verify(mockProcedure).apply(null);
+ verify(mockProcedure).run();
reset(mockProcedure);
assertFalse(capture);
- verify(mockProcedure, never()).apply(null);
+ verify(mockProcedure, never()).run();
}
@Test
snapshotManager.capture(lastLogEntry, -1);
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, Runtime.getRuntime().totalMemory());
ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
new MockRaftActorContext.MockPayload()), 9);
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, Runtime.getRuntime().totalMemory());
ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
new MockRaftActorContext.MockPayload()), -1);
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), replicatedToAllIndex);
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, 2000000L);
+ snapshotManager.persist(new byte[]{}, 2000000L);
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, Runtime.getRuntime().totalMemory());
assertEquals(true, snapshotManager.isCapturing());
@Test
public void testCallingPersistWithoutCaptureWillDoNothing(){
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
verify(mockDataPersistenceProvider, never()).saveSnapshot(any(Snapshot.class));
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
assertEquals(true, snapshotManager.isCapturing());
- snapshotManager.commit(100L, mockRaftActorBehavior);
+ snapshotManager.commit(100L, 1234L);
assertEquals(false, snapshotManager.isCapturing());
verify(mockDataPersistenceProvider).deleteSnapshots(criteriaCaptor.capture());
- assertEquals(90, criteriaCaptor.getValue().maxSequenceNr()); // sequenceNumber = 100
- // config snapShotBatchCount = 10
- // therefore maxSequenceNumber = 90
+ assertEquals(100L, criteriaCaptor.getValue().maxSequenceNr());
+ assertEquals(1233L, criteriaCaptor.getValue().maxTimestamp());
MessageCollectorActor.expectFirstMatching(actorRef, SnapshotComplete.class);
}
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.commit(100L, mockRaftActorBehavior);
+ snapshotManager.commit(100L, 0);
verify(mockReplicatedLog, never()).snapshotCommit();
@Test
public void testCommitBeforeCapture(){
- snapshotManager.commit(100L, mockRaftActorBehavior);
+ snapshotManager.commit(100L, 0);
verify(mockReplicatedLog, never()).snapshotCommit();
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
- snapshotManager.commit(100L, mockRaftActorBehavior);
+ snapshotManager.commit(100L, 0);
- snapshotManager.commit(100L, mockRaftActorBehavior);
+ snapshotManager.commit(100L, 0);
verify(mockReplicatedLog, times(1)).snapshotCommit();
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
snapshotManager.rollback();
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, Runtime.getRuntime().totalMemory());
snapshotManager.rollback();
doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
doReturn(5L).when(replicatedLogEntry).getTerm();
- long retIndex = snapshotManager.trimLog(10, mockRaftActorBehavior);
+ long retIndex = snapshotManager.trimLog(10);
assertEquals("return index", 10L, retIndex);
verify(mockReplicatedLog).snapshotPreCommit(10, 5);
doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
doReturn(5L).when(replicatedLogEntry).getTerm();
- long retIndex = snapshotManager.trimLog(10, mockRaftActorBehavior);
+ long retIndex = snapshotManager.trimLog(10);
assertEquals("return index", -1L, retIndex);
verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
doReturn(5L).when(replicatedLogEntry).getTerm();
- long retIndex = snapshotManager.trimLog(10, mockRaftActorBehavior);
+ long retIndex = snapshotManager.trimLog(10);
assertEquals("return index", -1L, retIndex);
verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
doReturn(false).when(mockReplicatedLog).isPresent(10);
- long retIndex = snapshotManager.trimLog(10, mockRaftActorBehavior);
+ long retIndex = snapshotManager.trimLog(10);
assertEquals("return index", -1L, retIndex);
verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
doReturn(5L).when(replicatedLogEntry).getTerm();
- snapshotManager.trimLog(10, mockRaftActorBehavior);
+ snapshotManager.trimLog(10);
verify(mockReplicatedLog, never()).snapshotPreCommit(anyLong(), anyLong());
verify(mockReplicatedLog, never()).snapshotCommit();
doReturn(replicatedLogEntry).when((mockReplicatedLog)).get(10);
doReturn(5L).when(replicatedLogEntry).getTerm();
- snapshotManager.trimLog(10, mockRaftActorBehavior);
+ snapshotManager.trimLog(10);
verify(mockReplicatedLog, never()).snapshotPreCommit(10, 5);
verify(mockReplicatedLog, never()).snapshotCommit();