import akka.actor.ActorRef;
import akka.actor.Props;
import akka.testkit.TestActorRef;
+import com.google.common.base.Stopwatch;
import com.google.protobuf.ByteString;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
private RaftActorBehavior follower;
+ private final short payloadVersion = 5;
+
@Override
@After
public void tearDown() throws Exception {
@Override
protected MockRaftActorContext createActorContext(ActorRef actorRef){
- return new MockRaftActorContext("follower", getSystem(), actorRef);
+ MockRaftActorContext context = new MockRaftActorContext("follower", getSystem(), actorRef);
+ context.setPayloadVersion(payloadVersion );
+ return context;
}
@Test
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101);
+ appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
follower.handleMessage(leaderActor, appendEntries);
syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
newReplicatedLogEntry(2, 101, "foo"));
// leader-2 is becoming the leader now and it says the commitIndex is 45
- appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100);
+ appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
follower.handleMessage(leaderActor, appendEntries);
syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101);
+ appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
follower.handleMessage(leaderActor, appendEntries);
syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
newReplicatedLogEntry(2, 101, "foo"));
// leader-2 is becoming the leader now and it says the commitIndex is 45
- appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100);
+ appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
follower.handleMessage(leaderActor, appendEntries);
syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
follower = createBehavior(context);
follower.handleMessage(leaderActor, appendEntries);
// AppendEntries is now sent with a bigger term
// this will set the receivers term to be the same as the sender's term
- AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, null, 101, -1);
+ AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, null, 101, -1, (short)0);
follower = createBehavior(context);
// before the new behavior was created (1 in this case)
// This will not work for a Candidate because as soon as a Candidate
// is created it increments the term
- AppendEntries appendEntries = new AppendEntries(1, "leader-1", 2, 1, entries, 4, -1);
+ AppendEntries appendEntries = new AppendEntries(1, "leader-1", 2, 1, entries, 4, -1, (short)0);
follower = createBehavior(context);
// before the new behavior was created (1 in this case)
// This will not work for a Candidate because as soon as a Candidate
// is created it increments the term
- AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1);
+ AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0);
follower = createBehavior(context);
List<ReplicatedLogEntry> entries = new ArrayList<>();
entries.add(newReplicatedLogEntry(1, 4, "four"));
- AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1);
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0);
follower = createBehavior(context);
follower = createBehavior(context);
- follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1));
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0));
assertEquals("Next index", 2, log.last().getIndex() + 1);
assertEquals("Entry 1", entries.get(0), log.get(1));
entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
leaderActor.underlyingActor().clear();
- follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1));
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0));
assertEquals("Next index", 3, log.last().getIndex() + 1);
assertEquals("Entry 1", entries.get(0), log.get(1));
List<ReplicatedLogEntry> entries = new ArrayList<>();
entries.add(newReplicatedLogEntry(1, 4, "four"));
- AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3);
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0);
follower = createBehavior(context);
ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
ApplySnapshot.class);
Snapshot snapshot = applySnapshot.getSnapshot();
+ assertNotNull(lastInstallSnapshot);
assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
snapshot.getLastAppliedTerm());
newReplicatedLogEntry(2, 101, "foo"));
// The new commitIndex is 101
- AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101);
+ AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101, (short)0);
follower.handleMessage(leaderActor, appendEntries);
syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
}
+ @Test
+ public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers(){
+ MockRaftActorContext context = createActorContext();
+
+ Stopwatch stopwatch = Stopwatch.createStarted();
+
+ follower = createBehavior(context);
+
+ MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
+
+ long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
+
+ assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
+ }
+
public ByteString getNextChunk (ByteString bs, int offset, int chunkSize){
int snapshotLength = bs.size();
int start = offset;
assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
+ assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
}
private ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {