import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
+import static org.mockito.Matchers.any;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
+
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.testkit.TestActorRef;
import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.Uninterruptibles;
import com.google.protobuf.ByteString;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
+import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
import org.opendaylight.controller.cluster.raft.messages.RequestVote;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
+import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
+import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
+import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
import scala.concurrent.duration.FiniteDuration;
-public class FollowerTest extends AbstractRaftActorBehaviorTest {
+public class FollowerTest extends AbstractRaftActorBehaviorTest<Follower> {
private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
- private RaftActorBehavior follower;
+ private Follower follower;
private final short payloadVersion = 5;
@Override
@After
public void tearDown() throws Exception {
- if(follower != null) {
+ if (follower != null) {
follower.close();
}
}
@Override
- protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
- return new TestFollower(actorContext);
+ protected Follower createBehavior(RaftActorContext actorContext) {
+ return spy(new Follower(actorContext));
}
@Override
}
@Override
- protected MockRaftActorContext createActorContext(ActorRef actorRef){
+ protected MockRaftActorContext createActorContext(ActorRef actorRef) {
MockRaftActorContext context = new MockRaftActorContext("follower", getSystem(), actorRef);
context.setPayloadVersion(payloadVersion );
return context;
}
- private static int getElectionTimeoutCount(RaftActorBehavior follower){
- if(follower instanceof TestFollower){
- return ((TestFollower) follower).getElectionTimeoutCount();
- }
- return -1;
- }
-
@Test
- public void testThatAnElectionTimeoutIsTriggered(){
+ public void testThatAnElectionTimeoutIsTriggered() {
MockRaftActorContext actorContext = createActorContext();
follower = new Follower(actorContext);
- MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class,
+ MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class,
actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
}
@Test
- public void testHandleElectionTimeout(){
- logStart("testHandleElectionTimeout");
+ public void testHandleElectionTimeoutWhenNoLeaderMessageReceived() {
+ logStart("testHandleElectionTimeoutWhenNoLeaderMessageReceived");
- follower = new Follower(createActorContext());
+ MockRaftActorContext context = createActorContext();
+ follower = new Follower(context);
- RaftActorBehavior raftBehavior = follower.handleMessage(followerActor, ElectionTimeout.INSTANCE);
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams().getElectionTimeOutInterval().toMillis(),
+ TimeUnit.MILLISECONDS);
+ RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
assertTrue(raftBehavior instanceof Candidate);
}
@Test
- public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull(){
+ public void testHandleElectionTimeoutWhenLeaderMessageReceived() {
+ logStart("testHandleElectionTimeoutWhenLeaderMessageReceived");
+
+ MockRaftActorContext context = createActorContext();
+ ((DefaultConfigParamsImpl) context.getConfigParams())
+ .setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
+ ((DefaultConfigParamsImpl) context.getConfigParams()).setElectionTimeoutFactor(4);
+
+ follower = new Follower(context);
+ context.setCurrentBehavior(follower);
+
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
+ .getElectionTimeOutInterval().toMillis() - 100, TimeUnit.MILLISECONDS);
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
+ -1, -1, (short) 1));
+
+ Uninterruptibles.sleepUninterruptibly(130, TimeUnit.MILLISECONDS);
+ RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
+ assertTrue(raftBehavior instanceof Follower);
+
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
+ .getElectionTimeOutInterval().toMillis() - 150, TimeUnit.MILLISECONDS);
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
+ -1, -1, (short) 1));
+
+ Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+ raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
+ assertTrue(raftBehavior instanceof Follower);
+ }
+
+ @Test
+ public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull() {
logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
- RaftActorContext context = createActorContext();
+ MockRaftActorContext context = createActorContext();
long term = 1000;
context.getTermInformation().update(term, null);
assertEquals("isVoteGranted", true, reply.isVoteGranted());
assertEquals("getTerm", term, reply.getTerm());
- assertEquals("schedule election", 1, getElectionTimeoutCount(follower));
+ verify(follower).scheduleElection(any(FiniteDuration.class));
}
@Test
- public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId(){
+ public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId() {
logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
- RaftActorContext context = createActorContext();
+ MockRaftActorContext context = createActorContext();
long term = 1000;
context.getTermInformation().update(term, "test");
RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
assertEquals("isVoteGranted", false, reply.isVoteGranted());
- assertEquals("schedule election", 0, getElectionTimeoutCount(follower));
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
}
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
assertFalse(syncStatus.isInitialSyncDone());
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
assertFalse(syncStatus.isInitialSyncDone());
}
@Test
- public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog() throws Exception {
+ public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog()
+ throws Exception {
logStart("testHandleFirstAppendEntries");
MockRaftActorContext context = createActorContext();
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
assertFalse(syncStatus.isInitialSyncDone());
}
@Test
- public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot() throws Exception {
+ public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot()
+ throws Exception {
logStart("testHandleFirstAppendEntries");
MockRaftActorContext context = createActorContext();
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
assertFalse(syncStatus.isInitialSyncDone());
}
@Test
- public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshotButCalculatedPreviousEntryMissing() throws Exception {
- logStart("testHandleFirstAppendEntries");
+ public void testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing()
+ throws Exception {
+ logStart(
+ "testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing");
MockRaftActorContext context = createActorContext();
context.getReplicatedLog().clear(0,2);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
assertFalse(syncStatus.isInitialSyncDone());
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
assertFalse(syncStatus.isInitialSyncDone());
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
assertFalse(syncStatus.isInitialSyncDone());
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
- FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
assertFalse(syncStatus.isInitialSyncDone());
* with a commitIndex that is greater than what has been applied to the
* state machine of the RaftActor, the RaftActor applies the state and
* sets it current applied state to the commitIndex of the sender.
- *
- * @throws Exception
*/
@Test
public void testHandleAppendEntriesWithNewerCommitIndex() throws Exception {
* This test verifies that when an AppendEntries is received a specific prevLogTerm
* which does not match the term that is in RaftActors log entry at prevLogIndex
* then the RaftActor does not change it's state and it returns a failure.
- *
- * @throws Exception
*/
@Test
public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
* This test verifies that when a new AppendEntries message is received with
* new entries and the logs of the sender and receiver match that the new
* entries get added to the log and the log is incremented by the number of
- * entries received in appendEntries
- *
- * @throws Exception
+ * entries received in appendEntries.
*/
@Test
public void testHandleAppendEntriesAddNewEntries() {
* This test verifies that when a new AppendEntries message is received with
* new entries and the logs of the sender and receiver are out-of-sync that
* the log is first corrected by removing the out of sync entries from the
- * log and then adding in the new entries sent with the AppendEntries message
+ * log and then adding in the new entries sent with the AppendEntries message.
*/
@Test
public void testHandleAppendEntriesCorrectReceiverLogEntries() {
}
@Test
- public void testHandleAppendEntriesPreviousLogEntryMissing(){
+ public void testHandleAppendEntriesPreviousLogEntryMissing() {
logStart("testHandleAppendEntriesPreviousLogEntryMissing");
- MockRaftActorContext context = createActorContext();
+ final MockRaftActorContext context = createActorContext();
// Prepare the receivers log
MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
}
@Test
- public void testHandleAppendEntriesAfterInstallingSnapshot(){
+ public void testHandleAppendEntriesAfterInstallingSnapshot() {
logStart("testHandleAppendAfterInstallingSnapshot");
MockRaftActorContext context = createActorContext();
/**
* This test verifies that when InstallSnapshot is received by
* the follower its applied correctly.
- *
- * @throws Exception
*/
@Test
public void testHandleInstallSnapshot() throws Exception {
int offset = 0;
int snapshotLength = bsSnapshot.size();
int chunkSize = 50;
- int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
int lastIncludedIndex = 1;
int chunkIndex = 1;
InstallSnapshot lastInstallSnapshot = null;
- for(int i = 0; i < totalChunks; i++) {
+ for (int i = 0; i < totalChunks; i++) {
byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
chunkData, chunkIndex, totalChunks);
assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
chunkIndex = 1;
- for(InstallSnapshotReply reply: replies) {
+ for (InstallSnapshotReply reply: replies) {
assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
assertEquals("getTerm", 1, reply.getTerm());
assertEquals("isSuccess", true, reply.isSuccess());
assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
}
- assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
}
/**
* Verify that when an AppendEntries is sent to a follower during a snapshot install
* the Follower short-circuits the processing of the AppendEntries message.
- *
- * @throws Exception
*/
@Test
public void testReceivingAppendEntriesDuringInstallSnapshot() throws Exception {
ByteString bsSnapshot = createSnapshot();
int snapshotLength = bsSnapshot.size();
int chunkSize = 50;
- int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
int lastIncludedIndex = 1;
// Check that snapshot installation is not in progress
- assertNull(((Follower) follower).getSnapshotTracker());
+ assertNull(follower.getSnapshotTracker());
// Make sure that we have more than 1 chunk to send
assertTrue(totalChunks > 1);
chunkData, 1, totalChunks));
// Check if snapshot installation is in progress now
- assertNotNull(((Follower) follower).getSnapshotTracker());
+ assertNotNull(follower.getSnapshotTracker());
// Send an append entry
- AppendEntries appendEntries = mock(AppendEntries.class);
- doReturn(context.getTermInformation().getCurrentTerm()).when(appendEntries).getTerm();
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 1, "3")), 2, -1, (short)1);
follower.handleMessage(leaderActor, appendEntries);
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
- assertEquals(context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
- assertEquals(context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
- assertEquals(context.getTermInformation().getCurrentTerm(), reply.getTerm());
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
+ assertEquals("getTerm", context.getTermInformation().getCurrentTerm(), reply.getTerm());
+
+ assertNotNull(follower.getSnapshotTracker());
+ }
+
+ @Test
+ public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() throws Exception {
+ logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
- // We should not hit the code that needs to look at prevLogIndex because we are short circuiting
- verify(appendEntries, never()).getPrevLogIndex();
+ // Check that snapshot installation is not in progress
+ assertNull(follower.getSnapshotTracker());
+
+ // Make sure that we have more than 1 chunk to send
+ assertTrue(totalChunks > 1);
+
+ // Send an install snapshot with the first chunk to start the process of installing a snapshot
+ byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, 1, totalChunks));
+
+ // Check if snapshot installation is in progress now
+ assertNotNull(follower.getSnapshotTracker());
+
+ // Send appendEntries with a new term and leader.
+ AppendEntries appendEntries = new AppendEntries(2, "new-leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 2, "3")), 2, -1, (short)1);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", 2, reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", 2, reply.getLogLastTerm());
+ assertEquals("getTerm", 2, reply.getTerm());
+
+ assertNull(follower.getSnapshotTracker());
}
@Test
logStart("testInitialSyncUpWithHandleInstallSnapshot");
MockRaftActorContext context = createActorContext();
+ context.setCommitIndex(-1);
follower = createBehavior(context);
int offset = 0;
int snapshotLength = bsSnapshot.size();
int chunkSize = 50;
- int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
int lastIncludedIndex = 1;
int chunkIndex = 1;
InstallSnapshot lastInstallSnapshot = null;
- for(int i = 0; i < totalChunks; i++) {
+ for (int i = 0; i < totalChunks; i++) {
byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
chunkData, chunkIndex, totalChunks);
assertEquals("getTerm", 1, reply.getTerm());
assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
- assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
}
@Test
- public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers(){
+ public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() {
MockRaftActorContext context = createActorContext();
Stopwatch stopwatch = Stopwatch.createStarted();
follower = createBehavior(context);
- MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
+ TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
+
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
+ assertTrue("Expected Candidate", newBehavior instanceof Candidate);
}
@Test
- public void testFollowerDoesNotScheduleAnElectionIfAutomaticElectionsAreDisabled(){
+ public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() {
MockRaftActorContext context = createActorContext();
- context.setConfigParams(new DefaultConfigParamsImpl(){
+ context.setConfigParams(new DefaultConfigParamsImpl() {
@Override
public FiniteDuration getElectionTimeOutInterval() {
return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
follower = createBehavior(context);
- MessageCollectorActor.assertNoneMatching(followerActor, ElectionTimeout.class, 500);
+ TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
+ assertSame("handleMessage result", follower, newBehavior);
+ }
+
+ @Test
+ public void testFollowerSchedulesElectionIfNonVoting() {
+ MockRaftActorContext context = createActorContext();
+ context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo(context.getId(), false))));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setHeartBeatInterval(
+ FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setElectionTimeoutFactor(1);
+
+ follower = new Follower(context, "leader", (short)1);
+
+ ElectionTimeout electionTimeout = MessageCollectorActor.expectFirstMatching(followerActor,
+ ElectionTimeout.class);
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), electionTimeout);
+ assertSame("handleMessage result", follower, newBehavior);
+ assertNull("Expected null leaderId", follower.getLeaderId());
}
@Test
- public void testElectionScheduledWhenAnyRaftRPCReceived(){
+ public void testElectionScheduledWhenAnyRaftRPCReceived() {
MockRaftActorContext context = createActorContext();
follower = createBehavior(context);
follower.handleMessage(leaderActor, new RaftRPC() {
return 100;
}
});
- assertEquals("schedule election", 1, getElectionTimeoutCount(follower));
+ verify(follower).scheduleElection(any(FiniteDuration.class));
}
@Test
- public void testElectionNotScheduledWhenNonRaftRPCMessageReceived(){
+ public void testElectionNotScheduledWhenNonRaftRPCMessageReceived() {
MockRaftActorContext context = createActorContext();
follower = createBehavior(context);
follower.handleMessage(leaderActor, "non-raft-rpc");
- assertEquals("schedule election", 0, getElectionTimeoutCount(follower));
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
}
- public byte[] getNextChunk (ByteString bs, int offset, int chunkSize){
+ public byte[] getNextChunk(ByteString bs, int offset, int chunkSize) {
int snapshotLength = bs.size();
int start = offset;
int size = chunkSize;
if (chunkSize > snapshotLength) {
size = snapshotLength;
} else {
- if ((start + chunkSize) > snapshotLength) {
+ if (start + chunkSize > snapshotLength) {
size = snapshotLength - start;
}
}
private static ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
- return new MockRaftActorContext.MockReplicatedLogEntry(term, index,
+ return new SimpleReplicatedLogEntry(index, term,
new MockRaftActorContext.MockPayload(data));
}
- private ByteString createSnapshot(){
+ private ByteString createSnapshot() {
HashMap<String, String> followerSnapshot = new HashMap<>();
followerSnapshot.put("1", "A");
followerSnapshot.put("2", "B");
}
@Override
- protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext,
+ protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
ActorRef actorRef, RaftRPC rpc) throws Exception {
super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
}
@Override
- protected void handleAppendEntriesAddSameEntryToLogReply(TestActorRef<MessageCollectorActor> replyActor)
+ protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor)
throws Exception {
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
assertEquals("isSuccess", true, reply.isSuccess());
}
-
- private static class TestFollower extends Follower {
-
- int electionTimeoutCount = 0;
-
- public TestFollower(RaftActorContext context) {
- super(context);
- }
-
- @Override
- protected void scheduleElection(FiniteDuration interval) {
- electionTimeoutCount++;
- super.scheduleElection(interval);
- }
-
- public int getElectionTimeoutCount() {
- return electionTimeoutCount;
- }
- }
}
Code Review / controller.git / blobdiff