import java.util.concurrent.TimeoutException;
import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import org.opendaylight.controller.cluster.DataPersistenceProvider;
import org.opendaylight.controller.cluster.datastore.DataPersistenceProviderMonitor;
public class RaftActorTest extends AbstractActorTest {
+ private TestActorFactory factory;
+
+ @Before
+ public void setUp(){
+ factory = new TestActorFactory(getSystem());
+ }
@After
- public void tearDown() {
+ public void tearDown() throws Exception {
+ factory.close();
MockAkkaJournal.clearJournal();
MockSnapshotStore.setMockSnapshot(null);
}
@Test
public void testRaftActorRecovery() throws Exception {
new JavaTestKit(getSystem()) {{
- String persistenceId = "follower10";
+ String persistenceId = factory.generateActorId("follower-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
// Set the heartbeat interval high to essentially disable election otherwise the test
// log entry.
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
- ActorRef followerActor = getSystem().actorOf(MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
+ ActorRef followerActor = factory.createActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
watch(followerActor);
int lastAppliedDuringSnapshotCapture = 3;
int lastIndexDuringSnapshotCapture = 4;
- // 4 messages as part of snapshot, which are applied to state
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ // 4 messages as part of snapshot, which are applied to state
+ 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(),
- snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1 ,
+ snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1,
lastAppliedDuringSnapshotCapture, 1);
MockSnapshotStore.setMockSnapshot(snapshot);
MockSnapshotStore.setPersistenceId(persistenceId);
unwatch(followerActor);
//reinstate the actor
- TestActorRef<MockRaftActor> ref = TestActorRef.create(getSystem(),
- MockRaftActor.props(persistenceId, Collections.<String,String>emptyMap(),
+ TestActorRef<MockRaftActor> ref = factory.createTestActor(
+ MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
Optional.<ConfigParams>of(config)));
ref.underlyingActor().waitForRecoveryComplete();
public void testHandleRecoveryWhenDataPersistenceRecoveryApplicable() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryApplicable";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
+ 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(
+ 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);
+ Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}};
}
public void testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), new DataPersistenceProviderMonitor()), persistenceId);
+ 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(
+ 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);
+ Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
}};
}
public void testUpdatingElectionTermCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testUpdatingElectionTermCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
dataPersistenceProviderMonitor.setPersistLatch(persistLatch);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
+
};
}
public void testAddingReplicatedLogEntryCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testAddingReplicatedLogEntryCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
verify(dataPersistenceProvider).persist(eq(logEntry), any(Procedure.class));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
+
};
}
public void testRemovingReplicatedLogEntryCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testRemovingReplicatedLogEntryCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
verify(dataPersistenceProvider, times(2)).persist(anyObject(), any(Procedure.class));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
+
};
}
public void testApplyLogEntriesCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testApplyLogEntriesCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
verify(dataPersistenceProvider, times(1)).persist(anyObject(), any(Procedure.class));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
+
};
}
public void testCaptureSnapshotReplyCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testCaptureSnapshotReplyCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(),
- MockRaftActor.props(persistenceId,Collections.<String,String>emptyMap(),
- Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
+ MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
+ Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("A"),
new MockRaftActorContext.MockPayload("B"),
new MockRaftActorContext.MockPayload("C"),
new MockRaftActorContext.MockPayload("D")));
- mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1));
RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
verify(dataPersistenceProvider).saveSnapshot(anyObject());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
public void testSaveSnapshotSuccessCallsDataPersistence() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testSaveSnapshotSuccessCallsDataPersistence";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
- mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,0, mock(Payload.class)));
- mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,1, mock(Payload.class)));
- mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,2, mock(Payload.class)));
- mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,3, mock(Payload.class)));
- mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,4, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 2, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 3, mock(Payload.class)));
+ mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 4, mock(Payload.class)));
ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("A"),
// Index 2 will not be in the log because it was removed due to snapshotting
assertNull(mockRaftActor.getReplicatedLog().get(2));
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testApplyState";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
verify(mockRaftActor.delegate).applyState(eq(mockActorRef), eq("apply-state"), anyObject());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
public void testApplySnapshot() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testApplySnapshot";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
ReplicatedLog oldReplicatedLog = mockRaftActor.getReplicatedLog();
- oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,0,mock(Payload.class)));
- oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,1,mock(Payload.class)));
+ oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
+ oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class)));
oldReplicatedLog.append(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
- mock(Payload.class)));
+ new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
+ mock(Payload.class)));
ByteString snapshotBytes = fromObject(Arrays.asList(
- new MockRaftActorContext.MockPayload("A"),
- new MockRaftActorContext.MockPayload("B"),
- new MockRaftActorContext.MockPayload("C"),
- new MockRaftActorContext.MockPayload("D")));
+ new MockRaftActorContext.MockPayload("A"),
+ new MockRaftActorContext.MockPayload("B"),
+ new MockRaftActorContext.MockPayload("C"),
+ new MockRaftActorContext.MockPayload("D")));
Snapshot snapshot = mock(Snapshot.class);
verify(mockRaftActor.delegate).applySnapshot(eq(snapshot.getState()));
assertTrue("The replicatedLog should have changed",
- oldReplicatedLog != mockRaftActor.getReplicatedLog());
+ oldReplicatedLog != mockRaftActor.getReplicatedLog());
assertEquals("lastApplied should be same as in the snapshot",
- (Long) 3L, mockRaftActor.getLastApplied());
+ (Long) 3L, mockRaftActor.getLastApplied());
assertEquals(0, mockRaftActor.getReplicatedLog().size());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
public void testSaveSnapshotFailure() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "testSaveSnapshotFailure";
+ String persistenceId = factory.generateActorId("leader-");
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("A"),
new MockRaftActorContext.MockPayload("B"),
new MockRaftActorContext.MockPayload("C"),
mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
- mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1));
+ mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1));
mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
assertEquals("Snapshot index should not have advanced because save snapshot failed", -1,
mockRaftActor.getReplicatedLog().getSnapshotIndex());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
@Test
public void testRaftRoleChangeNotifier() throws Exception {
new JavaTestKit(getSystem()) {{
- ActorRef notifierActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
+ ActorRef notifierActor = factory.createActor(Props.create(MessageCollectorActor.class));
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
- String id = "testRaftRoleChangeNotifier";
+ String persistenceId = factory.generateActorId("notifier-");
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(id,
- Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), notifierActor), id);
+ factory.createTestActor(MockRaftActor.props(persistenceId,
+ Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), notifierActor), persistenceId);
// sleeping for a minimum of 2 seconds, if it spans more its fine.
Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
- List<Object> matches = MessageCollectorActor.getAllMatching(notifierActor, RoleChanged.class);
+ List<Object> matches = MessageCollectorActor.getAllMatching(notifierActor, RoleChanged.class);
assertNotNull(matches);
assertEquals(3, matches.size());
// check if the notifier got a role change from null to Follower
RoleChanged raftRoleChanged = (RoleChanged) matches.get(0);
- assertEquals(id, raftRoleChanged.getMemberId());
+ assertEquals(persistenceId, raftRoleChanged.getMemberId());
assertNull(raftRoleChanged.getOldRole());
assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole());
// check if the notifier got a role change from Follower to Candidate
raftRoleChanged = (RoleChanged) matches.get(1);
- assertEquals(id, raftRoleChanged.getMemberId());
+ assertEquals(persistenceId, raftRoleChanged.getMemberId());
assertEquals(RaftState.Follower.name(), raftRoleChanged.getOldRole());
assertEquals(RaftState.Candidate.name(), raftRoleChanged.getNewRole());
// check if the notifier got a role change from Candidate to Leader
raftRoleChanged = (RoleChanged) matches.get(2);
- assertEquals(id, raftRoleChanged.getMemberId());
+ assertEquals(persistenceId, raftRoleChanged.getMemberId());
assertEquals(RaftState.Candidate.name(), raftRoleChanged.getOldRole());
assertEquals(RaftState.Leader.name(), raftRoleChanged.getNewRole());
}};
public void testFakeSnapshotsForLeaderWithInRealSnapshots() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "leader1";
+ String persistenceId = factory.generateActorId("leader-");
+ String follower1Id = factory.generateActorId("follower-");
ActorRef followerActor1 =
- getSystem().actorOf(Props.create(MessageCollectorActor.class));
+ factory.createActor(Props.create(MessageCollectorActor.class));
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
Map<String, String> peerAddresses = new HashMap<>();
- peerAddresses.put("follower-1", followerActor1.path().toString());
+ peerAddresses.put(follower1Id, followerActor1.path().toString());
TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(),
MockRaftActor.props(persistenceId, peerAddresses,
assertEquals(8, leaderActor.getReplicatedLog().size());
- leaderActor.onReceiveCommand(new CaptureSnapshot(6,1,4,1));
+ leaderActor.onReceiveCommand(new CaptureSnapshot(6, 1, 4, 1));
leaderActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
verify(leaderActor.delegate).createSnapshot();
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
//fake snapshot on index 5
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-1", 1, true, 5, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1));
assertEquals(8, leaderActor.getReplicatedLog().size());
//fake snapshot on index 6
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-1", 1, true, 6, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1));
assertEquals(8, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
assertEquals(8, leaderActor.getReplicatedLog().size());
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("foo-0"),
new MockRaftActorContext.MockPayload("foo-1"),
new MockRaftActorContext.MockPayload("foo-2"),
new ReplicatedLogImplEntry(8, 1, new MockRaftActorContext.MockPayload("foo-8")));
//fake snapshot on index 7, since lastApplied = 7 , we would keep the last applied
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-1", 1, true, 7, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 7, 1));
assertEquals(2, leaderActor.getReplicatedLog().size());
assertEquals(8, leaderActor.getReplicatedLog().lastIndex());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
public void testFakeSnapshotsForFollowerWithInRealSnapshots() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "follower1";
+ String persistenceId = factory.generateActorId("follower-");
+ String leaderId = factory.generateActorId("leader-");
+
ActorRef leaderActor1 =
- getSystem().actorOf(Props.create(MessageCollectorActor.class));
+ factory.createActor(Props.create(MessageCollectorActor.class));
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
Map<String, String> peerAddresses = new HashMap<>();
- peerAddresses.put("leader", leaderActor1.path().toString());
+ peerAddresses.put(leaderId, leaderActor1.path().toString());
TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(),
MockRaftActor.props(persistenceId, peerAddresses,
assertEquals(6, followerActor.getReplicatedLog().size());
//snapshot on 4
- followerActor.onReceiveCommand(new CaptureSnapshot(5,1,4,1));
+ followerActor.onReceiveCommand(new CaptureSnapshot(5, 1, 4, 1));
followerActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
verify(followerActor.delegate).createSnapshot();
(ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
new MockRaftActorContext.MockPayload("foo-6"))
);
- followerActor.onReceiveCommand(new AppendEntries(1, "leader", 5, 1, entries , 5, 5));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5));
assertEquals(7, followerActor.getReplicatedLog().size());
//fake snapshot on index 7
(ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
new MockRaftActorContext.MockPayload("foo-7"))
);
- followerActor.onReceiveCommand(new AppendEntries(1, "leader", 6, 1, entries, 6, 6));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6));
assertEquals(8, followerActor.getReplicatedLog().size());
assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("foo-0"),
new MockRaftActorContext.MockPayload("foo-1"),
new MockRaftActorContext.MockPayload("foo-2"),
new MockRaftActorContext.MockPayload("foo-7"))
);
// send an additional entry 8 with leaderCommit = 7
- followerActor.onReceiveCommand(new AppendEntries(1, "leader", 7, 1, entries , 7, 7));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7));
// 7 and 8, as lastapplied is 7
assertEquals(2, followerActor.getReplicatedLog().size());
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
public void testFakeSnapshotsForLeaderWithInInitiateSnapshots() throws Exception {
new JavaTestKit(getSystem()) {
{
- String persistenceId = "leader1";
+ String persistenceId = factory.generateActorId("leader-");
+ String follower1Id = factory.generateActorId("follower-");
+ String follower2Id = factory.generateActorId("follower-");
ActorRef followerActor1 =
- getSystem().actorOf(Props.create(MessageCollectorActor.class));
+ factory.createActor(Props.create(MessageCollectorActor.class), follower1Id);
ActorRef followerActor2 =
- getSystem().actorOf(Props.create(MessageCollectorActor.class));
+ factory.createActor(Props.create(MessageCollectorActor.class), follower2Id);
DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
Map<String, String> peerAddresses = new HashMap<>();
- peerAddresses.put("follower-1", followerActor1.path().toString());
- peerAddresses.put("follower-2", followerActor2.path().toString());
+ peerAddresses.put(follower1Id, followerActor1.path().toString());
+ peerAddresses.put(follower2Id, followerActor2.path().toString());
- TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(),
+ TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
MockRaftActor.props(persistenceId, peerAddresses,
Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
leaderActor.getRaftActorContext().getReplicatedLog().setSnapshotIndex(4);
assertEquals(5, leaderActor.getReplicatedLog().size());
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-1", 1, true, 9, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1));
assertEquals(5, leaderActor.getReplicatedLog().size());
// set the 2nd follower nextIndex to 1 which has been snapshotted
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-2", 1, true, 0, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1));
assertEquals(5, leaderActor.getReplicatedLog().size());
// simulate a real snapshot
leaderActor.onReceiveCommand(new InitiateInstallSnapshot());
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(String.format("expected to be Leader but was %s. Current Leader = %s ",
- leaderActor.getCurrentBehavior().state(),leaderActor.getLeaderId())
+ leaderActor.getCurrentBehavior().state(), leaderActor.getLeaderId())
, RaftState.Leader, leaderActor.getCurrentBehavior().state());
//reply from a slow follower does not initiate a fake snapshot
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-2", 1, true, 9, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1));
assertEquals("Fake snapshot should not happen when Initiate is in progress", 5, leaderActor.getReplicatedLog().size());
- ByteString snapshotBytes = fromObject(Arrays.asList(
+ ByteString snapshotBytes = fromObject(Arrays.asList(
new MockRaftActorContext.MockPayload("foo-0"),
new MockRaftActorContext.MockPayload("foo-1"),
new MockRaftActorContext.MockPayload("foo-2"),
assertEquals("Real snapshot didn't clear the log till lastApplied", 0, leaderActor.getReplicatedLog().size());
//reply from a slow follower after should not raise errors
- leaderActor.onReceiveCommand(new AppendEntriesReply("follower-2", 1, true, 5, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1));
assertEquals(0, leaderActor.getReplicatedLog().size());
-
- mockActorRef.tell(PoisonPill.getInstance(), getRef());
-
}
};
}
-
-
private ByteString fromObject(Object snapshot) throws Exception {
ByteArrayOutputStream b = null;
ObjectOutputStream o = null;
}
}
}
+
}
--- /dev/null
+/*
+ * Copyright (c) 2014 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;
+
+/*
+ * Copyright (c) 2014 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
+ */
+
+import akka.actor.Actor;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.PoisonPill;
+import akka.actor.Props;
+import akka.testkit.TestActorRef;
+import java.util.LinkedList;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * TestActorFactory provides methods to create both normal and test actors and to kill them when the factory is closed
+ * The ideal usage for TestActorFactory is with try with resources, <br/>
+ * For example <br/>
+ * <pre>
+ * try (TestActorFactory factory = new TestActorFactory(getSystem())){
+ * factory.createActor(props);
+ * factory.createTestActor(props);
+ * factory.generateActorId("leader-");
+ * }
+ * </pre>
+ */
+public class TestActorFactory implements AutoCloseable {
+ private final ActorSystem system;
+ List<ActorRef> createdActors = new LinkedList<>();
+ Logger LOG = LoggerFactory.getLogger(getClass());
+ private static int actorCount = 1;
+
+ public TestActorFactory(ActorSystem system){
+ this.system = system;
+ }
+
+ /**
+ * Create a normal actor with an auto-generated name
+ *
+ * @param props
+ * @return
+ */
+ public ActorRef createActor(Props props){
+ ActorRef actorRef = system.actorOf(props);
+ createdActors.add(actorRef);
+ return actorRef;
+ }
+
+ /**
+ * Create a normal actor with the passed in name
+ * @param props
+ * @param actorId name of actor
+ * @return
+ */
+ public ActorRef createActor(Props props, String actorId){
+ ActorRef actorRef = system.actorOf(props, actorId);
+ createdActors.add(actorRef);
+ return actorRef;
+ }
+
+ /**
+ * Create a test actor with the passed in name
+ * @param props
+ * @param actorId
+ * @param <T>
+ * @return
+ */
+ public <T extends Actor> TestActorRef<T> createTestActor(Props props, String actorId){
+ TestActorRef<T> actorRef = TestActorRef.create(system, props, actorId);
+ createdActors.add(actorRef);
+ return actorRef;
+ }
+
+ /**
+ * Create a test actor with an auto-generated name
+ * @param props
+ * @param <T>
+ * @return
+ */
+ public <T extends Actor> TestActorRef<T> createTestActor(Props props){
+ TestActorRef<T> actorRef = TestActorRef.create(system, props);
+ createdActors.add(actorRef);
+ return actorRef;
+ }
+
+ /**
+ * Generate a friendly but unique actor id/name
+ * @param prefix
+ * @return
+ */
+ public String generateActorId(String prefix){
+ return prefix + actorCount++;
+ }
+
+ @Override
+ public void close() throws Exception {
+ for(ActorRef actor : createdActors){
+ LOG.info("Killing actor {}", actor);
+ actor.tell(PoisonPill.getInstance(), null);
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+/*
+ * Copyright (c) 2015 Brocade Communications 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.behaviors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.dispatch.Dispatchers;
+import akka.testkit.JavaTestKit;
+import akka.testkit.TestActorRef;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.junit.After;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.MockRaftActorContext.SimpleReplicatedLog;
+import org.opendaylight.controller.cluster.raft.RaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftState;
+import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
+import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.RequestVote;
+import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
+import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.impl.SimpleLogger;
+import scala.concurrent.duration.FiniteDuration;
+
+/**
+ * Tests various leader election scenarios.
+ *
+ * @author Thomas Pantelis
+ */
+public class LeaderElectionScenariosTest {
+
+ private static final int HEARTBEAT_INTERVAL = 50;
+
+ public static class MemberActor extends MessageCollectorActor {
+
+ volatile RaftActorBehavior behavior;
+ Map<Class<?>, CountDownLatch> messagesReceivedLatches = new ConcurrentHashMap<>();
+ Map<Class<?>, Boolean> dropMessagesToBehavior = new ConcurrentHashMap<>();
+ CountDownLatch behaviorStateChangeLatch;
+
+ public static Props props() {
+ return Props.create(MemberActor.class).withDispatcher(Dispatchers.DefaultDispatcherId());
+ }
+
+ @Override
+ public void onReceive(Object message) throws Exception {
+ // Ignore scheduled SendHeartBeat messages.
+ if(message instanceof SendHeartBeat) {
+ return;
+ }
+
+ try {
+ if(behavior != null && !dropMessagesToBehavior.containsKey(message.getClass())) {
+ RaftActorBehavior oldBehavior = behavior;
+ behavior = behavior.handleMessage(getSender(), message);
+ if(behavior != oldBehavior && behaviorStateChangeLatch != null) {
+ behaviorStateChangeLatch.countDown();
+ }
+ }
+ } finally {
+ super.onReceive(message);
+
+ CountDownLatch latch = messagesReceivedLatches.get(message.getClass());
+ if(latch != null) {
+ latch.countDown();
+ }
+ }
+ }
+
+ void expectBehaviorStateChange() {
+ behaviorStateChangeLatch = new CountDownLatch(1);
+ }
+
+ void waitForBehaviorStateChange() {
+ assertTrue("Expected behavior state change",
+ Uninterruptibles.awaitUninterruptibly(behaviorStateChangeLatch, 5, TimeUnit.SECONDS));
+ }
+
+ void expectMessageClass(Class<?> expClass, int expCount) {
+ messagesReceivedLatches.put(expClass, new CountDownLatch(expCount));
+ }
+
+ void waitForExpectedMessages(Class<?> expClass) {
+ CountDownLatch latch = messagesReceivedLatches.get(expClass);
+ assertNotNull("No messages received for " + expClass, latch);
+ assertTrue("Missing messages of type " + expClass,
+ Uninterruptibles.awaitUninterruptibly(latch, 5, TimeUnit.SECONDS));
+ }
+
+ void dropMessagesToBehavior(Class<?> msgClass) {
+ dropMessagesToBehavior(msgClass, 1);
+ }
+
+ void dropMessagesToBehavior(Class<?> msgClass, int expCount) {
+ expectMessageClass(msgClass, expCount);
+ dropMessagesToBehavior.put(msgClass, Boolean.TRUE);
+ }
+
+ void clearDropMessagesToBehavior() {
+ dropMessagesToBehavior.clear();
+ }
+
+ @Override
+ public void clear() {
+ behaviorStateChangeLatch = null;
+ clearDropMessagesToBehavior();
+ messagesReceivedLatches.clear();
+ super.clear();
+ }
+
+ void forwardCapturedMessageToBehavior(Class<?> msgClass, ActorRef sender) throws Exception {
+ Object message = getFirstMatching(getSelf(), msgClass);
+ assertNotNull("Message of type " + msgClass + " not received", message);
+ getSelf().tell(message, sender);
+ }
+
+ void forwardCapturedMessagesToBehavior(Class<?> msgClass, ActorRef sender) throws Exception {
+ for(Object m: getAllMatching(getSelf(), msgClass)) {
+ getSelf().tell(m, sender);
+ }
+ }
+
+ <T> T getCapturedMessage(Class<T> msgClass) throws Exception {
+ Object message = getFirstMatching(getSelf(), msgClass);
+ assertNotNull("Message of type " + msgClass + " not received", message);
+ return (T) message;
+ }
+ }
+
+ static {
+ System.setProperty(SimpleLogger.LOG_KEY_PREFIX + MockRaftActorContext.class.getName(), "trace");
+ }
+
+ private final Logger testLog = LoggerFactory.getLogger(MockRaftActorContext.class);
+ private final ActorSystem system = ActorSystem.create("test");
+
+ @After
+ public void tearDown() {
+ JavaTestKit.shutdownActorSystem(system);
+ }
+
+ private DefaultConfigParamsImpl newConfigParams() {
+ DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
+ configParams.setHeartBeatInterval(new FiniteDuration(HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS));
+ configParams.setElectionTimeoutFactor(100000);
+ configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
+ return configParams;
+ }
+
+ private MockRaftActorContext newRaftActorContext(String id, ActorRef actor,
+ Map<String, String> peerAddresses) {
+ MockRaftActorContext context = new MockRaftActorContext(id, system, actor);
+ context.setPeerAddresses(peerAddresses);
+ context.getTermInformation().updateAndPersist(1, "");
+ return context;
+ }
+
+ private void verifyBehaviorState(String name, TestActorRef<MemberActor> actor, RaftState expState) {
+ assertEquals(name + " behavior state", expState, actor.underlyingActor().behavior.state());
+ }
+
+ private void initializeLeaderBehavior(TestActorRef<MemberActor> actor, RaftActorContext context,
+ int numActiveFollowers) throws Exception {
+ // Leader sends immediate heartbeats - we don't care about it so ignore it.
+
+ actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, numActiveFollowers);
+ Leader leader = new Leader(context);
+ actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+ actor.underlyingActor().behavior = leader;
+
+ actor.underlyingActor().forwardCapturedMessagesToBehavior(AppendEntriesReply.class, ActorRef.noSender());
+ actor.underlyingActor().clear();
+ }
+
+ private TestActorRef<MemberActor> newMemberActor(String name) throws Exception {
+ TestActorRef<MemberActor> actor = TestActorRef.create(system, MemberActor.props(), name);
+ MessageCollectorActor.waitUntilReady(actor);
+ return actor;
+ }
+
+ private void sendHeartbeat(TestActorRef<MemberActor> leaderActor) {
+ Uninterruptibles.sleepUninterruptibly(HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS);
+ leaderActor.underlyingActor().behavior.handleMessage(leaderActor, new SendHeartBeat());
+ }
+
+ @Test
+ public void testDelayedMessagesScenario() throws Exception {
+ testLog.info("Starting testDelayedMessagesScenario");
+
+ TestActorRef<MemberActor> member1Actor = newMemberActor("member1");
+ TestActorRef<MemberActor> member2Actor = newMemberActor("member2");
+ TestActorRef<MemberActor> member3Actor = newMemberActor("member3");
+
+ // Create member 2's behavior initially as Follower
+
+ MockRaftActorContext member2Context = newRaftActorContext("member2", member2Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member2ConfigParams = newConfigParams();
+ member2Context.setConfigParams(member2ConfigParams);
+
+ Follower member2Behavior = new Follower(member2Context);
+ member2Actor.underlyingActor().behavior = member2Behavior;
+
+ // Create member 3's behavior initially as Follower
+
+ MockRaftActorContext member3Context = newRaftActorContext("member3", member3Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member2", member2Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member3ConfigParams = newConfigParams();
+ member3Context.setConfigParams(member3ConfigParams);
+
+ Follower member3Behavior = new Follower(member3Context);
+ member3Actor.underlyingActor().behavior = member3Behavior;
+
+ // Create member 1's behavior initially as Leader
+
+ MockRaftActorContext member1Context = newRaftActorContext("member1", member1Actor,
+ ImmutableMap.<String,String>builder().
+ put("member2", member2Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member1ConfigParams = newConfigParams();
+ member1Context.setConfigParams(member1ConfigParams);
+
+ initializeLeaderBehavior(member1Actor, member1Context, 2);
+
+ member2Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().clear();
+
+ // Send ElectionTimeout to member 2 to simulate missing heartbeat from the Leader. member 2
+ // should switch to Candidate and send out RequestVote messages. Set member 1 and 3 actors
+ // to capture RequestVote but not to forward to the behavior just yet as we want to
+ // control the order of RequestVote messages to member 1 and 3.
+
+ member1Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class);
+
+ member2Actor.underlyingActor().expectBehaviorStateChange();
+
+ member3Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class);
+
+ member2Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ member2Actor.underlyingActor().waitForBehaviorStateChange();
+ verifyBehaviorState("member 2", member2Actor, RaftState.Candidate);
+
+ assertEquals("member 1 election term", 1, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 2, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 1, member3Context.getTermInformation().getCurrentTerm());
+
+ // At this point member 1 and 3 actors have captured the RequestVote messages. First
+ // forward the RequestVote message to member 1's behavior. Since the RequestVote term
+ // is greater than member 1's term, member 1 should switch to Follower without replying
+ // to RequestVote and update its term to 2.
+
+ member1Actor.underlyingActor().clearDropMessagesToBehavior();
+ member1Actor.underlyingActor().expectBehaviorStateChange();
+ member1Actor.underlyingActor().forwardCapturedMessageToBehavior(RequestVote.class, member2Actor);
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ member1Actor.underlyingActor().waitForBehaviorStateChange();
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+
+ // Now forward member 3's captured RequestVote message to its behavior. Since member 3 is
+ // already a Follower, it should update its term to 2 and send a RequestVoteReply back to
+ // member 2 granting the vote b/c the RequestVote's term, lastLogTerm, and lastLogIndex
+ // should satisfy the criteria for granting the vote. However, we'll delay sending the
+ // RequestVoteReply to member 2's behavior to simulate network latency.
+
+ member2Actor.underlyingActor().dropMessagesToBehavior(RequestVoteReply.class);
+
+ member3Actor.underlyingActor().clearDropMessagesToBehavior();
+ member3Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member3Actor.underlyingActor().forwardCapturedMessageToBehavior(RequestVote.class, member2Actor);
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Follower);
+
+ assertEquals("member 1 election term", 2, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 2, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 2, member3Context.getTermInformation().getCurrentTerm());
+
+ // Send ElectionTimeout to member 3 to simulate missing heartbeat from a Leader. member 3
+ // should switch to Candidate and send out RequestVote messages. member 1 should grant the
+ // vote and send a reply. After receiving the RequestVoteReply, member 3 should switch to leader.
+
+ member2Actor.underlyingActor().expectBehaviorStateChange();
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(RequestVoteReply.class, 1);
+ member3Actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, 2);
+
+ member3Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ RequestVoteReply requestVoteReply = member3Actor.underlyingActor().getCapturedMessage(RequestVoteReply.class);
+ assertEquals("getTerm", member3Context.getTermInformation().getCurrentTerm(), requestVoteReply.getTerm());
+ assertEquals("isVoteGranted", true, requestVoteReply.isVoteGranted());
+
+ verifyBehaviorState("member 3", member3Actor, RaftState.Leader);
+
+ // member 2 should've switched to Follower as member 3's RequestVote term (3) was greater
+ // than member 2's term (2).
+
+ member2Actor.underlyingActor().waitForBehaviorStateChange();
+ verifyBehaviorState("member 2", member2Actor, RaftState.Follower);
+
+ // The switch to leader should cause an immediate AppendEntries heartbeat from member 3.
+
+ member3Actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 3, member3Context.getTermInformation().getCurrentTerm());
+
+ // Now forward the original delayed RequestVoteReply from member 3 to member 2 that granted
+ // the vote. Since member 2 is now a Follower, the RequestVoteReply should be ignored.
+
+ member2Actor.underlyingActor().clearDropMessagesToBehavior();
+ member2Actor.underlyingActor().forwardCapturedMessageToBehavior(RequestVoteReply.class, member3Actor);
+
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Follower);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Leader);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 3, member3Context.getTermInformation().getCurrentTerm());
+
+ testLog.info("testDelayedMessagesScenario done");
+ }
+
+ @Test
+ public void testPartitionedLeadersScenario() throws Exception {
+ testLog.info("Starting testPartitionedLeadersScenario");
+
+ TestActorRef<MemberActor> member1Actor = newMemberActor("member1");
+ TestActorRef<MemberActor> member2Actor = newMemberActor("member2");
+ TestActorRef<MemberActor> member3Actor = newMemberActor("member3");
+
+ // Create member 2's behavior initially as Follower
+
+ MockRaftActorContext member2Context = newRaftActorContext("member2", member2Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member2ConfigParams = newConfigParams();
+ member2Context.setConfigParams(member2ConfigParams);
+
+ Follower member2Behavior = new Follower(member2Context);
+ member2Actor.underlyingActor().behavior = member2Behavior;
+
+ // Create member 3's behavior initially as Follower
+
+ MockRaftActorContext member3Context = newRaftActorContext("member3", member3Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member2", member2Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member3ConfigParams = newConfigParams();
+ member3Context.setConfigParams(member3ConfigParams);
+
+ Follower member3Behavior = new Follower(member3Context);
+ member3Actor.underlyingActor().behavior = member3Behavior;
+
+ // Create member 1's behavior initially as Leader
+
+ MockRaftActorContext member1Context = newRaftActorContext("member1", member1Actor,
+ ImmutableMap.<String,String>builder().
+ put("member2", member2Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member1ConfigParams = newConfigParams();
+ member1Context.setConfigParams(member1ConfigParams);
+
+ initializeLeaderBehavior(member1Actor, member1Context, 2);
+
+ member2Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().clear();
+
+ // Send ElectionTimeout to member 2 to simulate no heartbeat from the Leader (member 1).
+ // member 2 should switch to Candidate, start new term 2 and send out RequestVote messages.
+ // member 1 will switch to Follower b/c its term is less than the RequestVote term, also it
+ // won't send back a reply. member 3 will drop the message (ie won't forward it to its behavior) to
+ // simulate loss of network connectivity between member 2 and 3.
+
+ member1Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+
+ member2Actor.underlyingActor().expectBehaviorStateChange();
+
+ member3Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class);
+
+ member2Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ // member 1 should switch to Follower as the RequestVote term is greater than its term. It
+ // won't send back a RequestVoteReply in this case.
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+
+ // member 2 should switch to Candidate since member 1 didn't reply.
+
+ member2Actor.underlyingActor().waitForBehaviorStateChange();
+ verifyBehaviorState("member 2", member2Actor, RaftState.Candidate);
+
+ assertEquals("member 1 election term", 2, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 2, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 1, member3Context.getTermInformation().getCurrentTerm());
+
+ // Send ElectionTimeout to member 3 to simulate no heartbeat from the Leader (member 1).
+ // member 2 should switch to Candidate and send out RequestVote messages. member 1 will reply and
+ // grant the vote but member 2 will drop the message to simulate loss of network connectivity.
+
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member1Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class);
+ member2Actor.underlyingActor().dropMessagesToBehavior(AppendEntries.class);
+
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(RequestVoteReply.class, 1);
+ member3Actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, 1);
+
+ member3Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ RequestVoteReply requestVoteReply = member3Actor.underlyingActor().getCapturedMessage(RequestVoteReply.class);
+ assertEquals("getTerm", member3Context.getTermInformation().getCurrentTerm(), requestVoteReply.getTerm());
+ assertEquals("isVoteGranted", true, requestVoteReply.isVoteGranted());
+
+ // when member 3 switches to Leader it will immediately send out heartbeat AppendEntries to
+ // the followers. Wait for AppendEntries to member 1 and its AppendEntriesReply. The
+ // AppendEntries message to member 2 is dropped.
+
+ member1Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+ member2Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Candidate);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Leader);
+
+ assertEquals("member 1 election term", 2, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 2, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 2, member3Context.getTermInformation().getCurrentTerm());
+
+ // member 2 is partitioned from the Leader (member 3) and hasn't received any messages. It
+ // would get another ElectionTimeout so simulate that. member 1 should send back a reply
+ // granting the vote. Messages (RequestVote and AppendEntries) from member 2 to member 3
+ // are dropped to simulate loss of network connectivity. Note member 2 will increment its
+ // election term to 3.
+
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().expectMessageClass(RequestVoteReply.class, 1);
+ member2Actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, 1);
+
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().dropMessagesToBehavior(AppendEntries.class);
+ member3Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class);
+
+ member2Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ requestVoteReply = member2Actor.underlyingActor().getCapturedMessage(RequestVoteReply.class);
+ assertEquals("getTerm", member2Context.getTermInformation().getCurrentTerm(), requestVoteReply.getTerm());
+ assertEquals("isVoteGranted", true, requestVoteReply.isVoteGranted());
+
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ member1Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+ member2Actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+
+ // We end up with 2 partitioned leaders both leading member 1. The term for member 1 and 3
+ // is 3 and member 3's term is 2.
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Leader);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Leader);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 2, member3Context.getTermInformation().getCurrentTerm());
+
+ // Re-establish connectivity between member 2 and 3, ie stop dropping messages between
+ // the 2. Send heartbeats (AppendEntries) from member 3. Both member 1 and 2 should send back
+ // an unsuccessful AppendEntriesReply b/c their term (3) is greater than member 3's term (2).
+ // This should cause member 3 to switch to Follower.
+
+ RaftActorBehavior savedMember1Behavior = member1Actor.underlyingActor().behavior;
+ RaftActorBehavior savedMember2Behavior = member2Actor.underlyingActor().behavior;
+ RaftActorBehavior savedMember3Behavior = member3Actor.underlyingActor().behavior;
+ long savedMember3Term = member3Context.getTermInformation().getCurrentTerm();
+ String savedMember3VoterFor = member3Context.getTermInformation().getVotedFor();
+
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, 1);
+
+ sendHeartbeat(member3Actor);
+
+ member3Actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+
+ AppendEntriesReply appendEntriesReply = member3Actor.underlyingActor().
+ getCapturedMessage(AppendEntriesReply.class);
+ assertEquals("isSuccess", false, appendEntriesReply.isSuccess());
+ assertEquals("getTerm", 3, appendEntriesReply.getTerm());
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Leader);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Follower);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 3, member3Context.getTermInformation().getCurrentTerm());
+
+ // Revert back to the partitioned leaders state to test the other sequence where member 2
+ // sends heartbeats first before member 3. member 1 should return a successful
+ // AppendEntriesReply b/c his term matches member 2's. member 3 should switch to Follower
+ // as his term is less than member 2's.
+
+ member1Actor.underlyingActor().behavior = savedMember1Behavior;
+ member2Actor.underlyingActor().behavior = savedMember2Behavior;
+ member3Actor.underlyingActor().behavior = savedMember3Behavior;
+
+ member3Context.getTermInformation().update(savedMember3Term, savedMember3VoterFor);
+
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().expectMessageClass(AppendEntriesReply.class, 1);
+
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(AppendEntries.class, 1);
+
+ sendHeartbeat(member2Actor);
+
+ member1Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(AppendEntries.class);
+
+ member2Actor.underlyingActor().waitForExpectedMessages(AppendEntriesReply.class);
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Leader);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Follower);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", 3, member3Context.getTermInformation().getCurrentTerm());
+
+ testLog.info("testPartitionedLeadersScenario done");
+ }
+
+ @Test
+ public void testPartitionedCandidateOnStartupScenario() throws Exception {
+ testLog.info("Starting testPartitionedCandidateOnStartupScenario");
+
+ TestActorRef<MemberActor> member1Actor = newMemberActor("member1") ;
+ TestActorRef<MemberActor> member2Actor = newMemberActor("member2");
+ TestActorRef<MemberActor> member3Actor = newMemberActor("member3");
+
+ // Create member 2's behavior as Follower.
+
+ MockRaftActorContext member2Context = newRaftActorContext("member2", member2Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member2ConfigParams = newConfigParams();
+ member2Context.setConfigParams(member2ConfigParams);
+
+ Follower member2Behavior = new Follower(member2Context);
+ member2Actor.underlyingActor().behavior = member2Behavior;
+
+ // Create member 1's behavior as Leader.
+
+ MockRaftActorContext member1Context = newRaftActorContext("member1", member1Actor,
+ ImmutableMap.<String,String>builder().
+ put("member2", member2Actor.path().toString()).
+ put("member3", member3Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member1ConfigParams = newConfigParams();
+ member1Context.setConfigParams(member1ConfigParams);
+
+ initializeLeaderBehavior(member1Actor, member1Context, 1);
+
+ member2Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().clear();
+
+ // Initialize the ReplicatedLog and election term info for member 1 and 2. The current term
+ // will be 3 and the last term will be 2.
+
+ SimpleReplicatedLog replicatedLog = new SimpleReplicatedLog();
+ replicatedLog.append(new MockReplicatedLogEntry(2, 1, new MockPayload("")));
+ replicatedLog.append(new MockReplicatedLogEntry(3, 1, new MockPayload("")));
+
+ member1Context.setReplicatedLog(replicatedLog);
+ member1Context.getTermInformation().update(3, "");
+
+ member2Context.setReplicatedLog(replicatedLog);
+ member2Context.getTermInformation().update(3, member1Context.getId());
+
+ // Create member 3's behavior initially as a Candidate.
+
+ MockRaftActorContext member3Context = newRaftActorContext("member3", member3Actor,
+ ImmutableMap.<String,String>builder().
+ put("member1", member1Actor.path().toString()).
+ put("member2", member2Actor.path().toString()).build());
+
+ DefaultConfigParamsImpl member3ConfigParams = newConfigParams();
+ member3Context.setConfigParams(member3ConfigParams);
+
+ // Initialize the ReplicatedLog and election term info for Candidate member 3. The current term
+ // will be 2 and the last term will be 1 so it is behind the leader's log.
+
+ SimpleReplicatedLog candidateReplicatedLog = new SimpleReplicatedLog();
+ candidateReplicatedLog.append(new MockReplicatedLogEntry(1, 1, new MockPayload("")));
+ candidateReplicatedLog.append(new MockReplicatedLogEntry(2, 1, new MockPayload("")));
+
+ member3Context.setReplicatedLog(candidateReplicatedLog);
+ member3Context.getTermInformation().update(2, member1Context.getId());
+
+ // The member 3 Candidate will start a new term and send RequestVotes. However it will be
+ // partitioned from the cluster by having member 1 and 2 drop its RequestVote messages.
+
+ int numCandidateElections = 5;
+ long candidateElectionTerm = member3Context.getTermInformation().getCurrentTerm() + numCandidateElections;
+
+ member1Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class, numCandidateElections);
+
+ member2Actor.underlyingActor().dropMessagesToBehavior(RequestVote.class, numCandidateElections);
+
+ Candidate member3Behavior = new Candidate(member3Context);
+ member3Actor.underlyingActor().behavior = member3Behavior;
+
+ // Send several additional ElectionTimeouts to Candidate member 3. Each ElectionTimeout will
+ // start a new term so Candidate member 3's current term will be greater than the leader's
+ // current term.
+
+ for(int i = 0; i < numCandidateElections - 1; i++) {
+ member3Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+ }
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Leader);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Follower);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Candidate);
+
+ assertEquals("member 1 election term", 3, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", 3, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", candidateElectionTerm,
+ member3Context.getTermInformation().getCurrentTerm());
+
+ // Now send a couple more ElectionTimeouts to Candidate member 3 with the partition resolved.
+ //
+ // On the first RequestVote, Leader member 1 should switch to Follower as its term (s) is less than
+ // the RequestVote's term (8) from member 3. No RequestVoteReply should be sent by member 1.
+ // Follower member 2 should update its term since it less than the RequestVote's term and
+ // should return a RequestVoteReply but should not grant the vote as its last term and index
+ // is greater than the RequestVote's lastLogTerm and lastLogIndex, ie member 2's log is later
+ // or more up to date than member 3's.
+ //
+ // On the second RequestVote, both member 1 and 2 are followers so they should update their
+ // term and return a RequestVoteReply but should not grant the vote.
+
+ candidateElectionTerm += 2;
+ for(int i = 0; i < 2; i++) {
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(RequestVoteReply.class, 1);
+
+ member3Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ RequestVoteReply requestVoteReply = member3Actor.underlyingActor().getCapturedMessage(RequestVoteReply.class);
+ assertEquals("getTerm", member3Context.getTermInformation().getCurrentTerm(), requestVoteReply.getTerm());
+ assertEquals("isVoteGranted", false, requestVoteReply.isVoteGranted());
+ }
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Follower);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Follower);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Candidate);
+
+ // Even though member 3 didn't get voted for, member 1 and 2 should have updated their term
+ // to member 3's.
+
+ assertEquals("member 1 election term", candidateElectionTerm,
+ member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", candidateElectionTerm,
+ member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", candidateElectionTerm,
+ member3Context.getTermInformation().getCurrentTerm());
+
+ // At this point we have no leader. Candidate member 3 would continue to start new elections
+ // but wouldn't be granted a vote. One of the 2 followers would eventually time out from
+ // not having received a heartbeat from a leader and switch to candidate and start a new
+ // election. We'll simulate that here by sending an ElectionTimeout to member 1.
+
+ member1Actor.underlyingActor().clear();
+ member1Actor.underlyingActor().expectMessageClass(RequestVoteReply.class, 1);
+ member2Actor.underlyingActor().clear();
+ member2Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member3Actor.underlyingActor().clear();
+ member3Actor.underlyingActor().expectMessageClass(RequestVote.class, 1);
+ member3Actor.underlyingActor().expectBehaviorStateChange();
+
+ member1Actor.tell(new ElectionTimeout(), ActorRef.noSender());
+
+ member2Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+ member3Actor.underlyingActor().waitForExpectedMessages(RequestVote.class);
+
+ // The RequestVoteReply should come from Follower member 2 and the vote should be granted
+ // since member 2's last term and index matches member 1's.
+
+ member1Actor.underlyingActor().waitForExpectedMessages(RequestVoteReply.class);
+
+ RequestVoteReply requestVoteReply = member1Actor.underlyingActor().getCapturedMessage(RequestVoteReply.class);
+ assertEquals("getTerm", member1Context.getTermInformation().getCurrentTerm(), requestVoteReply.getTerm());
+ assertEquals("isVoteGranted", true, requestVoteReply.isVoteGranted());
+
+ // Candidate member 3 should change to follower as its term should be less than the
+ // RequestVote term (member 1 started a new term higher than the other member's terms).
+
+ member3Actor.underlyingActor().waitForBehaviorStateChange();
+
+ verifyBehaviorState("member 1", member1Actor, RaftState.Leader);
+ verifyBehaviorState("member 2", member2Actor, RaftState.Follower);
+ verifyBehaviorState("member 3", member3Actor, RaftState.Follower);
+
+ // newTerm should be 10.
+
+ long newTerm = candidateElectionTerm + 1;
+ assertEquals("member 1 election term", newTerm, member1Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 2 election term", newTerm, member2Context.getTermInformation().getCurrentTerm());
+ assertEquals("member 3 election term", newTerm, member3Context.getTermInformation().getCurrentTerm());
+
+ testLog.info("testPartitionedCandidateOnStartupScenario done");
+ }
+}
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages;
-import org.slf4j.impl.SimpleLogger;
import scala.concurrent.duration.FiniteDuration;
public class LeaderTest extends AbstractRaftActorBehaviorTest {
- static {
- // This enables trace logging for the tests.
- System.setProperty(SimpleLogger.LOG_KEY_PREFIX + MockRaftActorContext.class.getName(), "trace");
- }
-
private final ActorRef leaderActor =
getSystem().actorOf(Props.create(DoNothingActor.class));
private final ActorRef senderActor =
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
public class MessageCollectorActor extends UntypedActor {
+ private static final String ARE_YOU_READY = "ARE_YOU_READY";
+
private final List<Object> messages = new ArrayList<>();
@Override public void onReceive(Object message) throws Exception {
+ if(message.equals(ARE_YOU_READY)) {
+ getSender().tell("yes", getSelf());
+ return;
+ }
+
if(message instanceof String){
if("get-all-messages".equals(message)){
- getSender().tell(new ArrayList(messages), getSelf());
+ getSender().tell(new ArrayList<>(messages), getSelf());
}
- } else {
+ } else if(message != null) {
messages.add(message);
}
}
Timeout operationTimeout = new Timeout(operationDuration);
Future<Object> future = Patterns.ask(actor, "get-all-messages", operationTimeout);
- try {
- return (List<Object>) Await.result(future, operationDuration);
- } catch (Exception e) {
- throw e;
- }
+ return (List<Object>) Await.result(future, operationDuration);
}
/**
return output;
}
+ public static void waitUntilReady(ActorRef actor) throws Exception {
+ long timeout = 500;
+ FiniteDuration duration = Duration.create(timeout, TimeUnit.MILLISECONDS);
+ for(int i = 0; i < 10; i++) {
+ try {
+ Await.ready(Patterns.ask(actor, ARE_YOU_READY, timeout), duration);
+ return;
+ } catch (TimeoutException e) {
+ }
+ }
+
+ throw new TimeoutException("Actor not ready in time.");
+ }
}
--- /dev/null
+org.slf4j.simpleLogger.showDateTime=true
+org.slf4j.simpleLogger.dateTimeFormat=hh:mm:ss,S a
+org.slf4j.simpleLogger.logFile=System.out
+org.slf4j.simpleLogger.showShortLogName=true
+org.slf4j.simpleLogger.levelInBrackets=true
+org.slf4j.simpleLogger.org.opendaylight.controller.cluster.raft=trace
\ No newline at end of file
--- /dev/null
+org.slf4j.simpleLogger.showDateTime=true
+org.slf4j.simpleLogger.dateTimeFormat=hh:mm:ss,S a
+org.slf4j.simpleLogger.logFile=System.out
+org.slf4j.simpleLogger.showShortLogName=true
+org.slf4j.simpleLogger.levelInBrackets=true
+org.slf4j.simpleLogger.log.org.opendaylight.controller.cluster.datastore=trace
\ No newline at end of file
Code Review / controller.git / commitdiff