+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
package org.opendaylight.controller.cluster.raft;
import static org.junit.Assert.assertArrayEquals;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import com.google.common.collect.ImmutableMap;
import java.util.Arrays;
import java.util.HashMap;
-import java.util.List;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
doReturn(new HashMap<>()).when(mockRaftActorContext).getPeerAddresses();
doReturn(mockConfigParams).when(mockRaftActorContext).getConfigParams();
doReturn(10L).when(mockConfigParams).getSnapshotBatchCount();
+ doReturn(70).when(mockConfigParams).getSnapshotDataThresholdPercentage();
doReturn(mockReplicatedLog).when(mockRaftActorContext).getReplicatedLog();
doReturn("123").when(mockRaftActorContext).getId();
+ doReturn(mockDataPersistenceProvider).when(mockRaftActorContext).getPersistenceProvider();
doReturn("123").when(mockRaftActorBehavior).getLeaderId();
ElectionTerm mockElectionTerm = mock(ElectionTerm.class);
actorRef = factory.createTestActor(MessageCollectorActor.props(), factory.generateActorId("test-"));
doReturn(actorRef).when(mockRaftActorContext).getActor();
+ snapshotManager.setCreateSnapshotCallable(mockProcedure);
}
@After
}
@Test
- public void testCaptureToInstall(){
+ public void testCaptureToInstall() throws Exception {
// Force capturing toInstall = true
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
assertEquals(true, snapshotManager.isCapturing());
- CaptureSnapshot captureSnapshot = MessageCollectorActor.expectFirstMatching(actorRef, CaptureSnapshot.class);
+ verify(mockProcedure).apply(null);
+
+ CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
// LastIndex and LastTerm are picked up from the lastLogEntry
assertEquals(0L, captureSnapshot.getLastIndex());
}
@Test
- public void testCapture(){
+ public void testCapture() throws Exception {
boolean capture = snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
new MockRaftActorContext.MockPayload()), 9);
assertEquals(true, snapshotManager.isCapturing());
- CaptureSnapshot captureSnapshot = MessageCollectorActor.expectFirstMatching(actorRef, CaptureSnapshot.class);
+ verify(mockProcedure).apply(null);
+
+ CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
+
// LastIndex and LastTerm are picked up from the lastLogEntry
assertEquals(9L, captureSnapshot.getLastIndex());
assertEquals(1L, captureSnapshot.getLastTerm());
}
+ @Test
+ public void testCaptureWithNullLastLogEntry() throws Exception {
+ boolean capture = snapshotManager.capture(null, 1);
+
+ assertTrue(capture);
+
+ assertEquals(true, snapshotManager.isCapturing());
+
+ verify(mockProcedure).apply(null);
+
+ CaptureSnapshot captureSnapshot = snapshotManager.getCaptureSnapshot();
+
+ System.out.println(captureSnapshot);
+
+ // LastIndex and LastTerm are picked up from the lastLogEntry
+ assertEquals(-1L, captureSnapshot.getLastIndex());
+ assertEquals(-1L, captureSnapshot.getLastTerm());
+
+ // Since the actor does not have any followers (no peer addresses) lastApplied will be from lastLogEntry
+ assertEquals(-1L, captureSnapshot.getLastAppliedIndex());
+ assertEquals(-1L, captureSnapshot.getLastAppliedTerm());
+
+ //
+ assertEquals(-1L, captureSnapshot.getReplicatedToAllIndex());
+ assertEquals(-1L, captureSnapshot.getReplicatedToAllTerm());
+ actorRef.underlyingActor().clear();
+
+ }
+
+ @Test
+ public void testCaptureWithCreateProcedureError () throws Exception {
+ doThrow(new Exception("mock")).when(mockProcedure).apply(null);
+
+ boolean capture = snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
+ new MockRaftActorContext.MockPayload()), 9);
+
+ assertFalse(capture);
+
+ assertEquals(false, snapshotManager.isCapturing());
+
+ verify(mockProcedure).apply(null);
+ }
+
@Test
public void testIllegalCapture() throws Exception {
boolean capture = snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
assertTrue(capture);
- List<CaptureSnapshot> allMatching = MessageCollectorActor.getAllMatching(actorRef, CaptureSnapshot.class);
+ verify(mockProcedure).apply(null);
- assertEquals(1, allMatching.size());
+ reset(mockProcedure);
// This will not cause snapshot capture to start again
capture = snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(1,9,
assertFalse(capture);
- allMatching = MessageCollectorActor.getAllMatching(actorRef, CaptureSnapshot.class);
-
- assertEquals(1, allMatching.size());
+ verify(mockProcedure, never()).apply(null);
}
@Test
// when replicatedToAllIndex = -1
snapshotManager.capture(lastLogEntry, -1);
- snapshotManager.create(mockProcedure);
-
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(mockDataPersistenceProvider, bytes, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
verify(mockReplicatedLog).snapshotPreCommit(7L, 1L);
}
-
- @Test
- public void testCreate() throws Exception {
- // when replicatedToAllIndex = -1
- snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
- new MockRaftActorContext.MockPayload()), -1);
-
- snapshotManager.create(mockProcedure);
-
- verify(mockProcedure).apply(null);
-
- assertEquals("isCapturing", true, snapshotManager.isCapturing());
- }
-
- @Test
- public void testCallingCreateMultipleTimesCausesNoHarm() throws Exception {
- // when replicatedToAllIndex = -1
- snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
- new MockRaftActorContext.MockPayload()), -1);
-
- snapshotManager.create(mockProcedure);
-
- snapshotManager.create(mockProcedure);
-
- verify(mockProcedure, times(1)).apply(null);
- }
-
- @Test
- public void testCallingCreateBeforeCapture() throws Exception {
- snapshotManager.create(mockProcedure);
-
- verify(mockProcedure, times(0)).apply(null);
- }
-
- @Test
- public void testCallingCreateAfterPersist() throws Exception {
- // when replicatedToAllIndex = -1
- snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
- new MockRaftActorContext.MockPayload()), -1);
-
- snapshotManager.create(mockProcedure);
-
- verify(mockProcedure, times(1)).apply(null);
-
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
-
- reset(mockProcedure);
-
- snapshotManager.create(mockProcedure);
-
- verify(mockProcedure, never()).apply(null);
- }
-
@Test
public void testPersistWhenReplicatedToAllIndexNotMinus(){
doReturn(45L).when(mockReplicatedLog).getSnapshotIndex();
snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
new MockRaftActorContext.MockPayload()), 9);
- snapshotManager.create(mockProcedure);
-
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(mockDataPersistenceProvider, bytes, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
ArgumentCaptor<Snapshot> snapshotArgumentCaptor = ArgumentCaptor.forClass(Snapshot.class);
verify(mockDataPersistenceProvider).saveSnapshot(snapshotArgumentCaptor.capture());
verify(mockRaftActorBehavior).setReplicatedToAllIndex(9);
}
-
@Test
public void testPersistWhenReplicatedLogDataSizeGreaterThanThreshold(){
doReturn(Integer.MAX_VALUE).when(mockReplicatedLog).dataSize();
snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6,9,
new MockRaftActorContext.MockPayload()), -1);
- snapshotManager.create(mockProcedure);
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
+
+ verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
+
+ verify(mockRaftActorBehavior, never()).setReplicatedToAllIndex(anyLong());
+ }
+
+ @Test
+ public void testPersistWhenReplicatedLogSizeExceedsSnapshotBatchCount() {
+ doReturn(10L).when(mockReplicatedLog).size(); // matches snapshotBatchCount
+ doReturn(100).when(mockReplicatedLog).dataSize();
+
+ doReturn(5L).when(mockReplicatedLog).getSnapshotIndex();
+ doReturn(5L).when(mockReplicatedLog).getSnapshotTerm();
+
+ long replicatedToAllIndex = 1;
+ ReplicatedLogEntry replicatedLogEntry = mock(ReplicatedLogEntry.class);
+ doReturn(replicatedLogEntry).when(mockReplicatedLog).get(replicatedToAllIndex);
+ doReturn(6L).when(replicatedLogEntry).getTerm();
+ doReturn(replicatedToAllIndex).when(replicatedLogEntry).getIndex();
+
+ snapshotManager.capture(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
+ new MockRaftActorContext.MockPayload()), replicatedToAllIndex);
+
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, 2000000L);
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
verify(mockReplicatedLog).snapshotPreCommit(9L, 6L);
+
+ verify(mockRaftActorBehavior).setReplicatedToAllIndex(replicatedToAllIndex);
}
@Test
assertTrue(capture);
- snapshotManager.create(mockProcedure);
-
byte[] bytes = new byte[] {1,2,3,4,5,6,7,8,9,10};
- snapshotManager.persist(mockDataPersistenceProvider, bytes, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(bytes, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
+
+ assertEquals(true, snapshotManager.isCapturing());
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
SendInstallSnapshot sendInstallSnapshot = sendInstallSnapshotArgumentCaptor.getValue();
- assertTrue(Arrays.equals(bytes, sendInstallSnapshot.getSnapshot().toByteArray()));
+ assertTrue(Arrays.equals(bytes, sendInstallSnapshot.getSnapshot().getState()));
}
@Test
public void testCallingPersistWithoutCaptureWillDoNothing(){
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
verify(mockDataPersistenceProvider, never()).saveSnapshot(any(Snapshot.class));
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.create(mockProcedure);
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
-
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
verify(mockDataPersistenceProvider).saveSnapshot(any(Snapshot.class));
@Test
public void testCommit(){
+ doReturn(50L).when(mockDataPersistenceProvider).getLastSequenceNumber();
+
// when replicatedToAllIndex = -1
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.create(mockProcedure);
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ assertEquals(true, snapshotManager.isCapturing());
- snapshotManager.commit(mockDataPersistenceProvider, 100L);
+ snapshotManager.commit(100L, mockRaftActorBehavior);
+
+ assertEquals(false, snapshotManager.isCapturing());
verify(mockReplicatedLog).snapshotCommit();
- verify(mockDataPersistenceProvider).deleteMessages(100L);
+ verify(mockDataPersistenceProvider).deleteMessages(50L);
ArgumentCaptor<SnapshotSelectionCriteria> criteriaCaptor = ArgumentCaptor.forClass(SnapshotSelectionCriteria.class);
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.commit(mockDataPersistenceProvider, 100L);
+ snapshotManager.commit(100L, mockRaftActorBehavior);
verify(mockReplicatedLog, never()).snapshotCommit();
@Test
public void testCommitBeforeCapture(){
- snapshotManager.commit(mockDataPersistenceProvider, 100L);
+ snapshotManager.commit(100L, mockRaftActorBehavior);
verify(mockReplicatedLog, never()).snapshotCommit();
@Test
public void testCallingCommitMultipleTimesCausesNoHarm(){
+ doReturn(50L).when(mockDataPersistenceProvider).getLastSequenceNumber();
+
// when replicatedToAllIndex = -1
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.create(mockProcedure);
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.commit(100L, mockRaftActorBehavior);
- snapshotManager.commit(mockDataPersistenceProvider, 100L);
-
- snapshotManager.commit(mockDataPersistenceProvider, 100L);
+ snapshotManager.commit(100L, mockRaftActorBehavior);
verify(mockReplicatedLog, times(1)).snapshotCommit();
- verify(mockDataPersistenceProvider, times(1)).deleteMessages(100L);
+ verify(mockDataPersistenceProvider, times(1)).deleteMessages(50L);
verify(mockDataPersistenceProvider, times(1)).deleteSnapshots(any(SnapshotSelectionCriteria.class));
}
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.create(mockProcedure);
-
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
snapshotManager.rollback();
snapshotManager.captureToInstall(new MockRaftActorContext.MockReplicatedLogEntry(6, 9,
new MockRaftActorContext.MockPayload()), -1, "follower-1");
- snapshotManager.create(mockProcedure);
-
- snapshotManager.persist(mockDataPersistenceProvider, new byte[]{}, mockRaftActorBehavior
- , Runtime.getRuntime().totalMemory());
+ snapshotManager.persist(new byte[]{}, mockRaftActorBehavior, Runtime.getRuntime().totalMemory());
snapshotManager.rollback();
assertEquals("getTerm", -1L, reader.getTerm());
assertEquals("getIndex", -1L, reader.getIndex());
}
-}
\ No newline at end of file
+}