9eb2fb757bd2da0c6b2ce48ebbfeb72471ef48c1
[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 akka.actor.ActorRef;
4 import akka.actor.ActorSystem;
5 import akka.actor.PoisonPill;
6 import akka.actor.Props;
7 import akka.actor.Terminated;
8 import akka.japi.Creator;
9 import akka.japi.Procedure;
10 import akka.pattern.Patterns;
11 import akka.persistence.RecoveryCompleted;
12 import akka.persistence.SaveSnapshotFailure;
13 import akka.persistence.SaveSnapshotSuccess;
14 import akka.persistence.SnapshotMetadata;
15 import akka.persistence.SnapshotOffer;
16 import akka.persistence.SnapshotSelectionCriteria;
17 import akka.testkit.JavaTestKit;
18 import akka.testkit.TestActorRef;
19 import akka.util.Timeout;
20 import com.google.common.base.Optional;
21 import com.google.common.collect.Lists;
22 import com.google.common.util.concurrent.Uninterruptibles;
23 import com.google.protobuf.ByteString;
24 import org.junit.After;
25 import org.junit.Assert;
26 import org.junit.Test;
27 import org.opendaylight.controller.cluster.DataPersistenceProvider;
28 import org.opendaylight.controller.cluster.datastore.DataPersistenceProviderMonitor;
29 import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
30 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
31 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
32 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
33 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
34 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
35 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
36 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
37 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
38 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
39 import org.opendaylight.controller.cluster.raft.utils.MockAkkaJournal;
40 import org.opendaylight.controller.cluster.raft.utils.MockSnapshotStore;
41 import scala.concurrent.Await;
42 import scala.concurrent.Future;
43 import scala.concurrent.duration.Duration;
44 import scala.concurrent.duration.FiniteDuration;
45
46 import java.io.ByteArrayInputStream;
47 import java.io.ByteArrayOutputStream;
48 import java.io.IOException;
49 import java.io.ObjectInputStream;
50 import java.io.ObjectOutputStream;
51 import java.util.ArrayList;
52 import java.util.Arrays;
53 import java.util.Collections;
54 import java.util.List;
55 import java.util.Map;
56 import java.util.concurrent.CountDownLatch;
57 import java.util.concurrent.TimeUnit;
58 import java.util.concurrent.TimeoutException;
59
60 import static org.junit.Assert.assertEquals;
61 import static org.junit.Assert.assertNotEquals;
62 import static org.junit.Assert.assertNotNull;
63 import static org.junit.Assert.assertNull;
64 import static org.junit.Assert.assertTrue;
65 import static org.mockito.Matchers.any;
66 import static org.mockito.Matchers.anyObject;
67 import static org.mockito.Matchers.eq;
68 import static org.mockito.Mockito.doReturn;
69 import static org.mockito.Mockito.mock;
70 import static org.mockito.Mockito.times;
71 import static org.mockito.Mockito.verify;
72
73 public class RaftActorTest extends AbstractActorTest {
74
75
76     @After
77     public void tearDown() {
78         MockAkkaJournal.clearJournal();
79         MockSnapshotStore.setMockSnapshot(null);
80     }
81
82     public static class MockRaftActor extends RaftActor {
83
84         private final DataPersistenceProvider dataPersistenceProvider;
85         private final RaftActor delegate;
86
87         public static final class MockRaftActorCreator implements Creator<MockRaftActor> {
88             private static final long serialVersionUID = 1L;
89             private final Map<String, String> peerAddresses;
90             private final String id;
91             private final Optional<ConfigParams> config;
92             private final DataPersistenceProvider dataPersistenceProvider;
93
94             private MockRaftActorCreator(Map<String, String> peerAddresses, String id,
95                     Optional<ConfigParams> config, DataPersistenceProvider dataPersistenceProvider) {
96                 this.peerAddresses = peerAddresses;
97                 this.id = id;
98                 this.config = config;
99                 this.dataPersistenceProvider = dataPersistenceProvider;
100             }
101
102             @Override
103             public MockRaftActor create() throws Exception {
104                 return new MockRaftActor(id, peerAddresses, config, dataPersistenceProvider);
105             }
106         }
107
108         private final CountDownLatch recoveryComplete = new CountDownLatch(1);
109
110         private final List<Object> state;
111
112         public MockRaftActor(String id, Map<String, String> peerAddresses, Optional<ConfigParams> config, DataPersistenceProvider dataPersistenceProvider) {
113             super(id, peerAddresses, config);
114             state = new ArrayList<>();
115             this.delegate = mock(RaftActor.class);
116             if(dataPersistenceProvider == null){
117                 this.dataPersistenceProvider = new PersistentDataProvider();
118             } else {
119                 this.dataPersistenceProvider = dataPersistenceProvider;
120             }
121         }
122
123         public void waitForRecoveryComplete() {
124             try {
125                 assertEquals("Recovery complete", true, recoveryComplete.await(5,  TimeUnit.SECONDS));
126             } catch (InterruptedException e) {
127                 e.printStackTrace();
128             }
129         }
130
131         public List<Object> getState() {
132             return state;
133         }
134
135         public static Props props(final String id, final Map<String, String> peerAddresses,
136                 Optional<ConfigParams> config){
137             return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null));
138         }
139
140         public static Props props(final String id, final Map<String, String> peerAddresses,
141                                   Optional<ConfigParams> config, DataPersistenceProvider dataPersistenceProvider){
142             return Props.create(new MockRaftActorCreator(peerAddresses, id, config, dataPersistenceProvider));
143         }
144
145
146         @Override protected void applyState(ActorRef clientActor, String identifier, Object data) {
147             delegate.applyState(clientActor, identifier, data);
148             LOG.info("applyState called");
149         }
150
151
152
153
154         @Override
155         protected void startLogRecoveryBatch(int maxBatchSize) {
156         }
157
158         @Override
159         protected void appendRecoveredLogEntry(Payload data) {
160             state.add(data);
161         }
162
163         @Override
164         protected void applyCurrentLogRecoveryBatch() {
165         }
166
167         @Override
168         protected void onRecoveryComplete() {
169             delegate.onRecoveryComplete();
170             recoveryComplete.countDown();
171         }
172
173         @Override
174         protected void applyRecoverySnapshot(ByteString snapshot) {
175             delegate.applyRecoverySnapshot(snapshot);
176             try {
177                 Object data = toObject(snapshot);
178                 System.out.println("!!!!!applyRecoverySnapshot: "+data);
179                 if (data instanceof List) {
180                     state.addAll((List<?>) data);
181                 }
182             } catch (Exception e) {
183                 e.printStackTrace();
184             }
185         }
186
187         @Override protected void createSnapshot() {
188             delegate.createSnapshot();
189         }
190
191         @Override protected void applySnapshot(ByteString snapshot) {
192             delegate.applySnapshot(snapshot);
193         }
194
195         @Override protected void onStateChanged() {
196             delegate.onStateChanged();
197         }
198
199         @Override
200         protected DataPersistenceProvider persistence() {
201             return this.dataPersistenceProvider;
202         }
203
204         @Override public String persistenceId() {
205             return this.getId();
206         }
207
208         private Object toObject(ByteString bs) throws ClassNotFoundException, IOException {
209             Object obj = null;
210             ByteArrayInputStream bis = null;
211             ObjectInputStream ois = null;
212             try {
213                 bis = new ByteArrayInputStream(bs.toByteArray());
214                 ois = new ObjectInputStream(bis);
215                 obj = ois.readObject();
216             } finally {
217                 if (bis != null) {
218                     bis.close();
219                 }
220                 if (ois != null) {
221                     ois.close();
222                 }
223             }
224             return obj;
225         }
226
227         public ReplicatedLog getReplicatedLog(){
228             return this.getRaftActorContext().getReplicatedLog();
229         }
230
231     }
232
233
234     private static class RaftActorTestKit extends JavaTestKit {
235         private final ActorRef raftActor;
236
237         public RaftActorTestKit(ActorSystem actorSystem, String actorName) {
238             super(actorSystem);
239
240             raftActor = this.getSystem().actorOf(MockRaftActor.props(actorName,
241                     Collections.<String,String>emptyMap(), Optional.<ConfigParams>absent()), actorName);
242
243         }
244
245
246         public ActorRef getRaftActor() {
247             return raftActor;
248         }
249
250         public boolean waitForLogMessage(final Class<?> logEventClass, String message){
251             // Wait for a specific log message to show up
252             return
253                 new JavaTestKit.EventFilter<Boolean>(logEventClass
254                 ) {
255                     @Override
256                     protected Boolean run() {
257                         return true;
258                     }
259                 }.from(raftActor.path().toString())
260                     .message(message)
261                     .occurrences(1).exec();
262
263
264         }
265
266         protected void waitUntilLeader(){
267             waitUntilLeader(raftActor);
268         }
269
270         protected void waitUntilLeader(ActorRef actorRef) {
271             FiniteDuration duration = Duration.create(100, TimeUnit.MILLISECONDS);
272             for(int i = 0; i < 20 * 5; i++) {
273                 Future<Object> future = Patterns.ask(actorRef, new FindLeader(), new Timeout(duration));
274                 try {
275                     FindLeaderReply resp = (FindLeaderReply) Await.result(future, duration);
276                     if(resp.getLeaderActor() != null) {
277                         return;
278                     }
279                 } catch(TimeoutException e) {
280                 } catch(Exception e) {
281                     System.err.println("FindLeader threw ex");
282                     e.printStackTrace();
283                 }
284
285
286                 Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
287             }
288
289             Assert.fail("Leader not found for actorRef " + actorRef.path());
290         }
291
292     }
293
294
295     @Test
296     public void testConstruction() {
297         new RaftActorTestKit(getSystem(), "testConstruction").waitUntilLeader();
298     }
299
300     @Test
301     public void testFindLeaderWhenLeaderIsSelf(){
302         RaftActorTestKit kit = new RaftActorTestKit(getSystem(), "testFindLeader");
303         kit.waitUntilLeader();
304     }
305
306     @Test
307     public void testRaftActorRecovery() throws Exception {
308         new JavaTestKit(getSystem()) {{
309             String persistenceId = "follower10";
310
311             DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
312             // Set the heartbeat interval high to essentially disable election otherwise the test
313             // may fail if the actor is switched to Leader and the commitIndex is set to the last
314             // log entry.
315             config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
316
317             ActorRef followerActor = getSystem().actorOf(MockRaftActor.props(persistenceId,
318                     Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
319
320             watch(followerActor);
321
322             List<ReplicatedLogEntry> snapshotUnappliedEntries = new ArrayList<>();
323             ReplicatedLogEntry entry1 = new MockRaftActorContext.MockReplicatedLogEntry(1, 4,
324                     new MockRaftActorContext.MockPayload("E"));
325             snapshotUnappliedEntries.add(entry1);
326
327             int lastAppliedDuringSnapshotCapture = 3;
328             int lastIndexDuringSnapshotCapture = 4;
329
330                 // 4 messages as part of snapshot, which are applied to state
331             ByteString snapshotBytes  = fromObject(Arrays.asList(
332                         new MockRaftActorContext.MockPayload("A"),
333                         new MockRaftActorContext.MockPayload("B"),
334                         new MockRaftActorContext.MockPayload("C"),
335                         new MockRaftActorContext.MockPayload("D")));
336
337             Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
338                     snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1 ,
339                     lastAppliedDuringSnapshotCapture, 1);
340             MockSnapshotStore.setMockSnapshot(snapshot);
341             MockSnapshotStore.setPersistenceId(persistenceId);
342
343             // add more entries after snapshot is taken
344             List<ReplicatedLogEntry> entries = new ArrayList<>();
345             ReplicatedLogEntry entry2 = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
346                     new MockRaftActorContext.MockPayload("F"));
347             ReplicatedLogEntry entry3 = new MockRaftActorContext.MockReplicatedLogEntry(1, 6,
348                     new MockRaftActorContext.MockPayload("G"));
349             ReplicatedLogEntry entry4 = new MockRaftActorContext.MockReplicatedLogEntry(1, 7,
350                     new MockRaftActorContext.MockPayload("H"));
351             entries.add(entry2);
352             entries.add(entry3);
353             entries.add(entry4);
354
355             int lastAppliedToState = 5;
356             int lastIndex = 7;
357
358             MockAkkaJournal.addToJournal(5, entry2);
359             // 2 entries are applied to state besides the 4 entries in snapshot
360             MockAkkaJournal.addToJournal(6, new ApplyLogEntries(lastAppliedToState));
361             MockAkkaJournal.addToJournal(7, entry3);
362             MockAkkaJournal.addToJournal(8, entry4);
363
364             // kill the actor
365             followerActor.tell(PoisonPill.getInstance(), null);
366             expectMsgClass(duration("5 seconds"), Terminated.class);
367
368             unwatch(followerActor);
369
370             //reinstate the actor
371             TestActorRef<MockRaftActor> ref = TestActorRef.create(getSystem(),
372                     MockRaftActor.props(persistenceId, Collections.<String,String>emptyMap(),
373                             Optional.<ConfigParams>of(config)));
374
375             ref.underlyingActor().waitForRecoveryComplete();
376
377             RaftActorContext context = ref.underlyingActor().getRaftActorContext();
378             assertEquals("Journal log size", snapshotUnappliedEntries.size() + entries.size(),
379                     context.getReplicatedLog().size());
380             assertEquals("Last index", lastIndex, context.getReplicatedLog().lastIndex());
381             assertEquals("Last applied", lastAppliedToState, context.getLastApplied());
382             assertEquals("Commit index", lastAppliedToState, context.getCommitIndex());
383             assertEquals("Recovered state size", 6, ref.underlyingActor().getState().size());
384         }};
385     }
386
387     /**
388      * This test verifies that when recovery is applicable (typically when persistence is true) the RaftActor does
389      * process recovery messages
390      *
391      * @throws Exception
392      */
393
394     @Test
395     public void testHandleRecoveryWhenDataPersistenceRecoveryApplicable() throws Exception {
396         new JavaTestKit(getSystem()) {
397             {
398                 String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryApplicable";
399
400                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
401
402                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
403
404                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
405                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config)), persistenceId);
406
407                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
408
409                 // Wait for akka's recovery to complete so it doesn't interfere.
410                 mockRaftActor.waitForRecoveryComplete();
411
412                 ByteString snapshotBytes  = fromObject(Arrays.asList(
413                         new MockRaftActorContext.MockPayload("A"),
414                         new MockRaftActorContext.MockPayload("B"),
415                         new MockRaftActorContext.MockPayload("C"),
416                         new MockRaftActorContext.MockPayload("D")));
417
418                 Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
419                         Lists.<ReplicatedLogEntry>newArrayList(), 3, 1 ,3, 1);
420
421                 mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
422
423                 verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes));
424
425                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
426
427                 ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
428
429                 assertEquals("add replicated log entry", 1, replicatedLog.size());
430
431                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
432
433                 assertEquals("add replicated log entry", 2, replicatedLog.size());
434
435                 mockRaftActor.onReceiveRecover(new ApplyLogEntries(1));
436
437                 assertEquals("commit index 1", 1, mockRaftActor.getRaftActorContext().getCommitIndex());
438
439                 // The snapshot had 4 items + we added 2 more items during the test
440                 // We start removing from 5 and we should get 1 item in the replicated log
441                 mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(5));
442
443                 assertEquals("remove log entries", 1, replicatedLog.size());
444
445                 mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
446
447                 assertEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
448                 assertEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
449
450                 mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
451
452                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
453
454             }};
455     }
456
457     /**
458      * This test verifies that when recovery is not applicable (typically when persistence is false) the RaftActor does
459      * not process recovery messages
460      *
461      * @throws Exception
462      */
463     @Test
464     public void testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable() throws Exception {
465         new JavaTestKit(getSystem()) {
466             {
467                 String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable";
468
469                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
470
471                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
472
473                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
474                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), new DataPersistenceProviderMonitor()), persistenceId);
475
476                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
477
478                 // Wait for akka's recovery to complete so it doesn't interfere.
479                 mockRaftActor.waitForRecoveryComplete();
480
481                 ByteString snapshotBytes  = fromObject(Arrays.asList(
482                         new MockRaftActorContext.MockPayload("A"),
483                         new MockRaftActorContext.MockPayload("B"),
484                         new MockRaftActorContext.MockPayload("C"),
485                         new MockRaftActorContext.MockPayload("D")));
486
487                 Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(),
488                         Lists.<ReplicatedLogEntry>newArrayList(), 3, 1 ,3, 1);
489
490                 mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot));
491
492                 verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(ByteString.class));
493
494                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A")));
495
496                 ReplicatedLog replicatedLog = mockRaftActor.getReplicatedLog();
497
498                 assertEquals("add replicated log entry", 0, replicatedLog.size());
499
500                 mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(1, 1, new MockRaftActorContext.MockPayload("A")));
501
502                 assertEquals("add replicated log entry", 0, replicatedLog.size());
503
504                 mockRaftActor.onReceiveRecover(new ApplyLogEntries(1));
505
506                 assertEquals("commit index -1", -1, mockRaftActor.getRaftActorContext().getCommitIndex());
507
508                 mockRaftActor.onReceiveRecover(new RaftActor.DeleteEntries(2));
509
510                 assertEquals("remove log entries", 0, replicatedLog.size());
511
512                 mockRaftActor.onReceiveRecover(new RaftActor.UpdateElectionTerm(10, "foobar"));
513
514                 assertNotEquals("election term", 10, mockRaftActor.getRaftActorContext().getTermInformation().getCurrentTerm());
515                 assertNotEquals("voted for", "foobar", mockRaftActor.getRaftActorContext().getTermInformation().getVotedFor());
516
517                 mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class));
518
519                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
520             }};
521     }
522
523
524     @Test
525     public void testUpdatingElectionTermCallsDataPersistence() throws Exception {
526         new JavaTestKit(getSystem()) {
527             {
528                 String persistenceId = "testUpdatingElectionTermCallsDataPersistence";
529
530                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
531
532                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
533
534                 CountDownLatch persistLatch = new CountDownLatch(1);
535                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
536                 dataPersistenceProviderMonitor.setPersistLatch(persistLatch);
537
538                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
539                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
540
541                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
542
543                 mockRaftActor.getRaftActorContext().getTermInformation().updateAndPersist(10, "foobar");
544
545                 assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS));
546
547                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
548
549             }
550         };
551     }
552
553     @Test
554     public void testAddingReplicatedLogEntryCallsDataPersistence() throws Exception {
555         new JavaTestKit(getSystem()) {
556             {
557                 String persistenceId = "testAddingReplicatedLogEntryCallsDataPersistence";
558
559                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
560
561                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
562
563                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
564
565                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
566                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
567
568                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
569
570                 MockRaftActorContext.MockReplicatedLogEntry logEntry = new MockRaftActorContext.MockReplicatedLogEntry(10, 10, mock(Payload.class));
571
572                 mockRaftActor.getRaftActorContext().getReplicatedLog().appendAndPersist(logEntry);
573
574                 verify(dataPersistenceProvider).persist(eq(logEntry), any(Procedure.class));
575
576                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
577
578             }
579         };
580     }
581
582     @Test
583     public void testRemovingReplicatedLogEntryCallsDataPersistence() throws Exception {
584         new JavaTestKit(getSystem()) {
585             {
586                 String persistenceId = "testRemovingReplicatedLogEntryCallsDataPersistence";
587
588                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
589
590                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
591
592                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
593
594                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
595                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
596
597                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
598
599                 mockRaftActor.getReplicatedLog().appendAndPersist(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class)));
600
601                 mockRaftActor.getRaftActorContext().getReplicatedLog().removeFromAndPersist(0);
602
603                 verify(dataPersistenceProvider, times(2)).persist(anyObject(), any(Procedure.class));
604
605                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
606
607             }
608         };
609     }
610
611     @Test
612     public void testApplyLogEntriesCallsDataPersistence() throws Exception {
613         new JavaTestKit(getSystem()) {
614             {
615                 String persistenceId = "testApplyLogEntriesCallsDataPersistence";
616
617                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
618
619                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
620
621                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
622
623                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
624                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
625
626                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
627
628                 mockRaftActor.onReceiveCommand(new ApplyLogEntries(10));
629
630                 verify(dataPersistenceProvider, times(1)).persist(anyObject(), any(Procedure.class));
631
632                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
633
634             }
635         };
636     }
637
638     @Test
639     public void testCaptureSnapshotReplyCallsDataPersistence() throws Exception {
640         new JavaTestKit(getSystem()) {
641             {
642                 String persistenceId = "testCaptureSnapshotReplyCallsDataPersistence";
643
644                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
645
646                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
647
648                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
649
650                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(),
651                     MockRaftActor.props(persistenceId,Collections.<String,String>emptyMap(),
652                         Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
653
654                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
655
656                 ByteString snapshotBytes  = fromObject(Arrays.asList(
657                         new MockRaftActorContext.MockPayload("A"),
658                         new MockRaftActorContext.MockPayload("B"),
659                         new MockRaftActorContext.MockPayload("C"),
660                         new MockRaftActorContext.MockPayload("D")));
661
662                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1));
663
664                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
665
666                 mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
667
668                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
669
670                 verify(dataPersistenceProvider).saveSnapshot(anyObject());
671
672                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
673
674             }
675         };
676     }
677
678     @Test
679     public void testSaveSnapshotSuccessCallsDataPersistence() throws Exception {
680         new JavaTestKit(getSystem()) {
681             {
682                 String persistenceId = "testSaveSnapshotSuccessCallsDataPersistence";
683
684                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
685
686                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
687
688                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
689
690                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
691                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
692
693                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
694
695                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,0, mock(Payload.class)));
696                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,1, mock(Payload.class)));
697                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,2, mock(Payload.class)));
698                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,3, mock(Payload.class)));
699                 mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1,4, mock(Payload.class)));
700
701                 ByteString snapshotBytes = fromObject(Arrays.asList(
702                         new MockRaftActorContext.MockPayload("A"),
703                         new MockRaftActorContext.MockPayload("B"),
704                         new MockRaftActorContext.MockPayload("C"),
705                         new MockRaftActorContext.MockPayload("D")));
706
707                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
708                 mockRaftActor.setCurrentBehavior(new Follower(raftActorContext));
709
710                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1));
711
712                 verify(mockRaftActor.delegate).createSnapshot();
713
714                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
715
716                 mockRaftActor.onReceiveCommand(new SaveSnapshotSuccess(new SnapshotMetadata("foo", 100, 100)));
717
718                 verify(dataPersistenceProvider).deleteSnapshots(any(SnapshotSelectionCriteria.class));
719
720                 verify(dataPersistenceProvider).deleteMessages(100);
721
722                 assertEquals(2, mockRaftActor.getReplicatedLog().size());
723
724                 assertNotNull(mockRaftActor.getReplicatedLog().get(3));
725                 assertNotNull(mockRaftActor.getReplicatedLog().get(4));
726
727                 // Index 2 will not be in the log because it was removed due to snapshotting
728                 assertNull(mockRaftActor.getReplicatedLog().get(2));
729
730                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
731
732             }
733         };
734     }
735
736     @Test
737     public void testApplyState() throws Exception {
738
739         new JavaTestKit(getSystem()) {
740             {
741                 String persistenceId = "testApplyState";
742
743                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
744
745                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
746
747                 DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class);
748
749                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
750                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProvider), persistenceId);
751
752                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
753
754                 ReplicatedLogEntry entry = new MockRaftActorContext.MockReplicatedLogEntry(1, 5,
755                         new MockRaftActorContext.MockPayload("F"));
756
757                 mockRaftActor.onReceiveCommand(new ApplyState(mockActorRef, "apply-state", entry));
758
759                 verify(mockRaftActor.delegate).applyState(eq(mockActorRef), eq("apply-state"), anyObject());
760
761                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
762
763             }
764         };
765     }
766
767     @Test
768     public void testApplySnapshot() throws Exception {
769         new JavaTestKit(getSystem()) {
770             {
771                 String persistenceId = "testApplySnapshot";
772
773                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
774
775                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
776
777                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
778
779                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
780                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
781
782                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
783
784                 ReplicatedLog oldReplicatedLog = mockRaftActor.getReplicatedLog();
785
786                 oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,0,mock(Payload.class)));
787                 oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1,1,mock(Payload.class)));
788                 oldReplicatedLog.append(
789                     new MockRaftActorContext.MockReplicatedLogEntry(1, 2,
790                         mock(Payload.class)));
791
792                 ByteString snapshotBytes = fromObject(Arrays.asList(
793                     new MockRaftActorContext.MockPayload("A"),
794                     new MockRaftActorContext.MockPayload("B"),
795                     new MockRaftActorContext.MockPayload("C"),
796                     new MockRaftActorContext.MockPayload("D")));
797
798                 Snapshot snapshot = mock(Snapshot.class);
799
800                 doReturn(snapshotBytes.toByteArray()).when(snapshot).getState();
801
802                 doReturn(3L).when(snapshot).getLastAppliedIndex();
803
804                 mockRaftActor.onReceiveCommand(new ApplySnapshot(snapshot));
805
806                 verify(mockRaftActor.delegate).applySnapshot(eq(snapshotBytes));
807
808                 assertTrue("The replicatedLog should have changed",
809                     oldReplicatedLog != mockRaftActor.getReplicatedLog());
810
811                 assertEquals("lastApplied should be same as in the snapshot",
812                     (Long) 3L, mockRaftActor.getLastApplied());
813
814                 assertEquals(0, mockRaftActor.getReplicatedLog().size());
815
816                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
817
818             }
819         };
820     }
821
822     @Test
823     public void testSaveSnapshotFailure() throws Exception {
824         new JavaTestKit(getSystem()) {
825             {
826                 String persistenceId = "testSaveSnapshotFailure";
827
828                 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
829
830                 config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS));
831
832                 DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor();
833
834                 TestActorRef<MockRaftActor> mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId,
835                         Collections.<String,String>emptyMap(), Optional.<ConfigParams>of(config), dataPersistenceProviderMonitor), persistenceId);
836
837                 MockRaftActor mockRaftActor = mockActorRef.underlyingActor();
838
839                 ByteString snapshotBytes  = fromObject(Arrays.asList(
840                         new MockRaftActorContext.MockPayload("A"),
841                         new MockRaftActorContext.MockPayload("B"),
842                         new MockRaftActorContext.MockPayload("C"),
843                         new MockRaftActorContext.MockPayload("D")));
844
845                 RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext();
846
847                 mockRaftActor.setCurrentBehavior(new Leader(raftActorContext));
848
849                 mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1));
850
851                 mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes));
852
853                 mockRaftActor.onReceiveCommand(new SaveSnapshotFailure(new SnapshotMetadata("foobar", 10L, 1234L),
854                         new Exception()));
855
856                 assertEquals("Snapshot index should not have advanced because save snapshot failed", -1,
857                         mockRaftActor.getReplicatedLog().getSnapshotIndex());
858
859                 mockActorRef.tell(PoisonPill.getInstance(), getRef());
860
861             }
862         };
863     }
864
865     private ByteString fromObject(Object snapshot) throws Exception {
866         ByteArrayOutputStream b = null;
867         ObjectOutputStream o = null;
868         try {
869             b = new ByteArrayOutputStream();
870             o = new ObjectOutputStream(b);
871             o.writeObject(snapshot);
872             byte[] snapshotBytes = b.toByteArray();
873             return ByteString.copyFrom(snapshotBytes);
874         } finally {
875             if (o != null) {
876                 o.flush();
877                 o.close();
878             }
879             if (b != null) {
880                 b.close();
881             }
882         }
883     }
884 }

©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.