import org.opendaylight.controller.cluster.NonPersistentDataProvider;
import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
import org.opendaylight.controller.cluster.notifications.RoleChanged;
-import org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
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.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.Leader;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
UpdateElectionTerm updateElectionTerm = new UpdateElectionTerm(5, "member2");
mockRaftActor.handleRecover(updateElectionTerm);
+ org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm deprecatedUpdateElectionTerm =
+ 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(deleteEntries));
verify(mockSupport).handleRecoveryMessage(same(deprecatedDeleteEntries));
verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm));
+ verify(mockSupport).handleRecoveryMessage(same(deprecatedUpdateElectionTerm));
}
@Test
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;
}
};
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);
}};
}
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
//fake snapshot on index 5
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1, (short)0));
assertEquals(8, leaderActor.getReplicatedLog().size());
//fake snapshot on index 6
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1, (short)0));
assertEquals(8, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
assertFalse(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());
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(follower1Id, 1, true, 7, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 7, 1, (short)0));
assertEquals(2, leaderActor.getReplicatedLog().size());
assertEquals(8, leaderActor.getReplicatedLog().lastIndex());
(ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
new MockRaftActorContext.MockPayload("foo-6"))
);
- followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5, (short)0));
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, leaderId, 6, 1, entries, 6, 6));
+ followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6, (short)0));
assertEquals(8, followerActor.getReplicatedLog().size());
assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
assertFalse(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));
+ 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));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1, (short)0));
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
// set the 2nd follower nextIndex to 1 which has been snapshotted
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1, (short)0));
assertEquals(5, leaderActor.getReplicatedLog().size());
assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
//reply from a slow follower does not initiate a fake snapshot
- leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1, (short)0));
assertEquals("Fake snapshot should not happen when Initiate is in progress", 5, leaderActor.getReplicatedLog().size());
ByteString snapshotBytes = fromObject(Arrays.asList(
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));
+ leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1, (short)0));
assertEquals(0, leaderActor.getReplicatedLog().size());
}
};
DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
- Map<String, String> peerAddresses = new HashMap<>();
+ Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
MockRaftActor.props(persistenceId, peerAddresses,
DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
- Map<String, String> peerAddresses = new HashMap<>();
+ Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
MockRaftActor.props(persistenceId, peerAddresses,