+/*
+ * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
package org.opendaylight.controller.cluster.raft;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
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.eq;
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 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;
}};
}
+ @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);
notifierActor, LeaderStateChanged.class);
assertEquals(raftRoleChanged.getMemberId(), leaderStateChange.getLeaderId());
+ assertEquals(MockRaftActor.PAYLOAD_VERSION, leaderStateChange.getLeaderPayloadVersion());
notifierActor.underlyingActor().clear();
MockRaftActor raftActor = raftActorRef.underlyingActor();
final String newLeaderId = "new-leader";
+ final short newLeaderVersion = 6;
Follower follower = new Follower(raftActor.getRaftActorContext()) {
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
leaderId = newLeaderId;
+ setLeaderPayloadVersion(newLeaderVersion);
return this;
}
};
- raftActor.changeCurrentBehavior(follower);
+ raftActor.newBehavior(follower);
leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
assertEquals(persistenceId, leaderStateChange.getMemberId());
leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
assertEquals(persistenceId, leaderStateChange.getMemberId());
assertEquals(newLeaderId, leaderStateChange.getLeaderId());
+ assertEquals(newLeaderVersion, leaderStateChange.getLeaderPayloadVersion());
+
+ notifierActor.underlyingActor().clear();
+
+ raftActor.handleCommand("any");
+
+ Uninterruptibles.sleepUninterruptibly(505, TimeUnit.MILLISECONDS);
+ leaderStateChange = MessageCollectorActor.getFirstMatching(notifierActor, LeaderStateChanged.class);
+ assertNull(leaderStateChange);
}};
}
leaderActor.getRaftActorContext().getSnapshotManager().persist(snapshotBytes.toByteArray(),
leader, Runtime.getRuntime().totalMemory());
- assertFalse(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
+ assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
// The commit is needed to complete the snapshot creation process
- leaderActor.getRaftActorContext().getSnapshotManager().commit(-1);
+ leaderActor.getRaftActorContext().getSnapshotManager().commit(-1, leader);
// capture snapshot reply should remove the snapshotted entries only
assertEquals(3, leaderActor.getReplicatedLog().size());
(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-3"),
new MockRaftActorContext.MockPayload("foo-4")));
followerActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
- assertFalse(followerActor.getRaftActorContext().getSnapshotManager().isCapturing());
+ assertTrue(followerActor.getRaftActorContext().getSnapshotManager().isCapturing());
// The commit is needed to complete the snapshot creation process
- followerActor.getRaftActorContext().getSnapshotManager().commit(-1);
+ followerActor.getRaftActorContext().getSnapshotManager().commit(-1, follower);
// capture snapshot reply should remove the snapshotted entries only till replicatedToAllIndex
assertEquals(3, followerActor.getReplicatedLog().size()); //indexes 5,6,7 left in the log
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());
new MockRaftActorContext.MockPayload("foo-3"),
new MockRaftActorContext.MockPayload("foo-4")));
leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
- assertFalse(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
+ assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
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());
}
};
leaderActor.setCurrentBehavior(leader);
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
- // Persist another entry (this will cause a CaptureSnapshot to be triggered
- leaderActor.persistData(mockActorRef, "x", new MockRaftActorContext.MockPayload("duh"));
+ // Simulate an install snaphost to a follower.
+ leaderActor.getRaftActorContext().getSnapshotManager().captureToInstall(
+ leaderActor.getReplicatedLog().last(), -1, "member1");
// Now send a CaptureSnapshotReply
mockActorRef.tell(new CaptureSnapshotReply(fromObject("foo").toByteArray()), mockActorRef);
// Trimming log in this scenario is a no-op
assertEquals(-1, leaderActor.getReplicatedLog().getSnapshotIndex());
- assertFalse(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
+ assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
assertEquals(-1, leader.getReplicatedToAllIndex());
}};
// Trimming log in this scenario is a no-op
assertEquals(3, leaderActor.getReplicatedLog().getSnapshotIndex());
- assertFalse(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
+ assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
assertEquals(3, leader.getReplicatedToAllIndex());
}};
}
+ @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