import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.same;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.timeout;
import akka.actor.ActorRef;
import akka.actor.PoisonPill;
import akka.actor.Props;
import akka.actor.Terminated;
+import akka.dispatch.Dispatchers;
import akka.japi.Procedure;
import akka.persistence.SaveSnapshotFailure;
import akka.persistence.SaveSnapshotSuccess;
import org.junit.Test;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.NonPersistentDataProvider;
+import org.opendaylight.controller.cluster.PersistentDataProvider;
import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
import org.opendaylight.controller.cluster.notifications.RoleChanged;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries;
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
+import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.Leader;
kit.waitUntilLeader();
}
+
@Test
public void testRaftActorRecoveryWithPersistenceEnabled() throws Exception {
TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled starting");
}};
}
+ @Test
+ public void testUpdateElectionTermPersistedWithPersistenceDisabled() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ String persistenceId = factory.generateActorId("follower-");
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
+ config.setElectionTimeoutFactor(1);
+
+ InMemoryJournal.addWriteMessagesCompleteLatch(persistenceId, 1);
+
+ TestActorRef<MockRaftActor> ref = factory.createTestActor(MockRaftActor.props(persistenceId,
+ ImmutableMap.<String, String>builder().put("member1", "address").build(),
+ Optional.<ConfigParams>of(config), new NonPersistentDataProvider()).
+ withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
+
+ InMemoryJournal.waitForWriteMessagesComplete(persistenceId);
+ List<UpdateElectionTerm> entries = InMemoryJournal.get(persistenceId, UpdateElectionTerm.class);
+ assertEquals("UpdateElectionTerm entries", 1, entries.size());
+ UpdateElectionTerm updateEntry = entries.get(0);
+
+ factory.killActor(ref, this);
+
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ ref = factory.createTestActor(MockRaftActor.props(persistenceId,
+ ImmutableMap.<String, String>builder().put("member1", "address").build(),
+ Optional.<ConfigParams>of(config), new NonPersistentDataProvider()).
+ withDispatcher(Dispatchers.DefaultDispatcherId()),
+ factory.generateActorId("follower-"));
+
+ MockRaftActor actor = ref.underlyingActor();
+ actor.waitForRecoveryComplete();
+
+ RaftActorContext newContext = actor.getRaftActorContext();
+ assertEquals("electionTerm", updateEntry.getCurrentTerm(),
+ newContext.getTermInformation().getCurrentTerm());
+ assertEquals("votedFor", updateEntry.getVotedFor(), newContext.getTermInformation().getVotedFor());
+
+ entries = InMemoryJournal.get(persistenceId, UpdateElectionTerm.class);
+ assertEquals("UpdateElectionTerm entries", 1, entries.size());
+ }};
+ }
+
@Test
public void testRaftActorForwardsToRaftActorRecoverySupport() {
String persistenceId = factory.generateActorId("leader-");
new org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm(6, "member3");
mockRaftActor.handleRecover(deprecatedUpdateElectionTerm);
- verify(mockSupport).handleRecoveryMessage(same(snapshotOffer));
- verify(mockSupport).handleRecoveryMessage(same(logEntry));
- verify(mockSupport).handleRecoveryMessage(same(applyJournalEntries));
- verify(mockSupport).handleRecoveryMessage(same(applyLogEntries));
- verify(mockSupport).handleRecoveryMessage(same(deleteEntries));
- verify(mockSupport).handleRecoveryMessage(same(deprecatedDeleteEntries));
- verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm));
- verify(mockSupport).handleRecoveryMessage(same(deprecatedUpdateElectionTerm));
+ verify(mockSupport).handleRecoveryMessage(same(snapshotOffer), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(logEntry), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(applyJournalEntries), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(applyLogEntries), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(deleteEntries), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(deprecatedDeleteEntries), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm), any(PersistentDataProvider.class));
+ verify(mockSupport).handleRecoveryMessage(same(deprecatedUpdateElectionTerm), any(PersistentDataProvider.class));
}
@Test
mockRaftActor.onReceiveCommand(new ApplyJournalEntries(10));
- verify(dataPersistenceProvider, times(2)).persist(anyObject(), any(Procedure.class));
+ verify(dataPersistenceProvider).persist(any(ApplyJournalEntries.class), any(Procedure.class));
}
TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), notifierActor,
- new NonPersistentDataProvider()), persistenceId);
+ new NonPersistentDataProvider()).withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
List<RoleChanged> matches = MessageCollectorActor.expectMatching(notifierActor, RoleChanged.class, 3);
}
};
- raftActor.changeCurrentBehavior(follower);
+ raftActor.newBehavior(follower);
leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
assertEquals(persistenceId, leaderStateChange.getMemberId());
(ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
new MockRaftActorContext.MockPayload("foo-7"))
);
- followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6, (short)0));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6, (short) 0));
assertEquals(8, followerActor.getReplicatedLog().size());
assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
new MockRaftActorContext.MockPayload("foo-7"))
);
// send an additional entry 8 with leaderCommit = 7
- followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7, (short)0));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7, (short) 0));
// 7 and 8, as lastapplied is 7
assertEquals(2, followerActor.getReplicatedLog().size());
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1, (short)0));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1, (short) 0));
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
assertEquals("Real snapshot didn't clear the log till replicatedToAllIndex", 0, leaderActor.getReplicatedLog().size());
//reply from a slow follower after should not raise errors
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1, (short)0));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1, (short) 0));
assertEquals(0, leaderActor.getReplicatedLog().size());
}
};
}};
}
+ @Test
+ public void testRaftActorOnRecoverySnapshot() throws Exception {
+ TEST_LOG.info("testRaftActorOnRecoverySnapshot");
+
+ new JavaTestKit(getSystem()) {{
+ String persistenceId = factory.generateActorId("follower-");
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+
+ // Set the heartbeat interval high to essentially disable election otherwise the test
+ // may fail if the actor is switched to Leader
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+
+ ImmutableMap<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
+
+ // Create mock ReplicatedLogEntry
+ ReplicatedLogEntry replLogEntry = new MockRaftActorContext.MockReplicatedLogEntry(1,1,
+ new MockRaftActorContext.MockPayload("F", 1));
+
+ InMemoryJournal.addEntry(persistenceId, 1, replLogEntry);
+
+ TestActorRef<MockRaftActor> ref = factory.createTestActor(
+ MockRaftActor.props(persistenceId, peerAddresses, Optional.<ConfigParams>of(config)));
+
+ MockRaftActor mockRaftActor = ref.underlyingActor();
+
+ mockRaftActor.waitForRecoveryComplete();
+
+ mockRaftActor.waitForInitializeBehaviorComplete();
+
+ verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(any(ActorRef.class));
+ }};
+ }
+
+ @Test
+ public void testSwitchBehavior(){
+ String persistenceId = factory.generateActorId("leader-");
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setCustomRaftPolicyImplementationClass("org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
+ config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ config.setSnapshotBatchCount(5);
+
+ DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
+
+ Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().build();
+
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
+ MockRaftActor.props(persistenceId, peerAddresses,
+ Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+
+ MockRaftActor leaderActor = mockActorRef.underlyingActor();
+
+ leaderActor.waitForRecoveryComplete();
+
+ leaderActor.handleCommand(new SwitchBehavior(RaftState.Follower, 100));
+
+ assertEquals(100, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
+ assertEquals(RaftState.Follower, leaderActor.getCurrentBehavior().state());
+
+ leaderActor.handleCommand(new SwitchBehavior(RaftState.Leader, 110));
+
+ assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
+
+ leaderActor.handleCommand(new SwitchBehavior(RaftState.Candidate, 125));
+
+ assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
+
+ leaderActor.handleCommand(new SwitchBehavior(RaftState.IsolatedLeader, 125));
+
+ assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
+ assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
+ }
+
public static ByteString fromObject(Object snapshot) throws Exception {
ByteArrayOutputStream b = null;
ObjectOutputStream o = null;
}
}
+ @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,
+ Optional.<ConfigParams>of(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());
+ }
}
Code Review / controller.git / blobdiff