d90494f41d9bf67ed4b34ec5051781fef2022b44
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / RaftActorTest.java
1 package org.opendaylight.controller.cluster.raft;
2
3 import static org.junit.Assert.assertEquals;
4 import static org.junit.Assert.assertFalse;
5 import static org.junit.Assert.assertNotEquals;
6 import static org.junit.Assert.assertNotNull;
7 import static org.junit.Assert.assertNull;
8 import static org.junit.Assert.assertTrue;
9 import static org.mockito.Matchers.any;
10 import static org.mockito.Matchers.anyObject;
11 import static org.mockito.Matchers.eq;
12 import static org.mockito.Mockito.doReturn;
13 import static org.mockito.Mockito.mock;
14 import static org.mockito.Mockito.times;
15 import static org.mockito.Mockito.verify;
16 import akka.actor.ActorRef;
17 import akka.actor.ActorSystem;
18 import akka.actor.PoisonPill;
19 import akka.actor.Props;
20 import akka.actor.Terminated;
21 import akka.japi.Creator;
22 import akka.japi.Procedure;
23 import akka.pattern.Patterns;
24 import akka.persistence.RecoveryCompleted;
25 import akka.persistence.SaveSnapshotFailure;
26 import akka.persistence.SaveSnapshotSuccess;
27 import akka.persistence.SnapshotMetadata;
28 import akka.persistence.SnapshotOffer;
29 import akka.persistence.SnapshotSelectionCriteria;
30 import akka.testkit.JavaTestKit;
31 import akka.testkit.TestActorRef;
32 import akka.util.Timeout;
33 import com.google.common.base.Optional;
34 import com.google.common.collect.Lists;
35 import com.google.common.util.concurrent.Uninterruptibles;
36 import com.google.protobuf.ByteString;
37 import java.io.ByteArrayInputStream;
38 import java.io.ByteArrayOutputStream;
39 import java.io.IOException;
40 import java.io.ObjectInputStream;
41 import java.io.ObjectOutputStream;
42 import java.util.ArrayList;
43 import java.util.Arrays;
44 import java.util.Collections;
45 import java.util.HashMap;
46 import java.util.List;
47 import java.util.Map;
48 import java.util.concurrent.CountDownLatch;
49 import java.util.concurrent.TimeUnit;
50 import java.util.concurrent.TimeoutException;
51 import org.junit.After;
52 import org.junit.Assert;
53 import org.junit.Before;
54 import org.junit.Test;
55 import org.opendaylight.controller.cluster.DataPersistenceProvider;
56 import org.opendaylight.controller.cluster.datastore.DataPersistenceProviderMonitor;
57 import org.opendaylight.controller.cluster.notifications.RoleChanged;
58 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
59 import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
60 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
61 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
62 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
63 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
64 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
65 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
66 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
67 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
68 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
69 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
70 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
71 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
72 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
73 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
74 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
75 import scala.concurrent.Await;
76 import scala.concurrent.Future;
77 import scala.concurrent.duration.Duration;
78 import scala.concurrent.duration.FiniteDuration;
79
80 public class RaftActorTest extends AbstractActorTest {
81
82     private TestActorFactory factory;
83
84     @Before
85     public void setUp(){
86         factory = new TestActorFactory(getSystem());
87     }
88
89     @After
90     public void tearDown() throws Exception {
91         factory.close();
92         InMemoryJournal.clear();
93         InMemorySnapshotStore.clear();
94     }
95
96     public static class MockRaftActor extends RaftActor {
97
98         private final DataPersistenceProvider dataPersistenceProvider;
99         private final RaftActor delegate;
100         private final CountDownLatch recoveryComplete = new CountDownLatch(1);
101         private final List<Object> state;
102         private ActorRef roleChangeNotifier;
103         private final CountDownLatch initializeBehaviorComplete = new CountDownLatch(1);
104
105         public static final class MockRaftActorCreator implements Creator<MockRaftActor> {
106             private static final long serialVersionUID = 1L;
107             private final Map<String, String> peerAddresses;
108             private final String id;
109             private final Optional<ConfigParams> config;
110             private final DataPersistenceProvider dataPersistenceProvider;
111             private final ActorRef roleChangeNotifier;
112
113             private MockRaftActorCreator(Map<String, String> peerAddresses, String id,
114                 Optional<ConfigParams> config, DataPersistenceProvider dataPersistenceProvider,
115                 ActorRef roleChangeNotifier) {
116                 this.peerAddresses = peerAddresses;
117                 this.id = id;
118                 this.config = config;
119                 this.dataPersistenceProvider = dataPersistenceProvider;
120                 this.roleChangeNotifier = roleChangeNotifier;
121             }
122
123             @Override
124             public MockRaftActor create() throws Exception {
125                 MockRaftActor mockRaftActor = new MockRaftActor(id, peerAddresses, config,
126                     dataPersistenceProvider);
127                 mockRaftActor.roleChangeNotifier = this.roleChangeNotifier;
128                 return mockRaftActor;
129             }
130         }
131
132         public MockRaftActor(String id, Map<String, String> peerAddresses, Optional<ConfigParams> config,
133                              DataPersistenceProvider dataPersistenceProvider) {
134             super(id, peerAddresses, config);
135             state = new ArrayList<>();
136             this.delegate = mock(RaftActor.class);
137             if(dataPersistenceProvider == null){
138                 this.dataPersistenceProvider = new PersistentDataProvider();
139             } else {
140                 this.dataPersistenceProvider = dataPersistenceProvider;
141             }
142         }
143
144         public void waitForRecoveryComplete() {
145             try {
146                 assertEquals("Recovery complete", true, recoveryComplete.await(5,  TimeUnit.SECONDS));
147             } catch (InterruptedException e) {
148                 e.printStackTrace();
149             }
150         }
151
152         public void waitForInitializeBehaviorComplete() {
153             try {
154                 assertEquals("Behavior initialized", true, initializeBehaviorComplete.await(5,  TimeUnit.SECONDS));
155             } catch (InterruptedException e) {
156                 e.printStackTrace();
157             }
158         }
159
160         public List<Object> getState() {
161             return state;
162         }
163
164         public static Props props(final String id, final Map<String, String> peerAddresses,
165                 Optional<ConfigParams> config){
166             return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null, null));
167         }
168
169         public static Props props(final String id, final Map<String, String> peerAddresses,
170                                   Optional<ConfigParams> config, DataPersistenceProvider dataPersistenceProvider){
171             return Props.create(new MockRaftActorCreator(peerAddresses, id, config, dataPersistenceProvider, null));
172         }
173
174         public static Props props(final String id, final Map<String, String> peerAddresses,
175             Optional<ConfigParams> config, ActorRef roleChangeNotifier){
176             return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null, roleChangeNotifier));
177         }
178
179         @Override protected void applyState(ActorRef clientActor, String identifier, Object data) {
180             delegate.applyState(clientActor, identifier, data);
181             LOG.info("applyState called");
182         }
183
184         @Override
185         protected void startLogRecoveryBatch(int maxBatchSize) {
186         }
187
188         @Override
189         protected void appendRecoveredLogEntry(Payload data) {
190             state.add(data);
191         }
192
193         @Override
194         protected void applyCurrentLogRecoveryBatch() {
195         }
196
197         @Override
198         protected void onRecoveryComplete() {
199             delegate.onRecoveryComplete();
200             recoveryComplete.countDown();
201         }
202
203         @Override
204         protected void initializeBehavior() {
205             super.initializeBehavior();
206             initializeBehaviorComplete.countDown();
207         }
208
209         @Override
210         protected void applyRecoverySnapshot(byte[] bytes) {
211             delegate.applyRecoverySnapshot(bytes);
212             try {
213                 Object data = toObject(bytes);
214                 if (data instanceof List) {
215                     state.addAll((List<?>) data);
216                 }
217             } catch (Exception e) {
218                 e.printStackTrace();
219             }
220         }
221
222         @Override protected void createSnapshot() {
223             delegate.createSnapshot();
224         }
225
226         @Override protected void applySnapshot(byte [] snapshot) {
227             delegate.applySnapshot(snapshot);
228         }
229
230         @Override protected void onStateChanged() {
231             delegate.onStateChanged();
232         }
233
234         @Override
235         protected DataPersistenceProvider persistence() {
236             return this.dataPersistenceProvider;
237         }
238
239         @Override
240         protected Optional<ActorRef> getRoleChangeNotifier() {
241             return Optional.fromNullable(roleChangeNotifier);
242         }
243
244         @Override public String persistenceId() {
245             return this.getId();
246         }
247
248         private Object toObject(byte[] bs) throws ClassNotFoundException, IOException {
249             Object obj = null;
250             ByteArrayInputStream bis = null;
251             ObjectInputStream ois = null;
252             try {
253                 bis = new ByteArrayInputStream(bs);
254                 ois = new ObjectInputStream(bis);
255                 obj = ois.readObject();
256             } finally {
257                 if (bis != null) {
258                     bis.close();
259                 }
260                 if (ois != null) {
261                     ois.close();
262                 }
263             }
264             return obj;
265         }
266
267         public ReplicatedLog getReplicatedLog(){
268             return this.getRaftActorContext().getReplicatedLog();
269         }
270
271     }
272
273
274     private static class RaftActorTestKit extends JavaTestKit {
275         private final ActorRef raftActor;
276
277         public RaftActorTestKit(ActorSystem actorSystem, String actorName) {
278             super(actorSystem);
279
280             raftActor = this.getSystem().actorOf(MockRaftActor.props(actorName,
281                     Collections.<String,String>emptyMap(), Optional.<ConfigParams>absent()), actorName);
282
283         }
284
285
286         public ActorRef getRaftActor() {
287             return raftActor;
288         }
289
290         public boolean waitForLogMessage(final Class<?> logEventClass, String message){
291             // Wait for a specific log message to show up
292             return
293                 new JavaTestKit.EventFilter<Boolean>(logEventClass
294                 ) {
295                     @Override
296                     protected Boolean run() {
297                         return true;
298                     }
299                 }.from(raftActor.path().toString())
300                     .message(message)
301                     .occurrences(1).exec();
302
303
304         }
305
306         protected void waitUntilLeader(){
307             waitUntilLeader(raftActor);
308         }
309
310         protected void waitUntilLeader(ActorRef actorRef) {
311             FiniteDuration duration = Duration.create(100, TimeUnit.MILLISECONDS);
312             for(int i = 0; i < 20 * 5; i++) {
313                 Future<Object> future = Patterns.ask(actorRef, new FindLeader(), new Timeout(duration));
314                 try {
315                     FindLeaderReply resp = (FindLeaderReply) Await.result(future, duration);
316                     if(resp.getLeaderActor() != null) {
317                         return;
318                     }
319                 } catch(TimeoutException e) {
320                 } catch(Exception e) {
321                     System.err.println("FindLeader threw ex");
322                     e.printStackTrace();
323                 }
324
325
326                 Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
327             }
328
329             Assert.fail("Leader not found for actorRef " + actorRef.path());
330         }
331
332     }
333
334
335     @Test
336     public void testConstruction() {
337         new RaftActorTestKit(getSystem(), "testConstruction").waitUntilLeader();
338     }
339
340     @Test
341     public void testFindLeaderWhenLeaderIsSelf(){
342         RaftActorTestKit kit = new RaftActorTestKit(getSystem(), "testFindLeader");
343         kit.waitUntilLeader();
344     }
345
346     @Test
347     public void testRaftActorRecovery() throws Exception {
348         new JavaTestKit(getSystem()) {{
349             String persistenceId = factory.generateActorId("follower-");
350
351             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
352             // Set the heartbeat interval high to essentially disable election otherwise the test
353             // may fail if the actor is switched to Leader and the commitIndex is set to the last
354             // log entry.
355             config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
356
357             ActorRef followerActor = factory.createActor(MockRaftActor.props(persistenceId,
358                     Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
359
360             watch(followerActor);
361
362             List<ReplicatedLogEntry> snapshotUnappliedEntries = new ArrayList<>();
363             ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 4,
364                     new MockRaftActorContext.MockPayload("E"));
365             snapshotUnappliedEntries.add(entry1);
366
367             int lastAppliedDuringSnapshotCapture = 3;
368             int lastIndexDuringSnapshotCapture = 4;
369
370             // 4 messages as part of snapshot, which are applied to state
371             ByteString snapshotBytes = fromObject(Arrays.asList(
372                     new MockRaftActorContext.MockPayload("A"),
373                     new MockRaftActorContext.MockPayload("B"),
374                     new MockRaftActorContext.MockPayload("C"),
375                     new MockRaftActorContext.MockPayload("D")));
376
377             Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
378                     snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1,
379                     lastAppliedDuringSnapshotCapture, 1);
380             InMemorySnapshotStore.addSnapshot(persistenceId, snapshot);
381
382             // add more entries after snapshot is taken
383             List<ReplicatedLogEntry> entries = new ArrayList<>();
384             ReplicatedLogEntry entry2 = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
385                     new MockRaftActorContext.MockPayload("F"));
386             ReplicatedLogEntry entry3 = new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
387                     new MockRaftActorContext.MockPayload("G"));
388             ReplicatedLogEntry entry4 = new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
389                     new MockRaftActorContext.MockPayload("H"));
390             entries.add(entry2);
391             entries.add(entry3);
392             entries.add(entry4);
393
394             int lastAppliedToState = 5;
395             int lastIndex = 7;
396
397             InMemoryJournal.addEntry(persistenceId, 5, entry2);
398             // 2 entries are applied to state besides the 4 entries in snapshot
399             InMemoryJournal.addEntry(persistenceId, 6, new ApplyJournalEntries(lastAppliedToState));
400             InMemoryJournal.addEntry(persistenceId, 7, entry3);
401             InMemoryJournal.addEntry(persistenceId, 8, entry4);
402
403             // kill the actor
404             followerActor.tell(PoisonPill.getInstance(), null);
405             expectMsgClass(duration("5 seconds"), Terminated.class);
406
407             unwatch(followerActor);
408
409             //reinstate the actor
410             TestActorRef<MockRaftActor> ref = factory.createTestActor(
411                     MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
412                             Optional.<ConfigParams>of(config)));
413
414             ref.underlyingActor().waitForRecoveryComplete();
415
416             RaftActorContext context = ref.underlyingActor().getRaftActorContext();
417             assertEquals("Journal log size", snapshotUnappliedEntries.size() + entries.size(),
418                     context.getReplicatedLog().size());
419             assertEquals("Last index", lastIndex, context.getReplicatedLog().lastIndex());
420             assertEquals("Last applied", lastAppliedToState, context.getLastApplied());
421             assertEquals("Commit index", lastAppliedToState, context.getCommitIndex());
422             assertEquals("Recovered state size", 6, ref.underlyingActor().getState().size());
423         }};
424     }
425
426     @Test
427     public void testRaftActorRecoveryWithPreLithuimApplyLogEntries() throws Exception {
428         new JavaTestKit(getSystem()) {{
429             String persistenceId = factory.generateActorId("leader-");
430
431             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
432             config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
433
434             // Setup the persisted journal with some entries
435             ReplicatedLogEntry entry0 = new MockRaftActorContext.MockReplicatedLogEntry(1, 0,
436                     new MockRaftActorContext.MockPayload("zero"));
437             ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 1,
438                     new MockRaftActorContext.MockPayload("oen"));
439             ReplicatedLogEntry entry2 = new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
440                     new MockRaftActorContext.MockPayload("two"));
441
442             long seqNr = 1;
443             InMemoryJournal.addEntry(persistenceId, seqNr++, entry0);
444             InMemoryJournal.addEntry(persistenceId, seqNr++, entry1);
445             InMemoryJournal.addEntry(persistenceId, seqNr++, new ApplyLogEntries(1));
446             InMemoryJournal.addEntry(persistenceId, seqNr++, entry2);
447
448             int lastAppliedToState = 1;
449             int lastIndex = 2;
450
451             //reinstate the actor
452             TestActorRef<MockRaftActor> leaderActor = factory.createTestActor(
453                     MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
454                             Optional.<ConfigParams>of(config)));
455
456             leaderActor.underlyingActor().waitForRecoveryComplete();
457
458             RaftActorContext context = leaderActor.underlyingActor().getRaftActorContext();
459             assertEquals("Journal log size", 3, context.getReplicatedLog().size());
460             assertEquals("Last index", lastIndex, context.getReplicatedLog().lastIndex());
461             assertEquals("Last applied", lastAppliedToState, context.getLastApplied());
462             assertEquals("Commit index", lastAppliedToState, context.getCommitIndex());
463         }};
464     }
465
466     /**
467      * This test verifies that when recovery is applicable (typically when persistence is true) the RaftActor does
468      * process recovery messages
469      *
470      * @throws Exception
471      */
472
473     @Test
474     public void testHandleRecoveryWhenDataPersistenceRecoveryApplicable() throws Exception {
475         new JavaTestKit(getSystem()) {
476             {
477                 String persistenceId = factory.generateActorId("leader-");
478
479                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
480
481                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
482
483                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
484                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
485
486                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
487
488                 // Wait for akka's recovery to complete so it doesn't interfere.
489                 mockRaftActor.waitForRecoveryComplete();
490
491                 ByteString snapshotBytes = fromObject(Arrays.asList(
492                         new MockRaftActorContext.MockPayload("A"),
493                         new MockRaftActorContext.MockPayload("B"),
494                         new MockRaftActorContext.MockPayload("C"),
495                         new MockRaftActorContext.MockPayload("D")));
496
497                 Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
498                         Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
499
500                 mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
501
502                 verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes.toByteArray()));
503
504                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
505
506                 ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
507
508                 assertEquals("add replicated log entry", 1, replicatedLog.size());
509
510                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
511
512                 assertEquals("add replicated log entry", 2, replicatedLog.size());
513
514                 mockRaftActor.onReceiveRecover(new ApplyJournalEntries(1));
515
516                 assertEquals("commit index 1", 1, mockRaftActor.getRaftActorContext().getCommitIndex());
517
518                 // The snapshot had 4 items + we added 2 more items during the test
519                 // We start removing from 5 and we should get 1 item in the replicated log
520                 mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(5));
521
522                 assertEquals("remove log entries", 1, replicatedLog.size());
523
524                 mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
525
526                 assertEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
527                 assertEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
528
529                 mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
530
531             }};
532     }
533
534     /**
535      * This test verifies that when recovery is not applicable (typically when persistence is false) the RaftActor does
536      * not process recovery messages
537      *
538      * @throws Exception
539      */
540     @Test
541     public void testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable() throws Exception {
542         new JavaTestKit(getSystem()) {
543             {
544                 String persistenceId = factory.generateActorId("leader-");
545
546                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
547
548                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
549
550                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
551                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), new DataPersistenceProviderMonitor()), persistenceId);
552
553                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
554
555                 // Wait for akka's recovery to complete so it doesn't interfere.
556                 mockRaftActor.waitForRecoveryComplete();
557
558                 ByteString snapshotBytes = fromObject(Arrays.asList(
559                         new MockRaftActorContext.MockPayload("A"),
560                         new MockRaftActorContext.MockPayload("B"),
561                         new MockRaftActorContext.MockPayload("C"),
562                         new MockRaftActorContext.MockPayload("D")));
563
564                 Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
565                         Lists.<ReplicatedLogEntry>newArrayList(), 3, 1, 3, 1);
566
567                 mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
568
569                 verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(byte[].class));
570
571                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
572
573                 ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
574
575                 assertEquals("add replicated log entry", 0, replicatedLog.size());
576
577                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
578
579                 assertEquals("add replicated log entry", 0, replicatedLog.size());
580
581                 mockRaftActor.onReceiveRecover(new ApplyJournalEntries(1));
582
583                 assertEquals("commit index -1", -1, mockRaftActor.getRaftActorContext().getCommitIndex());
584
585                 mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(2));
586
587                 assertEquals("remove log entries", 0, replicatedLog.size());
588
589                 mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
590
591                 assertNotEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
592                 assertNotEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
593
594                 mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
595             }};
596     }
597
598
599     @Test
600     public void testUpdatingElectionTermCallsDataPersistence() throws Exception {
601         new JavaTestKit(getSystem()) {
602             {
603                 String persistenceId = factory.generateActorId("leader-");
604
605                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
606
607                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
608
609                 CountDownLatch persistLatch = new CountDownLatch(1);
610                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
611                 dataPersistenceProviderMonitor.setPersistLatch(persistLatch);
612
613                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
614                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
615
616                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
617
618                 mockRaftActor.waitForInitializeBehaviorComplete();
619
620                 mockRaftActor.getRaftActorContext().getTermInformation().updateAndPersist(10, "foobar");
621
622                 assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS));
623             }
624         };
625     }
626
627     @Test
628     public void testAddingReplicatedLogEntryCallsDataPersistence() throws Exception {
629         new JavaTestKit(getSystem()) {
630             {
631                 String persistenceId = factory.generateActorId("leader-");
632
633                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
634
635                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
636
637                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
638
639                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
640                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
641
642                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
643
644                 mockRaftActor.waitForInitializeBehaviorComplete();
645
646                 MockRaftActorContext.MockReplicatedLogEntry logEntry = new MockRaftActorContext.MockReplicatedLogEntry(10, 10, mock(Payload.class));
647
648                 mockRaftActor.getRaftActorContext().getReplicatedLog().appendAndPersist(logEntry);
649
650                 verify(dataPersistenceProvider).persist(eq(logEntry), any(Procedure.class));
651             }
652         };
653     }
654
655     @Test
656     public void testRemovingReplicatedLogEntryCallsDataPersistence() throws Exception {
657         new JavaTestKit(getSystem()) {
658             {
659                 String persistenceId = factory.generateActorId("leader-");
660
661                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
662
663                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
664
665                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
666
667                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
668                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
669
670                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
671
672                 mockRaftActor.waitForInitializeBehaviorComplete();
673
674                 mockRaftActor.getReplicatedLog().appendAndPersist(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
675
676                 mockRaftActor.getRaftActorContext().getReplicatedLog().removeFromAndPersist(0);
677
678                 verify(dataPersistenceProvider, times(2)).persist(anyObject(), any(Procedure.class));
679             }
680         };
681     }
682
683     @Test
684     public void testApplyJournalEntriesCallsDataPersistence() throws Exception {
685         new JavaTestKit(getSystem()) {
686             {
687                 String persistenceId = factory.generateActorId("leader-");
688
689                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
690
691                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
692
693                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
694
695                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
696                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
697
698                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
699
700                 mockRaftActor.waitForInitializeBehaviorComplete();
701
702                 mockRaftActor.onReceiveCommand(new ApplyJournalEntries(10));
703
704                 verify(dataPersistenceProvider, times(1)).persist(anyObject(), any(Procedure.class));
705
706             }
707
708         };
709     }
710
711     @Test
712     public void testCaptureSnapshotReplyCallsDataPersistence() throws Exception {
713         new JavaTestKit(getSystem()) {
714             {
715                 String persistenceId = factory.generateActorId("leader-");
716
717                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
718
719                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
720
721                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
722
723                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
724                         MockRaftActor.props(persistenceId, Collections.<String, String>emptyMap(),
725                                 Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
726
727                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
728
729                 mockRaftActor.waitForInitializeBehaviorComplete();
730
731                 ByteString snapshotBytes = fromObject(Arrays.asList(
732                         new MockRaftActorContext.MockPayload("A"),
733                         new MockRaftActorContext.MockPayload("B"),
734                         new MockRaftActorContext.MockPayload("C"),
735                         new MockRaftActorContext.MockPayload("D")));
736
737                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1,-1, 1, -1, 1));
738
739                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
740
741                 mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
742
743                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
744
745                 verify(dataPersistenceProvider).saveSnapshot(anyObject());
746
747             }
748         };
749     }
750
751     @Test
752     public void testSaveSnapshotSuccessCallsDataPersistence() throws Exception {
753         new JavaTestKit(getSystem()) {
754             {
755                 String persistenceId = factory.generateActorId("leader-");
756
757                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
758
759                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
760
761                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
762
763                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
764                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
765
766                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
767
768                 mockRaftActor.waitForInitializeBehaviorComplete();
769
770                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
771                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class)));
772                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 2, mock(Payload.class)));
773                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 3, mock(Payload.class)));
774                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 4, mock(Payload.class)));
775
776                 ByteString snapshotBytes = fromObject(Arrays.asList(
777                         new MockRaftActorContext.MockPayload("A"),
778                         new MockRaftActorContext.MockPayload("B"),
779                         new MockRaftActorContext.MockPayload("C"),
780                         new MockRaftActorContext.MockPayload("D")));
781
782                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
783                 mockRaftActor.setCurrentBehavior(new Follower(raftActorContext));
784
785                 long replicatedToAllIndex = 1;
786                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1, replicatedToAllIndex, 1));
787
788                 verify(mockRaftActor.delegate).createSnapshot();
789
790                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
791
792                 mockRaftActor.onReceiveCommand(new SaveSnapshotSuccess(new SnapshotMetadata("foo", 100, 100)));
793
794                 verify(dataPersistenceProvider).deleteSnapshots(any(SnapshotSelectionCriteria.class));
795
796                 verify(dataPersistenceProvider).deleteMessages(100);
797
798                 assertEquals(3, mockRaftActor.getReplicatedLog().size());
799                 assertEquals(1, mockRaftActor.getCurrentBehavior().getReplicatedToAllIndex());
800
801                 assertNotNull(mockRaftActor.getReplicatedLog().get(2));
802                 assertNotNull(mockRaftActor.getReplicatedLog().get(3));
803                 assertNotNull(mockRaftActor.getReplicatedLog().get(4));
804
805                 // Index 2 will not be in the log because it was removed due to snapshotting
806                 assertNull(mockRaftActor.getReplicatedLog().get(1));
807                 assertNull(mockRaftActor.getReplicatedLog().get(0));
808
809             }
810         };
811     }
812
813     @Test
814     public void testApplyState() throws Exception {
815
816         new JavaTestKit(getSystem()) {
817             {
818                 String persistenceId = factory.generateActorId("leader-");
819
820                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
821
822                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
823
824                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
825
826                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
827                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
828
829                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
830
831                 mockRaftActor.waitForInitializeBehaviorComplete();
832
833                 ReplicatedLogEntry entry = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
834                         new MockRaftActorContext.MockPayload("F"));
835
836                 mockRaftActor.onReceiveCommand(new ApplyState(mockActorRef, "apply-state", entry));
837
838                 verify(mockRaftActor.delegate).applyState(eq(mockActorRef), eq("apply-state"), anyObject());
839
840             }
841         };
842     }
843
844     @Test
845     public void testApplySnapshot() throws Exception {
846         new JavaTestKit(getSystem()) {
847             {
848                 String persistenceId = factory.generateActorId("leader-");
849
850                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
851
852                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
853
854                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
855
856                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
857                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
858
859                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
860
861                 mockRaftActor.waitForInitializeBehaviorComplete();
862
863                 ReplicatedLog oldReplicatedLog = mockRaftActor.getReplicatedLog();
864
865                 oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
866                 oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class)));
867                 oldReplicatedLog.append(
868                         new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
869                                 mock(Payload.class)));
870
871                 ByteString snapshotBytes = fromObject(Arrays.asList(
872                         new MockRaftActorContext.MockPayload("A"),
873                         new MockRaftActorContext.MockPayload("B"),
874                         new MockRaftActorContext.MockPayload("C"),
875                         new MockRaftActorContext.MockPayload("D")));
876
877                 Snapshot snapshot = mock(Snapshot.class);
878
879                 doReturn(snapshotBytes.toByteArray()).when(snapshot).getState();
880
881                 doReturn(3L).when(snapshot).getLastAppliedIndex();
882
883                 mockRaftActor.onReceiveCommand(new ApplySnapshot(snapshot));
884
885                 verify(mockRaftActor.delegate).applySnapshot(eq(snapshot.getState()));
886
887                 assertTrue("The replicatedLog should have changed",
888                         oldReplicatedLog != mockRaftActor.getReplicatedLog());
889
890                 assertEquals("lastApplied should be same as in the snapshot",
891                         (Long) 3L, mockRaftActor.getLastApplied());
892
893                 assertEquals(0, mockRaftActor.getReplicatedLog().size());
894
895             }
896         };
897     }
898
899     @Test
900     public void testSaveSnapshotFailure() throws Exception {
901         new JavaTestKit(getSystem()) {
902             {
903                 String persistenceId = factory.generateActorId("leader-");
904
905                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
906
907                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
908
909                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
910
911                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId,
912                         Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
913
914                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
915
916                 mockRaftActor.waitForInitializeBehaviorComplete();
917
918                 ByteString snapshotBytes = fromObject(Arrays.asList(
919                         new MockRaftActorContext.MockPayload("A"),
920                         new MockRaftActorContext.MockPayload("B"),
921                         new MockRaftActorContext.MockPayload("C"),
922                         new MockRaftActorContext.MockPayload("D")));
923
924                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
925
926                 mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
927
928                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1, -1, 1));
929
930                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
931
932                 mockRaftActor.onReceiveCommand(new SaveSnapshotFailure(new SnapshotMetadata("foobar", 10L, 1234L),
933                         new Exception()));
934
935                 assertEquals("Snapshot index should not have advanced because save snapshot failed", -1,
936                         mockRaftActor.getReplicatedLog().getSnapshotIndex());
937
938             }
939         };
940     }
941
942     @Test
943     public void testRaftRoleChangeNotifier() throws Exception {
944         new JavaTestKit(getSystem()) {{
945             ActorRef notifierActor = factory.createActor(Props.create(MessageCollectorActor.class));
946             MessageCollectorActor.waitUntilReady(notifierActor);
947
948             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
949             long heartBeatInterval = 100;
950             config.setHeartBeatInterval(FiniteDuration.create(heartBeatInterval, TimeUnit.MILLISECONDS));
951             config.setElectionTimeoutFactor(1);
952
953             String persistenceId = factory.generateActorId("notifier-");
954
955             factory.createTestActor(MockRaftActor.props(persistenceId,
956                     Collections.<String, String>emptyMap(), Optional.<ConfigParams>of(config), notifierActor), persistenceId);
957
958             List<RoleChanged> matches =  null;
959             for(int i = 0; i < 5000 / heartBeatInterval; i++) {
960                 matches = MessageCollectorActor.getAllMatching(notifierActor, RoleChanged.class);
961                 assertNotNull(matches);
962                 if(matches.size() == 3) {
963                     break;
964                 }
965                 Uninterruptibles.sleepUninterruptibly(heartBeatInterval, TimeUnit.MILLISECONDS);
966             }
967
968             assertEquals(3, matches.size());
969
970             // check if the notifier got a role change from null to Follower
971             RoleChanged raftRoleChanged = matches.get(0);
972             assertEquals(persistenceId, raftRoleChanged.getMemberId());
973             assertNull(raftRoleChanged.getOldRole());
974             assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole());
975
976             // check if the notifier got a role change from Follower to Candidate
977             raftRoleChanged = matches.get(1);
978             assertEquals(persistenceId, raftRoleChanged.getMemberId());
979             assertEquals(RaftState.Follower.name(), raftRoleChanged.getOldRole());
980             assertEquals(RaftState.Candidate.name(), raftRoleChanged.getNewRole());
981
982             // check if the notifier got a role change from Candidate to Leader
983             raftRoleChanged = matches.get(2);
984             assertEquals(persistenceId, raftRoleChanged.getMemberId());
985             assertEquals(RaftState.Candidate.name(), raftRoleChanged.getOldRole());
986             assertEquals(RaftState.Leader.name(), raftRoleChanged.getNewRole());
987         }};
988     }
989
990     @Test
991     public void testFakeSnapshotsForLeaderWithInRealSnapshots() throws Exception {
992         new JavaTestKit(getSystem()) {
993             {
994                 String persistenceId = factory.generateActorId("leader-");
995                 String follower1Id = factory.generateActorId("follower-");
996
997                 ActorRef followerActor1 =
998                         factory.createActor(Props.create(MessageCollectorActor.class));
999
1000                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1001                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
1002                 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
1003
1004                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
1005
1006                 Map<String, String> peerAddresses = new HashMap<>();
1007                 peerAddresses.put(follower1Id, followerActor1.path().toString());
1008
1009                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
1010                         MockRaftActor.props(persistenceId, peerAddresses,
1011                                 Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
1012
1013                 MockRaftActor leaderActor = mockActorRef.underlyingActor();
1014
1015                 leaderActor.getRaftActorContext().setCommitIndex(4);
1016                 leaderActor.getRaftActorContext().setLastApplied(4);
1017                 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
1018
1019                 leaderActor.waitForInitializeBehaviorComplete();
1020
1021                 // create 8 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
1022
1023                 Leader leader = new Leader(leaderActor.getRaftActorContext());
1024                 leaderActor.setCurrentBehavior(leader);
1025                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1026
1027                 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
1028                 leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 8, 1).build());
1029
1030                 assertEquals(8, leaderActor.getReplicatedLog().size());
1031
1032                 leaderActor.onReceiveCommand(new CaptureSnapshot(6, 1, 4, 1, 4, 1));
1033
1034                 leaderActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
1035                 verify(leaderActor.delegate).createSnapshot();
1036
1037                 assertEquals(8, leaderActor.getReplicatedLog().size());
1038
1039                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1040                 //fake snapshot on index 5
1041                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1));
1042
1043                 assertEquals(8, leaderActor.getReplicatedLog().size());
1044
1045                 //fake snapshot on index 6
1046                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1047                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1));
1048                 assertEquals(8, leaderActor.getReplicatedLog().size());
1049
1050                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1051
1052                 assertEquals(8, leaderActor.getReplicatedLog().size());
1053
1054                 ByteString snapshotBytes = fromObject(Arrays.asList(
1055                         new MockRaftActorContext.MockPayload("foo-0"),
1056                         new MockRaftActorContext.MockPayload("foo-1"),
1057                         new MockRaftActorContext.MockPayload("foo-2"),
1058                         new MockRaftActorContext.MockPayload("foo-3"),
1059                         new MockRaftActorContext.MockPayload("foo-4")));
1060                 leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
1061                 assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated());
1062
1063                 // capture snapshot reply should remove the snapshotted entries only
1064                 assertEquals(3, leaderActor.getReplicatedLog().size());
1065                 assertEquals(7, leaderActor.getReplicatedLog().lastIndex());
1066
1067                 // add another non-replicated entry
1068                 leaderActor.getReplicatedLog().append(
1069                         new ReplicatedLogImplEntry(8, 1, new MockRaftActorContext.MockPayload("foo-8")));
1070
1071                 //fake snapshot on index 7, since lastApplied = 7 , we would keep the last applied
1072                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 7, 1));
1073                 assertEquals(2, leaderActor.getReplicatedLog().size());
1074                 assertEquals(8, leaderActor.getReplicatedLog().lastIndex());
1075
1076             }
1077         };
1078     }
1079
1080     @Test
1081     public void testFakeSnapshotsForFollowerWithInRealSnapshots() throws Exception {
1082         new JavaTestKit(getSystem()) {
1083             {
1084                 String persistenceId = factory.generateActorId("follower-");
1085                 String leaderId = factory.generateActorId("leader-");
1086
1087
1088                 ActorRef leaderActor1 =
1089                         factory.createActor(Props.create(MessageCollectorActor.class));
1090
1091                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1092                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
1093                 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
1094
1095                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
1096
1097                 Map<String, String> peerAddresses = new HashMap<>();
1098                 peerAddresses.put(leaderId, leaderActor1.path().toString());
1099
1100                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
1101                         MockRaftActor.props(persistenceId, peerAddresses,
1102                                 Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
1103
1104                 MockRaftActor followerActor = mockActorRef.underlyingActor();
1105                 followerActor.getRaftActorContext().setCommitIndex(4);
1106                 followerActor.getRaftActorContext().setLastApplied(4);
1107                 followerActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
1108
1109                 followerActor.waitForInitializeBehaviorComplete();
1110
1111
1112                 Follower follower = new Follower(followerActor.getRaftActorContext());
1113                 followerActor.setCurrentBehavior(follower);
1114                 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
1115
1116                 // create 6 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot
1117                 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
1118                 followerActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 6, 1).build());
1119
1120                 // log has indices 0-5
1121                 assertEquals(6, followerActor.getReplicatedLog().size());
1122
1123                 //snapshot on 4
1124                 followerActor.onReceiveCommand(new CaptureSnapshot(5, 1, 4, 1, 4, 1));
1125
1126                 followerActor.getRaftActorContext().setSnapshotCaptureInitiated(true);
1127                 verify(followerActor.delegate).createSnapshot();
1128
1129                 assertEquals(6, followerActor.getReplicatedLog().size());
1130
1131                 //fake snapshot on index 6
1132                 List<ReplicatedLogEntry> entries =
1133                         Arrays.asList(
1134                                 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
1135                                         new MockRaftActorContext.MockPayload("foo-6"))
1136                         );
1137                 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5));
1138                 assertEquals(7, followerActor.getReplicatedLog().size());
1139
1140                 //fake snapshot on index 7
1141                 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
1142
1143                 entries =
1144                         Arrays.asList(
1145                                 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
1146                                         new MockRaftActorContext.MockPayload("foo-7"))
1147                         );
1148                 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6));
1149                 assertEquals(8, followerActor.getReplicatedLog().size());
1150
1151                 assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state());
1152
1153
1154                 ByteString snapshotBytes = fromObject(Arrays.asList(
1155                         new MockRaftActorContext.MockPayload("foo-0"),
1156                         new MockRaftActorContext.MockPayload("foo-1"),
1157                         new MockRaftActorContext.MockPayload("foo-2"),
1158                         new MockRaftActorContext.MockPayload("foo-3"),
1159                         new MockRaftActorContext.MockPayload("foo-4")));
1160                 followerActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
1161                 assertFalse(followerActor.getRaftActorContext().isSnapshotCaptureInitiated());
1162
1163                 // capture snapshot reply should remove the snapshotted entries only till replicatedToAllIndex
1164                 assertEquals(3, followerActor.getReplicatedLog().size()); //indexes 5,6,7 left in the log
1165                 assertEquals(7, followerActor.getReplicatedLog().lastIndex());
1166
1167                 entries =
1168                         Arrays.asList(
1169                                 (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 8,
1170                                         new MockRaftActorContext.MockPayload("foo-7"))
1171                         );
1172                 // send an additional entry 8 with leaderCommit = 7
1173                 followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7));
1174
1175                 // 7 and 8, as lastapplied is 7
1176                 assertEquals(2, followerActor.getReplicatedLog().size());
1177
1178             }
1179         };
1180     }
1181
1182     @Test
1183     public void testFakeSnapshotsForLeaderWithInInitiateSnapshots() throws Exception {
1184         new JavaTestKit(getSystem()) {
1185             {
1186                 String persistenceId = factory.generateActorId("leader-");
1187                 String follower1Id = factory.generateActorId("follower-");
1188                 String follower2Id = factory.generateActorId("follower-");
1189
1190                 ActorRef followerActor1 =
1191                         factory.createActor(Props.create(MessageCollectorActor.class), follower1Id);
1192                 ActorRef followerActor2 =
1193                         factory.createActor(Props.create(MessageCollectorActor.class), follower2Id);
1194
1195                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1196                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
1197                 config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
1198
1199                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
1200
1201                 Map<String, String> peerAddresses = new HashMap<>();
1202                 peerAddresses.put(follower1Id, followerActor1.path().toString());
1203                 peerAddresses.put(follower2Id, followerActor2.path().toString());
1204
1205                 TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
1206                         MockRaftActor.props(persistenceId, peerAddresses,
1207                                 Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
1208
1209                 MockRaftActor leaderActor = mockActorRef.underlyingActor();
1210                 leaderActor.getRaftActorContext().setCommitIndex(9);
1211                 leaderActor.getRaftActorContext().setLastApplied(9);
1212                 leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
1213
1214                 leaderActor.waitForInitializeBehaviorComplete();
1215
1216                 Leader leader = new Leader(leaderActor.getRaftActorContext());
1217                 leaderActor.setCurrentBehavior(leader);
1218                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1219
1220                 // create 5 entries in the log
1221                 MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder();
1222                 leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(5, 10, 1).build());
1223
1224                 //set the snapshot index to 4 , 0 to 4 are snapshotted
1225                 leaderActor.getRaftActorContext().getReplicatedLog().setSnapshotIndex(4);
1226                 //setting replicatedToAllIndex = 9, for the log to clear
1227                 leader.setReplicatedToAllIndex(9);
1228                 assertEquals(5, leaderActor.getReplicatedLog().size());
1229                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1230
1231                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1));
1232                 assertEquals(5, leaderActor.getReplicatedLog().size());
1233                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1234
1235                 // set the 2nd follower nextIndex to 1 which has been snapshotted
1236                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1));
1237                 assertEquals(5, leaderActor.getReplicatedLog().size());
1238                 assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1239
1240                 // simulate a real snapshot
1241                 leaderActor.onReceiveCommand(new SendHeartBeat());
1242                 assertEquals(5, leaderActor.getReplicatedLog().size());
1243                 assertEquals(String.format("expected to be Leader but was %s. Current Leader = %s ",
1244                         leaderActor.getCurrentBehavior().state(), leaderActor.getLeaderId())
1245                         , RaftState.Leader, leaderActor.getCurrentBehavior().state());
1246
1247
1248                 //reply from a slow follower does not initiate a fake snapshot
1249                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1));
1250                 assertEquals("Fake snapshot should not happen when Initiate is in progress", 5, leaderActor.getReplicatedLog().size());
1251
1252                 ByteString snapshotBytes = fromObject(Arrays.asList(
1253                         new MockRaftActorContext.MockPayload("foo-0"),
1254                         new MockRaftActorContext.MockPayload("foo-1"),
1255                         new MockRaftActorContext.MockPayload("foo-2"),
1256                         new MockRaftActorContext.MockPayload("foo-3"),
1257                         new MockRaftActorContext.MockPayload("foo-4")));
1258                 leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray()));
1259                 assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated());
1260
1261                 assertEquals("Real snapshot didn't clear the log till replicatedToAllIndex", 0, leaderActor.getReplicatedLog().size());
1262
1263                 //reply from a slow follower after should not raise errors
1264                 leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1));
1265                 assertEquals(0, leaderActor.getReplicatedLog().size());
1266             }
1267         };
1268     }
1269
1270
1271     private static class NonPersistentProvider implements DataPersistenceProvider {
1272         @Override
1273         public boolean isRecoveryApplicable() {
1274             return false;
1275         }
1276
1277         @Override
1278         public <T> void persist(T o, Procedure<T> procedure) {
1279             try {
1280                 procedure.apply(o);
1281             } catch (Exception e) {
1282                 e.printStackTrace();
1283             }
1284         }
1285
1286         @Override
1287         public void saveSnapshot(Object o) {
1288
1289         }
1290
1291         @Override
1292         public void deleteSnapshots(SnapshotSelectionCriteria criteria) {
1293
1294         }
1295
1296         @Override
1297         public void deleteMessages(long sequenceNumber) {
1298
1299         }
1300     }
1301
1302     @Test
1303     public void testRealSnapshotWhenReplicatedToAllIndexMinusOne() throws Exception {
1304         new JavaTestKit(getSystem()) {{
1305             String persistenceId = factory.generateActorId("leader-");
1306             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1307             config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
1308             config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
1309             config.setSnapshotBatchCount(5);
1310
1311             DataPersistenceProvider dataPersistenceProvider = new NonPersistentProvider();
1312
1313             Map<String, String> peerAddresses = new HashMap<>();
1314
1315             TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
1316                     MockRaftActor.props(persistenceId, peerAddresses,
1317                             Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
1318
1319             MockRaftActor leaderActor = mockActorRef.underlyingActor();
1320             leaderActor.getRaftActorContext().setCommitIndex(3);
1321             leaderActor.getRaftActorContext().setLastApplied(3);
1322             leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
1323
1324             leaderActor.waitForInitializeBehaviorComplete();
1325             for(int i=0;i< 4;i++) {
1326                 leaderActor.getReplicatedLog()
1327                         .append(new MockRaftActorContext.MockReplicatedLogEntry(1, i,
1328                                 new MockRaftActorContext.MockPayload("A")));
1329             }
1330
1331             Leader leader = new Leader(leaderActor.getRaftActorContext());
1332             leaderActor.setCurrentBehavior(leader);
1333             assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1334
1335             // Persist another entry (this will cause a CaptureSnapshot to be triggered
1336             leaderActor.persistData(mockActorRef, "x", new MockRaftActorContext.MockPayload("duh"));
1337
1338             // Now send a CaptureSnapshotReply
1339             mockActorRef.tell(new CaptureSnapshotReply(fromObject("foo").toByteArray()), mockActorRef);
1340
1341             // Trimming log in this scenario is a no-op
1342             assertEquals(-1, leaderActor.getReplicatedLog().getSnapshotIndex());
1343             assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated());
1344             assertEquals(-1, leader.getReplicatedToAllIndex());
1345
1346         }};
1347     }
1348
1349     @Test
1350     public void testRealSnapshotWhenReplicatedToAllIndexNotInReplicatedLog() throws Exception {
1351         new JavaTestKit(getSystem()) {{
1352             String persistenceId = factory.generateActorId("leader-");
1353             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1354             config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
1355             config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS));
1356             config.setSnapshotBatchCount(5);
1357
1358             DataPersistenceProvider dataPersistenceProvider = new NonPersistentProvider();
1359
1360             Map<String, String> peerAddresses = new HashMap<>();
1361
1362             TestActorRef<MockRaftActor> mockActorRef = factory.createTestActor(
1363                     MockRaftActor.props(persistenceId, peerAddresses,
1364                             Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
1365
1366             MockRaftActor leaderActor = mockActorRef.underlyingActor();
1367             leaderActor.getRaftActorContext().setCommitIndex(3);
1368             leaderActor.getRaftActorContext().setLastApplied(3);
1369             leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId);
1370             leaderActor.getReplicatedLog().setSnapshotIndex(3);
1371
1372             leaderActor.waitForInitializeBehaviorComplete();
1373             Leader leader = new Leader(leaderActor.getRaftActorContext());
1374             leaderActor.setCurrentBehavior(leader);
1375             leader.setReplicatedToAllIndex(3);
1376             assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state());
1377
1378             // Persist another entry (this will cause a CaptureSnapshot to be triggered
1379             leaderActor.persistData(mockActorRef, "x", new MockRaftActorContext.MockPayload("duh"));
1380
1381             // Now send a CaptureSnapshotReply
1382             mockActorRef.tell(new CaptureSnapshotReply(fromObject("foo").toByteArray()), mockActorRef);
1383
1384             // Trimming log in this scenario is a no-op
1385             assertEquals(3, leaderActor.getReplicatedLog().getSnapshotIndex());
1386             assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated());
1387             assertEquals(3, leader.getReplicatedToAllIndex());
1388
1389         }};
1390     }
1391
1392     private ByteString fromObject(Object snapshot) throws Exception {
1393         ByteArrayOutputStream b = null;
1394         ObjectOutputStream o = null;
1395         try {
1396             b = new ByteArrayOutputStream();
1397             o = new ObjectOutputStream(b);
1398             o.writeObject(snapshot);
1399             byte[] snapshotBytes = b.toByteArray();
1400             return ByteString.copyFrom(snapshotBytes);
1401         } finally {
1402             if (o != null) {
1403                 o.flush();
1404                 o.close();
1405             }
1406             if (b != null) {
1407                 b.close();
1408             }
1409         }
1410     }
1411
1412 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.