- @Test
- public void testHandleRecoveryWhenDataPersistenceRecoveryApplicable() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- // Wait for akka's recovery to complete so it doesn't interfere.
- mockRaftActor.waitForRecoveryComplete();
-
- ByteString snapshotBytes = fromObject(Arrays.asList(
- new MockRaftActorContext.MockPayload("A"),
- new MockRaftActorContext.MockPayload("B"),
- new MockRaftActorContext.MockPayload("C"),
- new MockRaftActorContext.MockPayload("D")));
-
- Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
- Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
-
- mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
-
- verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes.toByteArray()));
-
- mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
-
- ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
-
- assertEquals("add replicated log entry", 1, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
-
- assertEquals("add replicated log entry", 2, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new ApplyLogEntries(1));
-
- assertEquals("commit index 1", 1, mockRaftActor.getRaftActorContext().getCommitIndex());
-
- // The snapshot had 4 items + we added 2 more items during the test
- // We start removing from 5 and we should get 1 item in the replicated log
- mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(5));
-
- assertEquals("remove log entries", 1, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
-
- assertEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
- assertEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
-
- mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
-
- }};
- }
-
- /**
- * This test verifies that when recovery is not applicable (typically when persistence is false) the RaftActor does
- * not process recovery messages
- *
- * @throws Exception
- */
- @Test
- public void testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), new DataPersistenceProviderMonitor()), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- // Wait for akka's recovery to complete so it doesn't interfere.
- mockRaftActor.waitForRecoveryComplete();
-
- ByteString snapshotBytes = fromObject(Arrays.asList(
- new MockRaftActorContext.MockPayload("A"),
- new MockRaftActorContext.MockPayload("B"),
- new MockRaftActorContext.MockPayload("C"),
- new MockRaftActorContext.MockPayload("D")));
-
- Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
- Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
-
- mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
-
- verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(byte[].class));
-
- mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
-
- ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
-
- assertEquals("add replicated log entry", 0, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
-
- assertEquals("add replicated log entry", 0, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new ApplyLogEntries(1));
-
- assertEquals("commit index -1", -1, mockRaftActor.getRaftActorContext().getCommitIndex());
-
- mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(2));
-
- assertEquals("remove log entries", 0, replicatedLog.size());
-
- mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
-
- assertNotEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
- assertNotEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
-
- mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
- }};
- }
-
-
- @Test
- public void testUpdatingElectionTermCallsDataPersistence() throws Exception {
- new JavaTestKit(getSystem()) {
- {
- String persistenceId = factory.generateActorId("leader-");
-
- DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
-
- config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
-
- CountDownLatch persistLatch = new CountDownLatch(1);
- DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
- dataPersistenceProviderMonitor.setPersistLatch(persistLatch);
-
- TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
- Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
-
- MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
-
- mockRaftActor.waitForInitializeBehaviorComplete();
-
- mockRaftActor.getRaftActorContext().getTermInformation().updateAndPersist(10, "foobar");