logStart("testThatLeaderSendsAHeartbeatMessageToAllFollowers");
MockRaftActorContext actorContext = createActorContextWithFollower();
+ short payloadVersion = (short)5;
+ actorContext.setPayloadVersion(payloadVersion);
long term = 1;
actorContext.getTermInformation().update(term, "");
assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
assertEquals("getPrevLogTerm", -1, appendEntries.getPrevLogTerm());
assertEquals("Entries size", 0, appendEntries.getEntries().size());
+ assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
// The follower would normally reply - simulate that explicitly here.
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex - 1, term));
+ FOLLOWER_ID, term, true, lastIndex - 1, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
assertEquals("Entries size", 1, appendEntries.getEntries().size());
assertEquals("Entry getIndex", lastIndex, appendEntries.getEntries().get(0).getIndex());
assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
+ assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
}
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
for(int i=0;i<3;i++) {
sendReplicate(actorContext, lastIndex+i+1);
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex + i + 1, term));
+ FOLLOWER_ID, term, true, lastIndex + i + 1, term, (short)0));
}
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
// The follower would normally reply - simulate that explicitly here.
long lastIndex = actorContext.getReplicatedLog().lastIndex();
leader.handleMessage(followerActor, new AppendEntriesReply(
- FOLLOWER_ID, term, true, lastIndex, term));
+ FOLLOWER_ID, term, true, lastIndex, term, (short)0));
assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
followerActor.underlyingActor().clear();
assertTrue(raftBehavior instanceof Leader);
- MessageCollectorActor.expectFirstMatching(leaderActor, CaptureSnapshot.class);
+ assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
}
@Test
leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
- CaptureSnapshot cs = MessageCollectorActor.expectFirstMatching(leaderActor, CaptureSnapshot.class);
+ assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
+
+ CaptureSnapshot cs = actorContext.getSnapshotManager().getCaptureSnapshot();
assertTrue(cs.isInstallSnapshotInitiated());
assertEquals(3, cs.getLastAppliedIndex());
// if an initiate is started again when first is in progress, it shouldnt initiate Capture
leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
- List<CaptureSnapshot> captureSnapshots = MessageCollectorActor.getAllMatching(leaderActor, CaptureSnapshot.class);
- assertEquals("CaptureSnapshot should not get invoked when initiate is in progress", 1, captureSnapshots.size());
+ Assert.assertSame("CaptureSnapshot instance", cs, actorContext.getSnapshotManager().getCaptureSnapshot());
}
@Test
leader = new Leader(leaderActorContext);
// Send initial heartbeat reply with last index.
- leader.handleAppendEntriesReply(followerActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 10, 1));
+ leader.handleAppendEntriesReply(followerActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 10, 1, (short)0));
FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
assertEquals("getNextIndex", 11, followerInfo.getNextIndex());
- AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, false, 10, 1);
+ AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, false, 10, 1, (short)0);
RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
leader = new Leader(leaderActorContext);
- AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1);
+ short payloadVersion = 5;
+ AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1, payloadVersion);
RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
ApplyState applyState = applyStateList.get(0);
assertEquals(2, applyState.getReplicatedLogEntry().getIndex());
+
+ FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
+ assertEquals(payloadVersion, followerInfo.getPayloadVersion());
}
@Test
leader = new Leader(leaderActorContext);
- AppendEntriesReply reply = new AppendEntriesReply("unkown-follower", 1, false, 10, 1);
+ AppendEntriesReply reply = new AppendEntriesReply("unkown-follower", 1, false, 10, 1, (short)0);
RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
for(int i=1;i<6;i++) {
// Each AppendEntriesReply could end up rescheduling the heartbeat (without the fix for bug 2733)
- RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor, new AppendEntriesReply(follower1ActorId, 1, true, i, 1));
+ RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor, new AppendEntriesReply(follower1ActorId, 1, true, i, 1, (short)0));
assertTrue(newBehavior == leader);
Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
}