2 * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.raft;
11 import static org.junit.Assert.assertArrayEquals;
12 import static org.junit.Assert.assertEquals;
13 import static org.junit.Assert.assertNotNull;
14 import static org.junit.Assert.assertNotSame;
15 import static org.junit.Assert.assertNull;
16 import static org.junit.Assert.assertSame;
17 import static org.junit.Assert.assertTrue;
18 import static org.mockito.Matchers.any;
19 import static org.mockito.Matchers.anyObject;
20 import static org.mockito.Matchers.eq;
21 import static org.mockito.Matchers.same;
22 import static org.mockito.Mockito.doReturn;
23 import static org.mockito.Mockito.mock;
24 import static org.mockito.Mockito.never;
25 import static org.mockito.Mockito.reset;
26 import static org.mockito.Mockito.timeout;
27 import static org.mockito.Mockito.verify;
29 import akka.actor.ActorRef;
30 import akka.actor.PoisonPill;
31 import akka.actor.Props;
32 import akka.actor.Status.Failure;
33 import akka.actor.Terminated;
34 import akka.dispatch.Dispatchers;
35 import akka.japi.Procedure;
36 import akka.persistence.SaveSnapshotFailure;
37 import akka.persistence.SaveSnapshotSuccess;
38 import akka.persistence.SnapshotMetadata;
39 import akka.persistence.SnapshotOffer;
40 import akka.testkit.JavaTestKit;
41 import akka.testkit.TestActorRef;
42 import com.google.common.base.Optional;
43 import com.google.common.collect.ImmutableMap;
44 import com.google.common.util.concurrent.Uninterruptibles;
45 import com.google.protobuf.ByteString;
46 import java.io.ByteArrayOutputStream;
47 import java.io.ObjectOutputStream;
48 import java.util.ArrayList;
49 import java.util.Arrays;
50 import java.util.Collections;
51 import java.util.HashMap;
52 import java.util.List;
54 import java.util.concurrent.TimeUnit;
55 import java.util.concurrent.TimeoutException;
56 import org.apache.commons.lang3.SerializationUtils;
57 import org.junit.After;
58 import org.junit.Before;
59 import org.junit.Test;
60 import org.mockito.ArgumentCaptor;
61 import org.opendaylight.controller.cluster.DataPersistenceProvider;
62 import org.opendaylight.controller.cluster.NonPersistentDataProvider;
63 import org.opendaylight.controller.cluster.PersistentDataProvider;
64 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
65 import org.opendaylight.controller.cluster.notifications.RoleChanged;
66 import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
67 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
68 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
69 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
70 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
71 import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning;
72 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
73 import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
74 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
75 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
76 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
77 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
78 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshotReply;
79 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
80 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
81 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
82 import org.opendaylight.controller.cluster.raft.persisted.DeleteEntries;
83 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
84 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
85 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
86 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
87 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
88 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
89 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
90 import org.opendaylight.yangtools.concepts.Identifier;
91 import org.slf4j.Logger;
92 import org.slf4j.LoggerFactory;
93 import scala.concurrent.duration.Duration;
94 import scala.concurrent.duration.FiniteDuration;
96 public class RaftActorTest extends AbstractActorTest {
98 static final Logger TEST_LOG = LoggerFactory.getLogger(RaftActorTest.class);
100 private TestActorFactory factory;
103 public void setUp() {
104 factory = new TestActorFactory(getSystem());
108 public void tearDown() throws Exception {
110 InMemoryJournal.clear();
111 InMemorySnapshotStore.clear();
115 public void testConstruction() {
116 new RaftActorTestKit(getSystem(), "testConstruction").waitUntilLeader();
120 public void testFindLeaderWhenLeaderIsSelf() {
121 RaftActorTestKit kit = new RaftActorTestKit(getSystem(), "testFindLeader");
122 kit.waitUntilLeader();
127 public void testRaftActorRecoveryWithPersistenceEnabled() throws Exception {
128 TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled starting");
130 JavaTestKit kit = new JavaTestKit(getSystem());
131 String persistenceId = factory.generateActorId("follower-");
133 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
135 // Set the heartbeat interval high to essentially disable election otherwise the test
136 // may fail if the actor is switched to Leader and the commitIndex is set to the last
138 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
140 ImmutableMap<String, String> peerAddresses = ImmutableMap.<String, String>builder()
141 .put("member1", "address").build();
142 ActorRef followerActor = factory.createActor(MockRaftActor.props(persistenceId,
143 peerAddresses, config), persistenceId);
145 kit.watch(followerActor);
147 List<ReplicatedLogEntry> snapshotUnappliedEntries = new ArrayList<>();
148 ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 4,
149 new MockRaftActorContext.MockPayload("E"));
150 snapshotUnappliedEntries.add(entry1);
152 int lastAppliedDuringSnapshotCapture = 3;
153 int lastIndexDuringSnapshotCapture = 4;
155 // 4 messages as part of snapshot, which are applied to state
156 ByteString snapshotBytes = fromObject(Arrays.asList(
157 new MockRaftActorContext.MockPayload("A"),
158 new MockRaftActorContext.MockPayload("B"),
159 new MockRaftActorContext.MockPayload("C"),
160 new MockRaftActorContext.MockPayload("D")));
162 Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
163 snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1,
164 lastAppliedDuringSnapshotCapture, 1);
165 InMemorySnapshotStore.addSnapshot(persistenceId, snapshot);
167 // add more entries after snapshot is taken
168 List<ReplicatedLogEntry> entries = new ArrayList<>();
169 ReplicatedLogEntry entry2 = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
170 new MockRaftActorContext.MockPayload("F", 2));
171 ReplicatedLogEntry entry3 = new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
172 new MockRaftActorContext.MockPayload("G", 3));
173 ReplicatedLogEntry entry4 = new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
174 new MockRaftActorContext.MockPayload("H", 4));
179 final int lastAppliedToState = 5;
180 final int lastIndex = 7;
182 InMemoryJournal.addEntry(persistenceId, 5, entry2);
183 // 2 entries are applied to state besides the 4 entries in snapshot
184 InMemoryJournal.addEntry(persistenceId, 6, new ApplyJournalEntries(lastAppliedToState));
185 InMemoryJournal.addEntry(persistenceId, 7, entry3);
186 InMemoryJournal.addEntry(persistenceId, 8, entry4);
189 followerActor.tell(PoisonPill.getInstance(), null);
190 kit.expectMsgClass(JavaTestKit.duration("5 seconds"), Terminated.class);
192 kit.unwatch(followerActor);
194 //reinstate the actor
195 TestActorRef<MockRaftActor> ref = factory.createTestActor(
196 MockRaftActor.props(persistenceId, peerAddresses, config));
198 MockRaftActor mockRaftActor = ref.underlyingActor();
200 mockRaftActor.waitForRecoveryComplete();
202 RaftActorContext context = mockRaftActor.getRaftActorContext();
203 assertEquals("Journal log size", snapshotUnappliedEntries.size() + entries.size(),
204 context.getReplicatedLog().size());
205 assertEquals("Journal data size", 10, context.getReplicatedLog().dataSize());
206 assertEquals("Last index", lastIndex, context.getReplicatedLog().lastIndex());
207 assertEquals("Last applied", lastAppliedToState, context.getLastApplied());
208 assertEquals("Commit index", lastAppliedToState, context.getCommitIndex());
209 assertEquals("Recovered state size", 6, mockRaftActor.getState().size());
211 mockRaftActor.waitForInitializeBehaviorComplete();
213 assertEquals("getRaftState", RaftState.Follower, mockRaftActor.getRaftState());
215 TEST_LOG.info("testRaftActorRecoveryWithPersistenceEnabled ending");
219 public void testRaftActorRecoveryWithPersistenceDisabled() throws Exception {
220 String persistenceId = factory.generateActorId("follower-");
222 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
224 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
226 TestActorRef<MockRaftActor> ref = factory.createTestActor(MockRaftActor.props(persistenceId,
227 ImmutableMap.<String, String>builder().put("member1", "address").build(),
228 config, new NonPersistentDataProvider()), persistenceId);
230 MockRaftActor mockRaftActor = ref.underlyingActor();
232 mockRaftActor.waitForRecoveryComplete();
234 mockRaftActor.waitForInitializeBehaviorComplete();
236 assertEquals("getRaftState", RaftState.Follower, mockRaftActor.getRaftState());
240 public void testUpdateElectionTermPersistedWithPersistenceDisabled() throws Exception {
241 final JavaTestKit kit = new JavaTestKit(getSystem());
242 String persistenceId = factory.generateActorId("follower-");
243 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
244 config.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
245 config.setElectionTimeoutFactor(1);
247 InMemoryJournal.addWriteMessagesCompleteLatch(persistenceId, 1);
249 TestActorRef<MockRaftActor> ref = factory.createTestActor(MockRaftActor.props(persistenceId,
250 ImmutableMap.<String, String>builder().put("member1", "address").build(),
251 config, new NonPersistentDataProvider())
252 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
254 InMemoryJournal.waitForWriteMessagesComplete(persistenceId);
255 List<UpdateElectionTerm> entries = InMemoryJournal.get(persistenceId, UpdateElectionTerm.class);
256 assertEquals("UpdateElectionTerm entries", 1, entries.size());
257 final UpdateElectionTerm updateEntry = entries.get(0);
259 factory.killActor(ref, kit);
261 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
262 ref = factory.createTestActor(MockRaftActor.props(persistenceId,
263 ImmutableMap.<String, String>builder().put("member1", "address").build(), config,
264 new NonPersistentDataProvider()).withDispatcher(Dispatchers.DefaultDispatcherId()),
265 factory.generateActorId("follower-"));
267 MockRaftActor actor = ref.underlyingActor();
268 actor.waitForRecoveryComplete();
270 RaftActorContext newContext = actor.getRaftActorContext();
271 assertEquals("electionTerm", updateEntry.getCurrentTerm(),
272 newContext.getTermInformation().getCurrentTerm());
273 assertEquals("votedFor", updateEntry.getVotedFor(), newContext.getTermInformation().getVotedFor());
275 entries = InMemoryJournal.get(persistenceId, UpdateElectionTerm.class);
276 assertEquals("UpdateElectionTerm entries", 1, entries.size());
280 public void testRaftActorForwardsToRaftActorRecoverySupport() {
281 String persistenceId = factory.generateActorId("leader-");
283 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
285 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
287 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
288 Collections.<String, String>emptyMap(), config), persistenceId);
290 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
292 // Wait for akka's recovery to complete so it doesn't interfere.
293 mockRaftActor.waitForRecoveryComplete();
295 RaftActorRecoverySupport mockSupport = mock(RaftActorRecoverySupport.class);
296 mockRaftActor.setRaftActorRecoverySupport(mockSupport );
298 Snapshot snapshot = Snapshot.create(new byte[]{1}, Collections.<ReplicatedLogEntry>emptyList(), 3, 1, 3, 1);
299 SnapshotOffer snapshotOffer = new SnapshotOffer(new SnapshotMetadata("test", 6, 12345), snapshot);
300 mockRaftActor.handleRecover(snapshotOffer);
302 MockRaftActorContext.MockReplicatedLogEntry logEntry = new MockRaftActorContext.MockReplicatedLogEntry(1,
303 1, new MockRaftActorContext.MockPayload("1", 5));
304 mockRaftActor.handleRecover(logEntry);
306 ApplyJournalEntries applyJournalEntries = new ApplyJournalEntries(2);
307 mockRaftActor.handleRecover(applyJournalEntries);
309 DeleteEntries deleteEntries = new DeleteEntries(1);
310 mockRaftActor.handleRecover(deleteEntries);
312 UpdateElectionTerm updateElectionTerm = new UpdateElectionTerm(5, "member2");
313 mockRaftActor.handleRecover(updateElectionTerm);
315 verify(mockSupport).handleRecoveryMessage(same(snapshotOffer), any(PersistentDataProvider.class));
316 verify(mockSupport).handleRecoveryMessage(same(logEntry), any(PersistentDataProvider.class));
317 verify(mockSupport).handleRecoveryMessage(same(applyJournalEntries), any(PersistentDataProvider.class));
318 verify(mockSupport).handleRecoveryMessage(same(deleteEntries), any(PersistentDataProvider.class));
319 verify(mockSupport).handleRecoveryMessage(same(updateElectionTerm), any(PersistentDataProvider.class));
323 public void testRaftActorForwardsToRaftActorSnapshotMessageSupport() {
324 String persistenceId = factory.generateActorId("leader-");
326 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
328 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
330 RaftActorSnapshotMessageSupport mockSupport = mock(RaftActorSnapshotMessageSupport.class);
332 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
333 .config(config).snapshotMessageSupport(mockSupport).props());
335 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
337 // Wait for akka's recovery to complete so it doesn't interfere.
338 mockRaftActor.waitForRecoveryComplete();
340 ApplySnapshot applySnapshot = new ApplySnapshot(mock(Snapshot.class));
341 doReturn(true).when(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class));
342 mockRaftActor.handleCommand(applySnapshot);
344 CaptureSnapshot captureSnapshot = new CaptureSnapshot(1, 1, 1, 1, 0, 1, null);
345 doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshot), any(ActorRef.class));
346 mockRaftActor.handleCommand(captureSnapshot);
348 CaptureSnapshotReply captureSnapshotReply = new CaptureSnapshotReply(new byte[0]);
349 doReturn(true).when(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class));
350 mockRaftActor.handleCommand(captureSnapshotReply);
352 SaveSnapshotSuccess saveSnapshotSuccess = new SaveSnapshotSuccess(new SnapshotMetadata("", 0L, 0L));
353 doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class));
354 mockRaftActor.handleCommand(saveSnapshotSuccess);
356 SaveSnapshotFailure saveSnapshotFailure = new SaveSnapshotFailure(new SnapshotMetadata("", 0L, 0L),
358 doReturn(true).when(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class));
359 mockRaftActor.handleCommand(saveSnapshotFailure);
361 doReturn(true).when(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT),
362 any(ActorRef.class));
363 mockRaftActor.handleCommand(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT);
365 doReturn(true).when(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class));
366 mockRaftActor.handleCommand(GetSnapshot.INSTANCE);
368 verify(mockSupport).handleSnapshotMessage(same(applySnapshot), any(ActorRef.class));
369 verify(mockSupport).handleSnapshotMessage(same(captureSnapshot), any(ActorRef.class));
370 verify(mockSupport).handleSnapshotMessage(same(captureSnapshotReply), any(ActorRef.class));
371 verify(mockSupport).handleSnapshotMessage(same(saveSnapshotSuccess), any(ActorRef.class));
372 verify(mockSupport).handleSnapshotMessage(same(saveSnapshotFailure), any(ActorRef.class));
373 verify(mockSupport).handleSnapshotMessage(same(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT),
374 any(ActorRef.class));
375 verify(mockSupport).handleSnapshotMessage(same(GetSnapshot.INSTANCE), any(ActorRef.class));
378 @SuppressWarnings("unchecked")
380 public void testApplyJournalEntriesCallsDataPersistence() throws Exception {
381 String persistenceId = factory.generateActorId("leader-");
383 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
385 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
387 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
389 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
390 Collections.<String, String>emptyMap(), config, dataPersistenceProvider), persistenceId);
392 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
394 mockRaftActor.waitForInitializeBehaviorComplete();
396 mockRaftActor.waitUntilLeader();
398 mockRaftActor.onReceiveCommand(new ApplyJournalEntries(10));
400 verify(dataPersistenceProvider).persist(any(ApplyJournalEntries.class), any(Procedure.class));
404 public void testApplyState() throws Exception {
405 String persistenceId = factory.generateActorId("leader-");
407 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
409 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
411 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
413 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
414 Collections.<String, String>emptyMap(), config, dataPersistenceProvider), persistenceId);
416 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
418 mockRaftActor.waitForInitializeBehaviorComplete();
420 ReplicatedLogEntry entry = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
421 new MockRaftActorContext.MockPayload("F"));
423 final Identifier id = new MockIdentifier("apply-state");
424 mockRaftActor.onReceiveCommand(new ApplyState(mockActorRef, id, entry));
426 verify(mockRaftActor.actorDelegate).applyState(eq(mockActorRef), eq(id), anyObject());
430 public void testRaftRoleChangeNotifierWhenRaftActorHasNoPeers() throws Exception {
431 TestActorRef<MessageCollectorActor> notifierActor = factory.createTestActor(
432 Props.create(MessageCollectorActor.class));
433 MessageCollectorActor.waitUntilReady(notifierActor);
435 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
436 long heartBeatInterval = 100;
437 config.setHeartBeatInterval(FiniteDuration.create(heartBeatInterval, TimeUnit.MILLISECONDS));
438 config.setElectionTimeoutFactor(20);
440 String persistenceId = factory.generateActorId("notifier-");
442 final TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder()
443 .id(persistenceId).config(config).roleChangeNotifier(notifierActor).dataPersistenceProvider(
444 new NonPersistentDataProvider()).props().withDispatcher(Dispatchers.DefaultDispatcherId()),
447 List<RoleChanged> matches = MessageCollectorActor.expectMatching(notifierActor, RoleChanged.class, 3);
450 // check if the notifier got a role change from null to Follower
451 RoleChanged raftRoleChanged = matches.get(0);
452 assertEquals(persistenceId, raftRoleChanged.getMemberId());
453 assertNull(raftRoleChanged.getOldRole());
454 assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole());
456 // check if the notifier got a role change from Follower to Candidate
457 raftRoleChanged = matches.get(1);
458 assertEquals(persistenceId, raftRoleChanged.getMemberId());
459 assertEquals(RaftState.Follower.name(), raftRoleChanged.getOldRole());
460 assertEquals(RaftState.Candidate.name(), raftRoleChanged.getNewRole());
462 // check if the notifier got a role change from Candidate to Leader
463 raftRoleChanged = matches.get(2);
464 assertEquals(persistenceId, raftRoleChanged.getMemberId());
465 assertEquals(RaftState.Candidate.name(), raftRoleChanged.getOldRole());
466 assertEquals(RaftState.Leader.name(), raftRoleChanged.getNewRole());
468 LeaderStateChanged leaderStateChange = MessageCollectorActor.expectFirstMatching(
469 notifierActor, LeaderStateChanged.class);
471 assertEquals(raftRoleChanged.getMemberId(), leaderStateChange.getLeaderId());
472 assertEquals(MockRaftActor.PAYLOAD_VERSION, leaderStateChange.getLeaderPayloadVersion());
474 notifierActor.underlyingActor().clear();
476 MockRaftActor raftActor = raftActorRef.underlyingActor();
477 final String newLeaderId = "new-leader";
478 final short newLeaderVersion = 6;
479 Follower follower = new Follower(raftActor.getRaftActorContext()) {
481 public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
482 setLeaderId(newLeaderId);
483 setLeaderPayloadVersion(newLeaderVersion);
488 raftActor.newBehavior(follower);
490 leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
491 assertEquals(persistenceId, leaderStateChange.getMemberId());
492 assertEquals(null, leaderStateChange.getLeaderId());
494 raftRoleChanged = MessageCollectorActor.expectFirstMatching(notifierActor, RoleChanged.class);
495 assertEquals(RaftState.Leader.name(), raftRoleChanged.getOldRole());
496 assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole());
498 notifierActor.underlyingActor().clear();
500 raftActor.handleCommand("any");
502 leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
503 assertEquals(persistenceId, leaderStateChange.getMemberId());
504 assertEquals(newLeaderId, leaderStateChange.getLeaderId());
505 assertEquals(newLeaderVersion, leaderStateChange.getLeaderPayloadVersion());
507 notifierActor.underlyingActor().clear();
509 raftActor.handleCommand("any");
511 Uninterruptibles.sleepUninterruptibly(505, TimeUnit.MILLISECONDS);
512 leaderStateChange = MessageCollectorActor.getFirstMatching(notifierActor, LeaderStateChanged.class);
513 assertNull(leaderStateChange);
517 public void testRaftRoleChangeNotifierWhenRaftActorHasPeers() throws Exception {
518 ActorRef notifierActor = factory.createActor(Props.create(MessageCollectorActor.class));
519 MessageCollectorActor.waitUntilReady(notifierActor);
521 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
522 long heartBeatInterval = 100;
523 config.setHeartBeatInterval(FiniteDuration.create(heartBeatInterval, TimeUnit.MILLISECONDS));
524 config.setElectionTimeoutFactor(1);
526 String persistenceId = factory.generateActorId("notifier-");
528 factory.createActor(MockRaftActor.builder().id(persistenceId)
529 .peerAddresses(ImmutableMap.of("leader", "fake/path"))
530 .config(config).roleChangeNotifier(notifierActor).props());
532 List<RoleChanged> matches = null;
533 for (int i = 0; i < 5000 / heartBeatInterval; i++) {
534 matches = MessageCollectorActor.getAllMatching(notifierActor, RoleChanged.class);
535 assertNotNull(matches);
536 if (matches.size() == 3) {
539 Uninterruptibles.sleepUninterruptibly(heartBeatInterval, TimeUnit.MILLISECONDS);
542 assertNotNull(matches);
543 assertEquals(2, matches.size());
545 // check if the notifier got a role change from null to Follower
546 RoleChanged raftRoleChanged = matches.get(0);
547 assertEquals(persistenceId, raftRoleChanged.getMemberId());
548 assertNull(raftRoleChanged.getOldRole());
549 assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole());
551 // check if the notifier got a role change from Follower to Candidate
552 raftRoleChanged = matches.get(1);
553 assertEquals(persistenceId, raftRoleChanged.getMemberId());
554 assertEquals(RaftState.Follower.name(), raftRoleChanged.getOldRole());
555 assertEquals(RaftState.Candidate.name(), raftRoleChanged.getNewRole());
559 public void testFakeSnapshotsForLeaderWithInRealSnapshots() throws Exception {
560 final String persistenceId = factory.generateActorId("leader-");
561 final String follower1Id = factory.generateActorId("follower-");
563 ActorRef followerActor1 =
564 factory.createActor(Props.create(MessageCollectorActor.class));
566 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
567 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
568 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
570 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
572 Map<String, String> peerAddresses = new HashMap<>();
573 peerAddresses.put(follower1Id, followerActor1.path().toString());
575 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
576 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
578 MockRaftActor leaderActor = mockActorRef.underlyingActor();
580 leaderActor.getRaftActorContext().setCommitIndex(4);
581 leaderActor.getRaftActorContext().setLastApplied(4);
582 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
584 leaderActor.waitForInitializeBehaviorComplete();
586 // create 8 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
588 Leader leader = new Leader(leaderActor.getRaftActorContext());
589 leaderActor.setCurrentBehavior(leader);
590 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
592 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
593 leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 8, 1).build());
595 assertEquals(8, leaderActor.getReplicatedLog().size());
597 leaderActor.getRaftActorContext().getSnapshotManager().capture(
598 new MockRaftActorContext.MockReplicatedLogEntry(1, 6, new MockRaftActorContext.MockPayload("x")), 4);
600 verify(leaderActor.snapshotCohortDelegate).createSnapshot(any(ActorRef.class));
602 assertEquals(8, leaderActor.getReplicatedLog().size());
604 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
605 //fake snapshot on index 5
606 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1, (short)0));
608 assertEquals(8, leaderActor.getReplicatedLog().size());
610 //fake snapshot on index 6
611 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
612 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1, (short)0));
613 assertEquals(8, leaderActor.getReplicatedLog().size());
615 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
617 assertEquals(8, leaderActor.getReplicatedLog().size());
619 ByteString snapshotBytes = fromObject(Arrays.asList(
620 new MockRaftActorContext.MockPayload("foo-0"),
621 new MockRaftActorContext.MockPayload("foo-1"),
622 new MockRaftActorContext.MockPayload("foo-2"),
623 new MockRaftActorContext.MockPayload("foo-3"),
624 new MockRaftActorContext.MockPayload("foo-4")));
626 leaderActor.getRaftActorContext().getSnapshotManager().persist(snapshotBytes.toByteArray(),
627 Runtime.getRuntime().totalMemory());
629 assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
631 // The commit is needed to complete the snapshot creation process
632 leaderActor.getRaftActorContext().getSnapshotManager().commit(-1, -1);
634 // capture snapshot reply should remove the snapshotted entries only
635 assertEquals(3, leaderActor.getReplicatedLog().size());
636 assertEquals(7, leaderActor.getReplicatedLog().lastIndex());
638 // add another non-replicated entry
639 leaderActor.getReplicatedLog().append(
640 new ReplicatedLogImplEntry(8, 1, new MockRaftActorContext.MockPayload("foo-8")));
642 //fake snapshot on index 7, since lastApplied = 7 , we would keep the last applied
643 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 7, 1, (short)0));
644 assertEquals(2, leaderActor.getReplicatedLog().size());
645 assertEquals(8, leaderActor.getReplicatedLog().lastIndex());
649 public void testFakeSnapshotsForFollowerWithInRealSnapshots() throws Exception {
650 final String persistenceId = factory.generateActorId("follower-");
651 final String leaderId = factory.generateActorId("leader-");
654 ActorRef leaderActor1 =
655 factory.createActor(Props.create(MessageCollectorActor.class));
657 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
658 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
659 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
661 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
663 Map<String, String> peerAddresses = new HashMap<>();
664 peerAddresses.put(leaderId, leaderActor1.path().toString());
666 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
667 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
669 MockRaftActor followerActor = mockActorRef.underlyingActor();
670 followerActor.getRaftActorContext().setCommitIndex(4);
671 followerActor.getRaftActorContext().setLastApplied(4);
672 followerActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
674 followerActor.waitForInitializeBehaviorComplete();
677 Follower follower = new Follower(followerActor.getRaftActorContext());
678 followerActor.setCurrentBehavior(follower);
679 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
681 // create 6 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
682 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
683 followerActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 6, 1).build());
685 // log has indices 0-5
686 assertEquals(6, followerActor.getReplicatedLog().size());
689 followerActor.getRaftActorContext().getSnapshotManager().capture(
690 new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
691 new MockRaftActorContext.MockPayload("D")), 4);
693 verify(followerActor.snapshotCohortDelegate).createSnapshot(any(ActorRef.class));
695 assertEquals(6, followerActor.getReplicatedLog().size());
697 //fake snapshot on index 6
698 List<ReplicatedLogEntry> entries =
700 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
701 new MockRaftActorContext.MockPayload("foo-6"))
703 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5, (short)0));
704 assertEquals(7, followerActor.getReplicatedLog().size());
706 //fake snapshot on index 7
707 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
711 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
712 new MockRaftActorContext.MockPayload("foo-7"))
714 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6, (short) 0));
715 assertEquals(8, followerActor.getReplicatedLog().size());
717 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
720 ByteString snapshotBytes = fromObject(Arrays.asList(
721 new MockRaftActorContext.MockPayload("foo-0"),
722 new MockRaftActorContext.MockPayload("foo-1"),
723 new MockRaftActorContext.MockPayload("foo-2"),
724 new MockRaftActorContext.MockPayload("foo-3"),
725 new MockRaftActorContext.MockPayload("foo-4")));
726 followerActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
727 assertTrue(followerActor.getRaftActorContext().getSnapshotManager().isCapturing());
729 // The commit is needed to complete the snapshot creation process
730 followerActor.getRaftActorContext().getSnapshotManager().commit(-1, -1);
732 // capture snapshot reply should remove the snapshotted entries only till replicatedToAllIndex
733 assertEquals(3, followerActor.getReplicatedLog().size()); //indexes 5,6,7 left in the log
734 assertEquals(7, followerActor.getReplicatedLog().lastIndex());
738 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 8,
739 new MockRaftActorContext.MockPayload("foo-7"))
741 // send an additional entry 8 with leaderCommit = 7
742 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7, (short) 0));
744 // 7 and 8, as lastapplied is 7
745 assertEquals(2, followerActor.getReplicatedLog().size());
749 public void testFakeSnapshotsForLeaderWithInInitiateSnapshots() throws Exception {
750 final String persistenceId = factory.generateActorId("leader-");
751 final String follower1Id = factory.generateActorId("follower-");
752 final String follower2Id = factory.generateActorId("follower-");
754 final ActorRef followerActor1 = factory.createActor(Props.create(MessageCollectorActor.class), follower1Id);
755 final ActorRef followerActor2 = factory.createActor(Props.create(MessageCollectorActor.class), follower2Id);
757 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
758 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
759 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
761 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
763 Map<String, String> peerAddresses = new HashMap<>();
764 peerAddresses.put(follower1Id, followerActor1.path().toString());
765 peerAddresses.put(follower2Id, followerActor2.path().toString());
767 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
768 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
770 MockRaftActor leaderActor = mockActorRef.underlyingActor();
771 leaderActor.getRaftActorContext().setCommitIndex(9);
772 leaderActor.getRaftActorContext().setLastApplied(9);
773 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
775 leaderActor.waitForInitializeBehaviorComplete();
777 Leader leader = new Leader(leaderActor.getRaftActorContext());
778 leaderActor.setCurrentBehavior(leader);
779 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
781 // create 5 entries in the log
782 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
783 leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(5, 10, 1).build());
785 //set the snapshot index to 4 , 0 to 4 are snapshotted
786 leaderActor.getRaftActorContext().getReplicatedLog().setSnapshotIndex(4);
787 //setting replicatedToAllIndex = 9, for the log to clear
788 leader.setReplicatedToAllIndex(9);
789 assertEquals(5, leaderActor.getReplicatedLog().size());
790 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
792 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1, (short) 0));
793 assertEquals(5, leaderActor.getReplicatedLog().size());
794 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
796 // set the 2nd follower nextIndex to 1 which has been snapshotted
797 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1, (short)0));
798 assertEquals(5, leaderActor.getReplicatedLog().size());
799 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
801 // simulate a real snapshot
802 leaderActor.onReceiveCommand(SendHeartBeat.INSTANCE);
803 assertEquals(5, leaderActor.getReplicatedLog().size());
804 assertEquals(String.format("expected to be Leader but was %s. Current Leader = %s ",
805 leaderActor.getCurrentBehavior().state(), leaderActor.getLeaderId()),
806 RaftState.Leader, leaderActor.getCurrentBehavior().state());
809 //reply from a slow follower does not initiate a fake snapshot
810 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1, (short)0));
811 assertEquals("Fake snapshot should not happen when Initiate is in progress", 5,
812 leaderActor.getReplicatedLog().size());
814 ByteString snapshotBytes = fromObject(Arrays.asList(
815 new MockRaftActorContext.MockPayload("foo-0"),
816 new MockRaftActorContext.MockPayload("foo-1"),
817 new MockRaftActorContext.MockPayload("foo-2"),
818 new MockRaftActorContext.MockPayload("foo-3"),
819 new MockRaftActorContext.MockPayload("foo-4")));
820 leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
821 assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
823 assertEquals("Real snapshot didn't clear the log till replicatedToAllIndex", 0,
824 leaderActor.getReplicatedLog().size());
826 //reply from a slow follower after should not raise errors
827 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1, (short) 0));
828 assertEquals(0, leaderActor.getReplicatedLog().size());
832 public void testRealSnapshotWhenReplicatedToAllIndexMinusOne() throws Exception {
833 String persistenceId = factory.generateActorId("leader-");
834 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
835 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
836 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
837 config.setSnapshotBatchCount(5);
839 DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
841 Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
843 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
844 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
846 MockRaftActor leaderActor = mockActorRef.underlyingActor();
847 leaderActor.getRaftActorContext().setCommitIndex(3);
848 leaderActor.getRaftActorContext().setLastApplied(3);
849 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
851 leaderActor.waitForInitializeBehaviorComplete();
852 for (int i = 0; i < 4; i++) {
853 leaderActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, i,
854 new MockRaftActorContext.MockPayload("A")));
857 Leader leader = new Leader(leaderActor.getRaftActorContext());
858 leaderActor.setCurrentBehavior(leader);
859 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
861 // Simulate an install snaphost to a follower.
862 leaderActor.getRaftActorContext().getSnapshotManager().captureToInstall(
863 leaderActor.getReplicatedLog().last(), -1, "member1");
865 // Now send a CaptureSnapshotReply
866 mockActorRef.tell(new CaptureSnapshotReply(fromObject("foo").toByteArray()), mockActorRef);
868 // Trimming log in this scenario is a no-op
869 assertEquals(-1, leaderActor.getReplicatedLog().getSnapshotIndex());
870 assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
871 assertEquals(-1, leader.getReplicatedToAllIndex());
875 public void testRealSnapshotWhenReplicatedToAllIndexNotInReplicatedLog() throws Exception {
876 String persistenceId = factory.generateActorId("leader-");
877 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
878 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
879 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
880 config.setSnapshotBatchCount(5);
882 DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
884 Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().put("member1", "address").build();
886 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
887 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
889 MockRaftActor leaderActor = mockActorRef.underlyingActor();
890 leaderActor.getRaftActorContext().setCommitIndex(3);
891 leaderActor.getRaftActorContext().setLastApplied(3);
892 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
893 leaderActor.getReplicatedLog().setSnapshotIndex(3);
895 leaderActor.waitForInitializeBehaviorComplete();
896 Leader leader = new Leader(leaderActor.getRaftActorContext());
897 leaderActor.setCurrentBehavior(leader);
898 leader.setReplicatedToAllIndex(3);
899 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
901 // Persist another entry (this will cause a CaptureSnapshot to be triggered
902 leaderActor.persistData(mockActorRef, new MockIdentifier("x"),
903 new MockRaftActorContext.MockPayload("duh"));
905 // Now send a CaptureSnapshotReply
906 mockActorRef.tell(new CaptureSnapshotReply(fromObject("foo").toByteArray()), mockActorRef);
908 // Trimming log in this scenario is a no-op
909 assertEquals(3, leaderActor.getReplicatedLog().getSnapshotIndex());
910 assertTrue(leaderActor.getRaftActorContext().getSnapshotManager().isCapturing());
911 assertEquals(3, leader.getReplicatedToAllIndex());
915 public void testSwitchBehavior() {
916 String persistenceId = factory.generateActorId("leader-");
917 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
918 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
919 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
920 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
921 config.setSnapshotBatchCount(5);
923 DataPersistenceProvider dataPersistenceProvider = new NonPersistentDataProvider();
925 Map<String, String> peerAddresses = ImmutableMap.<String, String>builder().build();
927 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
928 MockRaftActor.props(persistenceId, peerAddresses, config, dataPersistenceProvider), persistenceId);
930 MockRaftActor leaderActor = mockActorRef.underlyingActor();
932 leaderActor.waitForRecoveryComplete();
934 leaderActor.handleCommand(new SwitchBehavior(RaftState.Follower, 100));
936 assertEquals(100, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
937 assertEquals(RaftState.Follower, leaderActor.getCurrentBehavior().state());
939 leaderActor.handleCommand(new SwitchBehavior(RaftState.Leader, 110));
941 assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
942 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
944 leaderActor.handleCommand(new SwitchBehavior(RaftState.Candidate, 125));
946 assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
947 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
949 leaderActor.handleCommand(new SwitchBehavior(RaftState.IsolatedLeader, 125));
951 assertEquals(110, leaderActor.getRaftActorContext().getTermInformation().getCurrentTerm());
952 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
955 public static ByteString fromObject(Object snapshot) throws Exception {
956 ByteArrayOutputStream bos = null;
957 ObjectOutputStream os = null;
959 bos = new ByteArrayOutputStream();
960 os = new ObjectOutputStream(bos);
961 os.writeObject(snapshot);
962 byte[] snapshotBytes = bos.toByteArray();
963 return ByteString.copyFrom(snapshotBytes);
976 public void testUpdateConfigParam() throws Exception {
977 DefaultConfigParamsImpl emptyConfig = new DefaultConfigParamsImpl();
978 String persistenceId = factory.generateActorId("follower-");
979 ImmutableMap<String, String> peerAddresses =
980 ImmutableMap.<String, String>builder().put("member1", "address").build();
981 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
983 TestActorRef<MockRaftActor> actorRef = factory.createTestActor(
984 MockRaftActor.props(persistenceId, peerAddresses, emptyConfig, dataPersistenceProvider), persistenceId);
985 MockRaftActor mockRaftActor = actorRef.underlyingActor();
986 mockRaftActor.waitForInitializeBehaviorComplete();
988 RaftActorBehavior behavior = mockRaftActor.getCurrentBehavior();
989 mockRaftActor.updateConfigParams(emptyConfig);
990 assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
991 assertEquals("Behavior State", RaftState.Follower,
992 mockRaftActor.getCurrentBehavior().state());
994 DefaultConfigParamsImpl disableConfig = new DefaultConfigParamsImpl();
995 disableConfig.setCustomRaftPolicyImplementationClass(
996 "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
997 mockRaftActor.updateConfigParams(disableConfig);
998 assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior());
999 assertEquals("Behavior State", RaftState.Follower,
1000 mockRaftActor.getCurrentBehavior().state());
1002 behavior = mockRaftActor.getCurrentBehavior();
1003 mockRaftActor.updateConfigParams(disableConfig);
1004 assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
1005 assertEquals("Behavior State", RaftState.Follower,
1006 mockRaftActor.getCurrentBehavior().state());
1008 DefaultConfigParamsImpl defaultConfig = new DefaultConfigParamsImpl();
1009 defaultConfig.setCustomRaftPolicyImplementationClass(
1010 "org.opendaylight.controller.cluster.raft.policy.DefaultRaftPolicy");
1011 mockRaftActor.updateConfigParams(defaultConfig);
1012 assertNotSame("Different Behavior", behavior, mockRaftActor.getCurrentBehavior());
1013 assertEquals("Behavior State", RaftState.Follower,
1014 mockRaftActor.getCurrentBehavior().state());
1016 behavior = mockRaftActor.getCurrentBehavior();
1017 mockRaftActor.updateConfigParams(defaultConfig);
1018 assertSame("Same Behavior", behavior, mockRaftActor.getCurrentBehavior());
1019 assertEquals("Behavior State", RaftState.Follower,
1020 mockRaftActor.getCurrentBehavior().state());
1024 public void testGetSnapshot() throws Exception {
1025 TEST_LOG.info("testGetSnapshot starting");
1027 final JavaTestKit kit = new JavaTestKit(getSystem());
1029 String persistenceId = factory.generateActorId("test-actor-");
1030 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1031 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1035 InMemoryJournal.addEntry(persistenceId, seqN++, new UpdateElectionTerm(term, "member-1"));
1036 InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 0,
1037 new MockRaftActorContext.MockPayload("A")));
1038 InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 1,
1039 new MockRaftActorContext.MockPayload("B")));
1040 InMemoryJournal.addEntry(persistenceId, seqN++, new ApplyJournalEntries(1));
1041 InMemoryJournal.addEntry(persistenceId, seqN++, new MockRaftActorContext.MockReplicatedLogEntry(term, 2,
1042 new MockRaftActorContext.MockPayload("C")));
1044 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
1045 ImmutableMap.<String, String>builder().put("member1", "address").build(), config)
1046 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1047 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
1049 mockRaftActor.waitForRecoveryComplete();
1051 mockRaftActor.snapshotCohortDelegate = mock(RaftActorSnapshotCohort.class);
1053 raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
1055 ArgumentCaptor<ActorRef> replyActor = ArgumentCaptor.forClass(ActorRef.class);
1056 verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).createSnapshot(replyActor.capture());
1058 byte[] stateSnapshot = new byte[]{1,2,3};
1059 replyActor.getValue().tell(new CaptureSnapshotReply(stateSnapshot), ActorRef.noSender());
1061 GetSnapshotReply reply = kit.expectMsgClass(GetSnapshotReply.class);
1063 assertEquals("getId", persistenceId, reply.getId());
1064 Snapshot replySnapshot = SerializationUtils.deserialize(reply.getSnapshot());
1065 assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm());
1066 assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor());
1067 assertEquals("getLastAppliedIndex", 1L, replySnapshot.getLastAppliedIndex());
1068 assertEquals("getLastAppliedTerm", term, replySnapshot.getLastAppliedTerm());
1069 assertEquals("getLastIndex", 2L, replySnapshot.getLastIndex());
1070 assertEquals("getLastTerm", term, replySnapshot.getLastTerm());
1071 assertArrayEquals("getState", stateSnapshot, replySnapshot.getState());
1072 assertEquals("getUnAppliedEntries size", 1, replySnapshot.getUnAppliedEntries().size());
1073 assertEquals("UnApplied entry index ", 2L, replySnapshot.getUnAppliedEntries().get(0).getIndex());
1075 // Test with timeout
1077 mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(
1078 Duration.create(200, TimeUnit.MILLISECONDS));
1079 reset(mockRaftActor.snapshotCohortDelegate);
1081 raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
1082 Failure failure = kit.expectMsgClass(akka.actor.Status.Failure.class);
1083 assertEquals("Failure cause type", TimeoutException.class, failure.cause().getClass());
1085 mockRaftActor.getSnapshotMessageSupport().setSnapshotReplyActorTimeout(Duration.create(30, TimeUnit.SECONDS));
1087 // Test with persistence disabled.
1089 mockRaftActor.setPersistence(false);
1090 reset(mockRaftActor.snapshotCohortDelegate);
1092 raftActorRef.tell(GetSnapshot.INSTANCE, kit.getRef());
1093 reply = kit.expectMsgClass(GetSnapshotReply.class);
1094 verify(mockRaftActor.snapshotCohortDelegate, never()).createSnapshot(any(ActorRef.class));
1096 assertEquals("getId", persistenceId, reply.getId());
1097 replySnapshot = SerializationUtils.deserialize(reply.getSnapshot());
1098 assertEquals("getElectionTerm", term, replySnapshot.getElectionTerm());
1099 assertEquals("getElectionVotedFor", "member-1", replySnapshot.getElectionVotedFor());
1100 assertEquals("getLastAppliedIndex", -1L, replySnapshot.getLastAppliedIndex());
1101 assertEquals("getLastAppliedTerm", -1L, replySnapshot.getLastAppliedTerm());
1102 assertEquals("getLastIndex", -1L, replySnapshot.getLastIndex());
1103 assertEquals("getLastTerm", -1L, replySnapshot.getLastTerm());
1104 assertEquals("getState length", 0, replySnapshot.getState().length);
1105 assertEquals("getUnAppliedEntries size", 0, replySnapshot.getUnAppliedEntries().size());
1107 TEST_LOG.info("testGetSnapshot ending");
1111 public void testRestoreFromSnapshot() throws Exception {
1112 TEST_LOG.info("testRestoreFromSnapshot starting");
1114 String persistenceId = factory.generateActorId("test-actor-");
1115 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1116 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1118 List<ReplicatedLogEntry> snapshotUnappliedEntries = new ArrayList<>();
1119 snapshotUnappliedEntries.add(new MockRaftActorContext.MockReplicatedLogEntry(1, 4,
1120 new MockRaftActorContext.MockPayload("E")));
1122 int snapshotLastApplied = 3;
1123 int snapshotLastIndex = 4;
1125 List<MockPayload> state = Arrays.asList(
1126 new MockRaftActorContext.MockPayload("A"),
1127 new MockRaftActorContext.MockPayload("B"),
1128 new MockRaftActorContext.MockPayload("C"),
1129 new MockRaftActorContext.MockPayload("D"));
1130 ByteString stateBytes = fromObject(state);
1132 Snapshot snapshot = Snapshot.create(stateBytes.toByteArray(), snapshotUnappliedEntries,
1133 snapshotLastIndex, 1, snapshotLastApplied, 1, 1, "member-1");
1135 InMemorySnapshotStore.addSnapshotSavedLatch(persistenceId);
1137 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
1138 .config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot)).props()
1139 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1140 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
1142 mockRaftActor.waitForRecoveryComplete();
1144 Snapshot savedSnapshot = InMemorySnapshotStore.waitForSavedSnapshot(persistenceId, Snapshot.class);
1145 assertEquals("getElectionTerm", snapshot.getElectionTerm(), savedSnapshot.getElectionTerm());
1146 assertEquals("getElectionVotedFor", snapshot.getElectionVotedFor(), savedSnapshot.getElectionVotedFor());
1147 assertEquals("getLastAppliedIndex", snapshot.getLastAppliedIndex(), savedSnapshot.getLastAppliedIndex());
1148 assertEquals("getLastAppliedTerm", snapshot.getLastAppliedTerm(), savedSnapshot.getLastAppliedTerm());
1149 assertEquals("getLastIndex", snapshot.getLastIndex(), savedSnapshot.getLastIndex());
1150 assertEquals("getLastTerm", snapshot.getLastTerm(), savedSnapshot.getLastTerm());
1151 assertArrayEquals("getState", snapshot.getState(), savedSnapshot.getState());
1152 assertEquals("getUnAppliedEntries", snapshot.getUnAppliedEntries(), savedSnapshot.getUnAppliedEntries());
1154 verify(mockRaftActor.snapshotCohortDelegate, timeout(5000)).applySnapshot(any(byte[].class));
1156 RaftActorContext context = mockRaftActor.getRaftActorContext();
1157 assertEquals("Journal log size", 1, context.getReplicatedLog().size());
1158 assertEquals("Last index", snapshotLastIndex, context.getReplicatedLog().lastIndex());
1159 assertEquals("Last applied", snapshotLastApplied, context.getLastApplied());
1160 assertEquals("Commit index", snapshotLastApplied, context.getCommitIndex());
1161 assertEquals("Recovered state", state, mockRaftActor.getState());
1162 assertEquals("Current term", 1L, context.getTermInformation().getCurrentTerm());
1163 assertEquals("Voted for", "member-1", context.getTermInformation().getVotedFor());
1165 // Test with data persistence disabled
1167 snapshot = Snapshot.create(new byte[0], Collections.<ReplicatedLogEntry>emptyList(),
1168 -1, -1, -1, -1, 5, "member-1");
1170 persistenceId = factory.generateActorId("test-actor-");
1172 raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
1173 .config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot))
1174 .persistent(Optional.of(Boolean.FALSE)).props()
1175 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1176 mockRaftActor = raftActorRef.underlyingActor();
1178 mockRaftActor.waitForRecoveryComplete();
1179 assertEquals("snapshot committed", true,
1180 Uninterruptibles.awaitUninterruptibly(mockRaftActor.snapshotCommitted, 5, TimeUnit.SECONDS));
1182 context = mockRaftActor.getRaftActorContext();
1183 assertEquals("Current term", 5L, context.getTermInformation().getCurrentTerm());
1184 assertEquals("Voted for", "member-1", context.getTermInformation().getVotedFor());
1186 TEST_LOG.info("testRestoreFromSnapshot ending");
1190 public void testRestoreFromSnapshotWithRecoveredData() throws Exception {
1191 TEST_LOG.info("testRestoreFromSnapshotWithRecoveredData starting");
1193 String persistenceId = factory.generateActorId("test-actor-");
1194 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1195 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1197 List<MockPayload> state = Arrays.asList(new MockRaftActorContext.MockPayload("A"));
1198 Snapshot snapshot = Snapshot.create(fromObject(state).toByteArray(), Arrays.<ReplicatedLogEntry>asList(),
1199 5, 2, 5, 2, 2, "member-1");
1201 InMemoryJournal.addEntry(persistenceId, 1, new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
1202 new MockRaftActorContext.MockPayload("B")));
1204 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
1205 .config(config).restoreFromSnapshot(SerializationUtils.serialize(snapshot)).props()
1206 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1207 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
1209 mockRaftActor.waitForRecoveryComplete();
1211 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
1212 verify(mockRaftActor.snapshotCohortDelegate, never()).applySnapshot(any(byte[].class));
1214 RaftActorContext context = mockRaftActor.getRaftActorContext();
1215 assertEquals("Journal log size", 1, context.getReplicatedLog().size());
1216 assertEquals("Last index", 0, context.getReplicatedLog().lastIndex());
1217 assertEquals("Last applied", -1, context.getLastApplied());
1218 assertEquals("Commit index", -1, context.getCommitIndex());
1219 assertEquals("Current term", 0, context.getTermInformation().getCurrentTerm());
1220 assertEquals("Voted for", null, context.getTermInformation().getVotedFor());
1222 TEST_LOG.info("testRestoreFromSnapshotWithRecoveredData ending");
1226 public void testNonVotingOnRecovery() throws Exception {
1227 TEST_LOG.info("testNonVotingOnRecovery starting");
1229 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1230 config.setElectionTimeoutFactor(1);
1231 config.setHeartBeatInterval(FiniteDuration.create(1, TimeUnit.MILLISECONDS));
1233 String persistenceId = factory.generateActorId("test-actor-");
1234 InMemoryJournal.addEntry(persistenceId, 1, new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
1235 new ServerConfigurationPayload(Arrays.asList(new ServerInfo(persistenceId, false)))));
1237 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
1238 .config(config).props().withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1239 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
1241 mockRaftActor.waitForInitializeBehaviorComplete();
1243 // Sleep a bit and verify it didn't get an election timeout and schedule an election.
1245 Uninterruptibles.sleepUninterruptibly(400, TimeUnit.MILLISECONDS);
1246 assertEquals("getRaftState", RaftState.Follower, mockRaftActor.getRaftState());
1248 TEST_LOG.info("testNonVotingOnRecovery ending");
1252 public void testLeaderTransitioning() throws Exception {
1253 TEST_LOG.info("testLeaderTransitioning starting");
1255 TestActorRef<MessageCollectorActor> notifierActor = factory.createTestActor(
1256 Props.create(MessageCollectorActor.class));
1258 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1259 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1261 String persistenceId = factory.generateActorId("test-actor-");
1263 TestActorRef<MockRaftActor> raftActorRef = factory.createTestActor(MockRaftActor.builder().id(persistenceId)
1264 .config(config).roleChangeNotifier(notifierActor).props()
1265 .withDispatcher(Dispatchers.DefaultDispatcherId()), persistenceId);
1266 MockRaftActor mockRaftActor = raftActorRef.underlyingActor();
1268 mockRaftActor.waitForInitializeBehaviorComplete();
1270 raftActorRef.tell(new AppendEntries(1L, "leader", 0L, 1L, Collections.<ReplicatedLogEntry>emptyList(),
1271 0L, -1L, (short)1), ActorRef.noSender());
1272 LeaderStateChanged leaderStateChange = MessageCollectorActor.expectFirstMatching(
1273 notifierActor, LeaderStateChanged.class);
1274 assertEquals("getLeaderId", "leader", leaderStateChange.getLeaderId());
1276 MessageCollectorActor.clearMessages(notifierActor);
1278 raftActorRef.tell(LeaderTransitioning.INSTANCE, ActorRef.noSender());
1280 leaderStateChange = MessageCollectorActor.expectFirstMatching(notifierActor, LeaderStateChanged.class);
1281 assertEquals("getMemberId", persistenceId, leaderStateChange.getMemberId());
1282 assertEquals("getLeaderId", null, leaderStateChange.getLeaderId());
1284 TEST_LOG.info("testLeaderTransitioning ending");