+ @Test
+ public void testUpdateConfigParam() throws Exception {
+ DefaultConfigParamsImpl emptyConfig = new DefaultConfigParamsImpl();
+ String persistenceId = factory.generateActorId("follower-");
+ ImmutableMap<String, String> peerAddresses =
+ ImmutableMap.<String, String>builder().put("member1", "address").build();
+ DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
+
+ TestActorRef<MockRaftActor> actorRef = factory.createTestActor(
+ MockRaftActor.props(persistenceId, peerAddresses, emptyConfig, dataPersistenceProvider), persistenceId);
+ MockRaftActor mockRaftActor = actorRef.underlyingActor();
+ mockRaftActor.waitForInitializeBehaviorComplete();
+
+ RaftActorBehavior behavior = mockRaftActor.getCurrentBehavior();
+ mockRaftActor.updateConfigParams(emptyConfig);
+ assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
+ assertEquals("Behavior State", RaftState.Follower,
+ mockRaftActor.getCurrentBehavior().state());
+
+ DefaultConfigParamsImpl disableConfig = new DefaultConfigParamsImpl();
+ disableConfig.setCustomRaftPolicyImplementationClass(
+ "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
+ mockRaftActor.updateConfigParams(disableConfig);
+ assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior());
+ assertEquals("Behavior State", RaftState.Follower,
+ mockRaftActor.getCurrentBehavior().state());
+
+ behavior = mockRaftActor.getCurrentBehavior();
+ mockRaftActor.updateConfigParams(disableConfig);
+ assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
+ assertEquals("Behavior State", RaftState.Follower,
+ mockRaftActor.getCurrentBehavior().state());
+
+ DefaultConfigParamsImpl defaultConfig = new DefaultConfigParamsImpl();
+ defaultConfig.setCustomRaftPolicyImplementationClass(
+ "org.opendaylight.controller.cluster.raft.policy.DefaultRaftPolicy");
+ mockRaftActor.updateConfigParams(defaultConfig);
+ assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior());
+ assertEquals("Behavior State", RaftState.Follower,
+ mockRaftActor.getCurrentBehavior().state());
+
+ behavior = mockRaftActor.getCurrentBehavior();
+ mockRaftActor.updateConfigParams(defaultConfig);
+ assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
+ assertEquals("Behavior State", RaftState.Follower,
+ mockRaftActor.getCurrentBehavior().state());
+ }
+
+ @Test
+ public void testGetSnapshot() throws Exception {
+ TEST_LOG.info("testGetSnapshot starting");
+
+ JavaTestKit kit = new JavaTestKit(getSystem());
+
+ String persistenceId = factory.generateActorId("test-actor-");
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ long term = 3;
+ long seqN = 1;
+ InMemoryJournal.addEntry(persistenceId, seqN++, new UpdateElectionTerm(term, "member-1"));
+ InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 0,
+ new MockRaftActorContext.MockPayload("A")));
+ InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 1,
+ new MockRaftActorContext.MockPayload("B")));
+ InMemoryJournal.addEntry(persistenceId, seqN++, new ApplyJournalEntries(1));
+ InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 2,
+ new MockRaftActorContext.MockPayload("C")));
+
+ TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ ImmutableMap.<String, String>builder().put("member1", "address").build(), config).
+ withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForRecoveryComplete();
+
+ // Wait for snapshot after recovery
+ verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(any(ActorRef.class));
+
+ mockRaftActor.snapshotCohortDelegate = mock(RaftActorSnapshotCohort.class);
+
+ raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
+
+ ArgumentCaptor<ActorRef> replyActor = ArgumentCaptor.forClass(ActorRef.class);
+ verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(replyActor.capture());
+
+ byte[] stateSnapshot = new byte[]{1,2,3};
+ replyActor.getValue().tell(new CaptureSnapshotReply(stateSnapshot), ActorRef.noSender());
+
+ GetSnapshotReply reply = kit.expectMsgClass(GetSnapshotReply.class);
+
+ assertEquals("getId", persistenceId, reply.getId());
+ Snapshot replySnapshot = SerializationUtils.deserialize(reply.getSnapshot());
+ assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor());
+ assertEquals("getLastAppliedIndex", 1L, replySnapshot.getLastAppliedIndex());
+ assertEquals("getLastAppliedTerm", term, replySnapshot.getLastAppliedTerm());
+ assertEquals("getLastIndex", 2L, replySnapshot.getLastIndex());
+ assertEquals("getLastTerm", term, replySnapshot.getLastTerm());
+ assertArrayEquals("getState", stateSnapshot, replySnapshot.getState());
+ assertEquals("getUnAppliedEntries size", 1, replySnapshot.getUnAppliedEntries().size());
+ assertEquals("UnApplied entry index ", 2L, replySnapshot.getUnAppliedEntries().get(0).getIndex());
+
+ // Test with timeout
+
+ mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(Duration.create(200, TimeUnit.MILLISECONDS));
+ reset(mockRaftActor.snapshotCohortDelegate);
+
+ raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
+ Failure failure = kit.expectMsgClass(akka.actor.Status.Failure.class);
+ assertEquals("Failure cause type", TimeoutException.class, failure.cause().getClass());
+
+ mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(Duration.create(30, TimeUnit.SECONDS));
+
+ // Test with persistence disabled.
+
+ mockRaftActor.setPersistence(false);
+ reset(mockRaftActor.snapshotCohortDelegate);
+
+ raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
+ reply = kit.expectMsgClass(GetSnapshotReply.class);
+ verify(mockRaftActor.snapshotCohortDelegate, never()).createSnapshot(any(ActorRef.class));
+
+ assertEquals("getId", persistenceId, reply.getId());
+ replySnapshot = SerializationUtils.deserialize(reply.getSnapshot());
+ assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor());
+ assertEquals("getLastAppliedIndex", -1L, replySnapshot.getLastAppliedIndex());
+ assertEquals("getLastAppliedTerm", -1L, replySnapshot.getLastAppliedTerm());
+ assertEquals("getLastIndex", -1L, replySnapshot.getLastIndex());
+ assertEquals("getLastTerm", -1L, replySnapshot.getLastTerm());
+ assertEquals("getState length", 0, replySnapshot.getState().length);
+ assertEquals("getUnAppliedEntries size", 0, replySnapshot.getUnAppliedEntries().size());
+
+ TEST_LOG.info("testGetSnapshot ending");
+ }
+
+ @Test
+ public void testRestoreFromSnapshot() throws Exception {
+ TEST_LOG.info("testRestoreFromSnapshot starting");
+
+ String persistenceId = factory.generateActorId("test-actor-");
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ List<ReplicatedLogEntry> snapshotUnappliedEntries = new ArrayList<>();
+ snapshotUnappliedEntries.add(new MockRaftActorContext.MockReplicatedLogEntry(1, 4,
+ new MockRaftActorContext.MockPayload("E")));
+
+ int snapshotLastApplied = 3;
+ int snapshotLastIndex = 4;
+
+ List<MockPayload> state = Arrays.asList(
+ new MockRaftActorContext.MockPayload("A"),
+ new MockRaftActorContext.MockPayload("B"),
+ new MockRaftActorContext.MockPayload("C"),
+ new MockRaftActorContext.MockPayload("D"));
+ ByteString stateBytes = fromObject(state);
+
+ Snapshot snapshot = Snapshot.create(stateBytes.toByteArray(), snapshotUnappliedEntries,
+ snapshotLastIndex, 1, snapshotLastApplied, 1, 1, "member-1");
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(persistenceId);
+
+ TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId).
+ config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot)).props().
+ withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForRecoveryComplete();
+
+ Snapshot savedSnapshot = InMemorySnapshotStore.waitForSavedSnapshot(persistenceId, Snapshot.class);
+ assertEquals("getElectionTerm", snapshot.getElectionTerm(), savedSnapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", snapshot.getElectionVotedFor(), savedSnapshot.getElectionVotedFor());
+ assertEquals("getLastAppliedIndex", snapshot.getLastAppliedIndex(), savedSnapshot.getLastAppliedIndex());
+ assertEquals("getLastAppliedTerm", snapshot.getLastAppliedTerm(), savedSnapshot.getLastAppliedTerm());
+ assertEquals("getLastIndex", snapshot.getLastIndex(), savedSnapshot.getLastIndex());
+ assertEquals("getLastTerm", snapshot.getLastTerm(), savedSnapshot.getLastTerm());
+ assertArrayEquals("getState", snapshot.getState(), savedSnapshot.getState());
+ assertEquals("getUnAppliedEntries", snapshot.getUnAppliedEntries(), savedSnapshot.getUnAppliedEntries());
+
+ verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).applySnapshot(any(byte[].class));
+
+ RaftActorContext context = mockRaftActor.getRaftActorContext();
+ assertEquals("Journal log size", 1, context.getReplicatedLog().size());
+ assertEquals("Last index", snapshotLastIndex, context.getReplicatedLog().lastIndex());
+ assertEquals("Last applied", snapshotLastApplied, context.getLastApplied());
+ assertEquals("Commit index", snapshotLastApplied, context.getCommitIndex());
+ assertEquals("Recovered state", state, mockRaftActor.getState());
+ assertEquals("Current term", 1L, context.getTermInformation().getCurrentTerm());
+ assertEquals("Voted for", "member-1", context.getTermInformation().getVotedFor());
+
+ // Test with data persistence disabled
+
+ snapshot = Snapshot.create(new byte[0], Collections.<ReplicatedLogEntry>emptyList(),
+ -1, -1, -1, -1, 5, "member-1");
+
+ persistenceId = factory.generateActorId("test-actor-");
+
+ raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId).
+ config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot)).
+ persistent(Optional.of(Boolean.FALSE)).props().
+ withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForRecoveryComplete();
+ assertEquals("snapshot committed", true,
+ Uninterruptibles.awaitUninterruptibly(mockRaftActor.snapshotCommitted, 5, TimeUnit.SECONDS));
+
+ context = mockRaftActor.getRaftActorContext();
+ assertEquals("Current term", 5L, context.getTermInformation().getCurrentTerm());
+ assertEquals("Voted for", "member-1", context.getTermInformation().getVotedFor());
+
+ TEST_LOG.info("testRestoreFromSnapshot ending");
+ }
+
+ @Test
+ public void testRestoreFromSnapshotWithRecoveredData() throws Exception {
+ TEST_LOG.info("testRestoreFromSnapshotWithRecoveredData starting");
+
+ String persistenceId = factory.generateActorId("test-actor-");
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ List<MockPayload> state = Arrays.asList(new MockRaftActorContext.MockPayload("A"));
+ Snapshot snapshot = Snapshot.create(fromObject(state).toByteArray(), Arrays.<ReplicatedLogEntry>asList(),
+ 5, 2, 5, 2, 2, "member-1");
+
+ InMemoryJournal.addEntry(persistenceId, 1, new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
+ new MockRaftActorContext.MockPayload("B")));
+
+ TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId).
+ config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot)).props().
+ withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForRecoveryComplete();
+
+ Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
+ verify(mockRaftActor.snapshotCohortDelegate, never()).applySnapshot(any(byte[].class));
+
+ RaftActorContext context = mockRaftActor.getRaftActorContext();
+ assertEquals("Journal log size", 1, context.getReplicatedLog().size());
+ assertEquals("Last index", 0, context.getReplicatedLog().lastIndex());
+ assertEquals("Last applied", -1, context.getLastApplied());
+ assertEquals("Commit index", -1, context.getCommitIndex());
+ assertEquals("Current term", 0, context.getTermInformation().getCurrentTerm());
+ assertEquals("Voted for", null, context.getTermInformation().getVotedFor());
+
+ TEST_LOG.info("testRestoreFromSnapshotWithRecoveredData ending");
+ }
+
+ @Test
+ public void testNonVotingOnRecovery() throws Exception {
+ TEST_LOG.info("testNonVotingOnRecovery starting");
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setElectionTimeoutFactor(1);
+ config.setHeartBeatInterval(FiniteDuration.create(1, TimeUnit.MILLISECONDS));
+
+ String persistenceId = factory.generateActorId("test-actor-");
+ InMemoryJournal.addEntry(persistenceId, 1, new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
+ new ServerConfigurationPayload(Arrays.asList(new ServerInfo(persistenceId, false)))));
+
+ TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId).
+ config(config).props().withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForInitializeBehaviorComplete();
+
+ // Sleep a bit and verify it didn't get an election timeout and schedule an election.
+
+ Uninterruptibles.sleepUninterruptibly(400, TimeUnit.MILLISECONDS);
+ assertEquals("getRaftState", RaftState.Follower, mockRaftActor.getRaftState());
+
+ TEST_LOG.info("testNonVotingOnRecovery ending");
+ }
+
+ @Test
+ public void testLeaderTransitioning() throws Exception {
+ TEST_LOG.info("testLeaderTransitioning starting");
+
+ TestActorRef<MessageCollectorActor> notifierActor = factory.createTestActor(
+ Props.create(MessageCollectorActor.class));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ String persistenceId = factory.generateActorId("test-actor-");
+
+ TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId).
+ config(config).roleChangeNotifier(notifierActor).props().withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+ MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
+
+ mockRaftActor.waitForInitializeBehaviorComplete();
+
+ raftActorRef.tell(new AppendEntries(1L, "leader", 0L, 1L, Collections.<ReplicatedLogEntry>emptyList(),
+ 0L, -1L, (short)1), ActorRef.noSender());
+ LeaderStateChanged leaderStateChange = MessageCollectorActor.expectFirstMatching(
+ notifierActor, LeaderStateChanged.class);
+ assertEquals("getLeaderId", "leader", leaderStateChange.getLeaderId());
+
+ MessageCollectorActor.clearMessages(notifierActor);
+
+ raftActorRef.tell(LeaderTransitioning.INSTANCE, ActorRef.noSender());
+
+ leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
+ assertEquals("getMemberId", persistenceId, leaderStateChange.getMemberId());
+ assertEquals("getLeaderId", null, leaderStateChange.getLeaderId());
+
+ TEST_LOG.info("testLeaderTransitioning ending");
+ }