Merge "Create transaction on the backend datastore only when neccessary"
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / ShardTest.java
1 package org.opendaylight.controller.cluster.datastore;
2
3 import static org.junit.Assert.assertEquals;
4 import static org.junit.Assert.assertFalse;
5 import static org.junit.Assert.assertNotNull;
6 import static org.junit.Assert.assertNull;
7 import static org.junit.Assert.assertTrue;
8 import static org.mockito.Mockito.doReturn;
9 import static org.mockito.Mockito.inOrder;
10 import static org.mockito.Mockito.mock;
11 import static org.mockito.Mockito.reset;
12 import static org.opendaylight.controller.cluster.datastore.DataStoreVersions.CURRENT_VERSION;
13 import akka.actor.ActorRef;
14 import akka.actor.ActorSelection;
15 import akka.actor.PoisonPill;
16 import akka.actor.Props;
17 import akka.dispatch.Dispatchers;
18 import akka.dispatch.OnComplete;
19 import akka.japi.Creator;
20 import akka.pattern.Patterns;
21 import akka.persistence.SaveSnapshotSuccess;
22 import akka.testkit.TestActorRef;
23 import akka.util.Timeout;
24 import com.google.common.base.Function;
25 import com.google.common.base.Optional;
26 import com.google.common.util.concurrent.Futures;
27 import com.google.common.util.concurrent.ListenableFuture;
28 import com.google.common.util.concurrent.Uninterruptibles;
29 import java.io.IOException;
30 import java.util.Collections;
31 import java.util.HashSet;
32 import java.util.List;
33 import java.util.Map;
34 import java.util.Set;
35 import java.util.concurrent.CountDownLatch;
36 import java.util.concurrent.TimeUnit;
37 import java.util.concurrent.atomic.AtomicBoolean;
38 import java.util.concurrent.atomic.AtomicReference;
39 import org.junit.Test;
40 import org.mockito.InOrder;
41 import org.opendaylight.controller.cluster.DataPersistenceProvider;
42 import org.opendaylight.controller.cluster.DelegatingPersistentDataProvider;
43 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
44 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
45 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
46 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
47 import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply;
48 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
49 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
50 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
51 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
52 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
53 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
54 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
55 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
56 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
57 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
58 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
59 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
60 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
61 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
62 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
63 import org.opendaylight.controller.cluster.datastore.modification.Modification;
64 import org.opendaylight.controller.cluster.datastore.modification.ModificationPayload;
65 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
66 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
67 import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
68 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
69 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
70 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
71 import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
72 import org.opendaylight.controller.cluster.raft.RaftActorContext;
73 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
74 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
75 import org.opendaylight.controller.cluster.raft.Snapshot;
76 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
77 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
78 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
79 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
80 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
81 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
82 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
83 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
84 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
85 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
86 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
87 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
88 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
89 import org.opendaylight.controller.protobuff.messages.cohort3pc.ThreePhaseCommitCohortMessages;
90 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
91 import org.opendaylight.yangtools.yang.common.QName;
92 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
93 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
94 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
95 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
96 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
97 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
98 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
99 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
100 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
101 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateTip;
102 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
103 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
104 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
105 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
106 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
107 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
108 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
109 import scala.concurrent.Await;
110 import scala.concurrent.Future;
111 import scala.concurrent.duration.FiniteDuration;
112
113 public class ShardTest extends AbstractShardTest {
114     private static final QName CARS_QNAME = QName.create("urn:opendaylight:params:xml:ns:yang:controller:md:sal:dom:store:test:cars", "2014-03-13", "cars");
115
116     @Test
117     public void testRegisterChangeListener() throws Exception {
118         new ShardTestKit(getSystem()) {{
119             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
120                     newShardProps(),  "testRegisterChangeListener");
121
122             waitUntilLeader(shard);
123
124             shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
125
126             MockDataChangeListener listener = new MockDataChangeListener(1);
127             ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener),
128                     "testRegisterChangeListener-DataChangeListener");
129
130             shard.tell(new RegisterChangeListener(TestModel.TEST_PATH,
131                     dclActor, AsyncDataBroker.DataChangeScope.BASE), getRef());
132
133             RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
134                     RegisterChangeListenerReply.class);
135             String replyPath = reply.getListenerRegistrationPath().toString();
136             assertTrue("Incorrect reply path: " + replyPath, replyPath.matches(
137                     "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
138
139             YangInstanceIdentifier path = TestModel.TEST_PATH;
140             writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
141
142             listener.waitForChangeEvents(path);
143
144             dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
145             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
146         }};
147     }
148
149     @SuppressWarnings("serial")
150     @Test
151     public void testChangeListenerNotifiedWhenNotTheLeaderOnRegistration() throws Exception {
152         // This test tests the timing window in which a change listener is registered before the
153         // shard becomes the leader. We verify that the listener is registered and notified of the
154         // existing data when the shard becomes the leader.
155         new ShardTestKit(getSystem()) {{
156             // For this test, we want to send the RegisterChangeListener message after the shard
157             // has recovered from persistence and before it becomes the leader. So we subclass
158             // Shard to override onReceiveCommand and, when the first ElectionTimeout is received,
159             // we know that the shard has been initialized to a follower and has started the
160             // election process. The following 2 CountDownLatches are used to coordinate the
161             // ElectionTimeout with the sending of the RegisterChangeListener message.
162             final CountDownLatch onFirstElectionTimeout = new CountDownLatch(1);
163             final CountDownLatch onChangeListenerRegistered = new CountDownLatch(1);
164             Creator<Shard> creator = new Creator<Shard>() {
165                 boolean firstElectionTimeout = true;
166
167                 @Override
168                 public Shard create() throws Exception {
169                     // Use a non persistent provider because this test actually invokes persist on the journal
170                     // this will cause all other messages to not be queued properly after that.
171                     // The basic issue is that you cannot use TestActorRef with a persistent actor (at least when
172                     // it does do a persist)
173                     return new Shard(shardID, Collections.<String,String>emptyMap(),
174                             dataStoreContextBuilder.persistent(false).build(), SCHEMA_CONTEXT) {
175                         @Override
176                         public void onReceiveCommand(final Object message) throws Exception {
177                             if(message instanceof ElectionTimeout && firstElectionTimeout) {
178                                 // Got the first ElectionTimeout. We don't forward it to the
179                                 // base Shard yet until we've sent the RegisterChangeListener
180                                 // message. So we signal the onFirstElectionTimeout latch to tell
181                                 // the main thread to send the RegisterChangeListener message and
182                                 // start a thread to wait on the onChangeListenerRegistered latch,
183                                 // which the main thread signals after it has sent the message.
184                                 // After the onChangeListenerRegistered is triggered, we send the
185                                 // original ElectionTimeout message to proceed with the election.
186                                 firstElectionTimeout = false;
187                                 final ActorRef self = getSelf();
188                                 new Thread() {
189                                     @Override
190                                     public void run() {
191                                         Uninterruptibles.awaitUninterruptibly(
192                                                 onChangeListenerRegistered, 5, TimeUnit.SECONDS);
193                                         self.tell(message, self);
194                                     }
195                                 }.start();
196
197                                 onFirstElectionTimeout.countDown();
198                             } else {
199                                 super.onReceiveCommand(message);
200                             }
201                         }
202                     };
203                 }
204             };
205
206             MockDataChangeListener listener = new MockDataChangeListener(1);
207             ActorRef dclActor = getSystem().actorOf(DataChangeListener.props(listener),
208                     "testRegisterChangeListenerWhenNotLeaderInitially-DataChangeListener");
209
210             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
211                     Props.create(new DelegatingShardCreator(creator)),
212                     "testRegisterChangeListenerWhenNotLeaderInitially");
213
214             // Write initial data into the in-memory store.
215             YangInstanceIdentifier path = TestModel.TEST_PATH;
216             writeToStore(shard, path, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
217
218             // Wait until the shard receives the first ElectionTimeout message.
219             assertEquals("Got first ElectionTimeout", true,
220                     onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
221
222             // Now send the RegisterChangeListener and wait for the reply.
223             shard.tell(new RegisterChangeListener(path, dclActor,
224                     AsyncDataBroker.DataChangeScope.SUBTREE), getRef());
225
226             RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
227                     RegisterChangeListenerReply.class);
228             assertNotNull("getListenerRegistrationPath", reply.getListenerRegistrationPath());
229
230             // Sanity check - verify the shard is not the leader yet.
231             shard.tell(new FindLeader(), getRef());
232             FindLeaderReply findLeadeReply =
233                     expectMsgClass(duration("5 seconds"), FindLeaderReply.class);
234             assertNull("Expected the shard not to be the leader", findLeadeReply.getLeaderActor());
235
236             // Signal the onChangeListenerRegistered latch to tell the thread above to proceed
237             // with the election process.
238             onChangeListenerRegistered.countDown();
239
240             // Wait for the shard to become the leader and notify our listener with the existing
241             // data in the store.
242             listener.waitForChangeEvents(path);
243
244             dclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
245             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
246         }};
247     }
248
249     @Test
250     public void testCreateTransaction(){
251         new ShardTestKit(getSystem()) {{
252             ActorRef shard = getSystem().actorOf(newShardProps(), "testCreateTransaction");
253
254             waitUntilLeader(shard);
255
256             shard.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
257
258             shard.tell(new CreateTransaction("txn-1",
259                     TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(), getRef());
260
261             CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
262                     CreateTransactionReply.class);
263
264             String path = reply.getTransactionActorPath().toString();
265             assertTrue("Unexpected transaction path " + path,
266                     path.contains("akka://test/user/testCreateTransaction/shard-txn-1"));
267
268             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
269         }};
270     }
271
272     @Test
273     public void testCreateTransactionOnChain(){
274         new ShardTestKit(getSystem()) {{
275             final ActorRef shard = getSystem().actorOf(newShardProps(), "testCreateTransactionOnChain");
276
277             waitUntilLeader(shard);
278
279             shard.tell(new CreateTransaction("txn-1",
280                     TransactionProxy.TransactionType.READ_ONLY.ordinal() , "foobar").toSerializable(),
281                     getRef());
282
283             CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
284                     CreateTransactionReply.class);
285
286             String path = reply.getTransactionActorPath().toString();
287             assertTrue("Unexpected transaction path " + path,
288                     path.contains("akka://test/user/testCreateTransactionOnChain/shard-txn-1"));
289
290             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
291         }};
292     }
293
294     @SuppressWarnings("serial")
295     @Test
296     public void testPeerAddressResolved() throws Exception {
297         new ShardTestKit(getSystem()) {{
298             final CountDownLatch recoveryComplete = new CountDownLatch(1);
299             class TestShard extends Shard {
300                 TestShard() {
301                     super(shardID, Collections.<String, String>singletonMap(shardID.toString(), null),
302                             newDatastoreContext(), SCHEMA_CONTEXT);
303                 }
304
305                 Map<String, String> getPeerAddresses() {
306                     return getRaftActorContext().getPeerAddresses();
307                 }
308
309                 @Override
310                 protected void onRecoveryComplete() {
311                     try {
312                         super.onRecoveryComplete();
313                     } finally {
314                         recoveryComplete.countDown();
315                     }
316                 }
317             }
318
319             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
320                     Props.create(new DelegatingShardCreator(new Creator<Shard>() {
321                         @Override
322                         public TestShard create() throws Exception {
323                             return new TestShard();
324                         }
325                     })), "testPeerAddressResolved");
326
327             //waitUntilLeader(shard);
328             assertEquals("Recovery complete", true,
329                     Uninterruptibles.awaitUninterruptibly(recoveryComplete, 5, TimeUnit.SECONDS));
330
331             String address = "akka://foobar";
332             shard.underlyingActor().onReceiveCommand(new PeerAddressResolved(shardID.toString(), address));
333
334             assertEquals("getPeerAddresses", address,
335                     ((TestShard)shard.underlyingActor()).getPeerAddresses().get(shardID.toString()));
336
337             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
338         }};
339     }
340
341     @Test
342     public void testApplySnapshot() throws Exception {
343         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(),
344                 "testApplySnapshot");
345
346         DataTree store = InMemoryDataTreeFactory.getInstance().create();
347         store.setSchemaContext(SCHEMA_CONTEXT);
348
349         writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
350
351         YangInstanceIdentifier root = YangInstanceIdentifier.builder().build();
352         NormalizedNode<?,?> expected = readStore(store, root);
353
354         ApplySnapshot applySnapshot = new ApplySnapshot(Snapshot.create(
355                 SerializationUtils.serializeNormalizedNode(expected),
356                 Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4));
357
358         shard.underlyingActor().onReceiveCommand(applySnapshot);
359
360         NormalizedNode<?,?> actual = readStore(shard, root);
361
362         assertEquals("Root node", expected, actual);
363
364         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
365     }
366
367     @Test
368     public void testApplyState() throws Exception {
369
370         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyState");
371
372         NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
373
374         ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
375                 newModificationPayload(new WriteModification(TestModel.TEST_PATH, node))));
376
377         shard.underlyingActor().onReceiveCommand(applyState);
378
379         NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH);
380         assertEquals("Applied state", node, actual);
381
382         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
383     }
384
385     @Test
386     public void testApplyStateWithCandidatePayload() throws Exception {
387
388         TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testApplyState");
389
390         NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
391         DataTreeCandidate candidate = DataTreeCandidates.fromNormalizedNode(TestModel.TEST_PATH, node);
392
393         ApplyState applyState = new ApplyState(null, "test", new ReplicatedLogImplEntry(1, 2,
394                 DataTreeCandidatePayload.create(candidate)));
395
396         shard.underlyingActor().onReceiveCommand(applyState);
397
398         NormalizedNode<?,?> actual = readStore(shard, TestModel.TEST_PATH);
399         assertEquals("Applied state", node, actual);
400
401         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
402     }
403
404     DataTree setupInMemorySnapshotStore() throws DataValidationFailedException {
405         DataTree testStore = InMemoryDataTreeFactory.getInstance().create();
406         testStore.setSchemaContext(SCHEMA_CONTEXT);
407
408         writeToStore(testStore, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
409
410         NormalizedNode<?, ?> root = readStore(testStore, YangInstanceIdentifier.builder().build());
411
412         InMemorySnapshotStore.addSnapshot(shardID.toString(), Snapshot.create(
413                 SerializationUtils.serializeNormalizedNode(root),
414                 Collections.<ReplicatedLogEntry>emptyList(), 0, 1, -1, -1));
415         return testStore;
416     }
417
418     private static DataTreeCandidatePayload payloadForModification(DataTree source, DataTreeModification mod) throws DataValidationFailedException {
419         source.validate(mod);
420         final DataTreeCandidate candidate = source.prepare(mod);
421         source.commit(candidate);
422         return DataTreeCandidatePayload.create(candidate);
423     }
424
425     @Test
426     public void testDataTreeCandidateRecovery() throws Exception {
427         // Set up the InMemorySnapshotStore.
428         final DataTree source = setupInMemorySnapshotStore();
429
430         final DataTreeModification writeMod = source.takeSnapshot().newModification();
431         writeMod.write(TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
432
433         // Set up the InMemoryJournal.
434         InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, payloadForModification(source, writeMod)));
435
436         int nListEntries = 16;
437         Set<Integer> listEntryKeys = new HashSet<>();
438
439         // Add some ModificationPayload entries
440         for (int i = 1; i <= nListEntries; i++) {
441             listEntryKeys.add(Integer.valueOf(i));
442
443             YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
444                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
445
446             final DataTreeModification mod = source.takeSnapshot().newModification();
447             mod.merge(path, ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
448
449             InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
450                 payloadForModification(source, mod)));
451         }
452
453         InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
454                 new ApplyJournalEntries(nListEntries));
455
456         testRecovery(listEntryKeys);
457     }
458
459     @Test
460     public void testModicationRecovery() throws Exception {
461
462         // Set up the InMemorySnapshotStore.
463         setupInMemorySnapshotStore();
464
465         // Set up the InMemoryJournal.
466
467         InMemoryJournal.addEntry(shardID.toString(), 0, new ReplicatedLogImplEntry(0, 1, newModificationPayload(
468                   new WriteModification(TestModel.OUTER_LIST_PATH,
469                           ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build()))));
470
471         int nListEntries = 16;
472         Set<Integer> listEntryKeys = new HashSet<>();
473
474         // Add some ModificationPayload entries
475         for(int i = 1; i <= nListEntries; i++) {
476             listEntryKeys.add(Integer.valueOf(i));
477             YangInstanceIdentifier path = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
478                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i).build();
479             Modification mod = new MergeModification(path,
480                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, i));
481             InMemoryJournal.addEntry(shardID.toString(), i, new ReplicatedLogImplEntry(i, 1,
482                     newModificationPayload(mod)));
483         }
484
485         InMemoryJournal.addEntry(shardID.toString(), nListEntries + 1,
486                 new ApplyJournalEntries(nListEntries));
487
488         testRecovery(listEntryKeys);
489     }
490
491     private static ModificationPayload newModificationPayload(final Modification... mods) throws IOException {
492         MutableCompositeModification compMod = new MutableCompositeModification();
493         for(Modification mod: mods) {
494             compMod.addModification(mod);
495         }
496
497         return new ModificationPayload(compMod);
498     }
499
500     @Test
501     public void testConcurrentThreePhaseCommits() throws Throwable {
502         new ShardTestKit(getSystem()) {{
503             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
504                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
505                     "testConcurrentThreePhaseCommits");
506
507             waitUntilLeader(shard);
508
509          // Setup 3 simulated transactions with mock cohorts backed by real cohorts.
510
511             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
512
513             String transactionID1 = "tx1";
514             MutableCompositeModification modification1 = new MutableCompositeModification();
515             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
516                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification1);
517
518             String transactionID2 = "tx2";
519             MutableCompositeModification modification2 = new MutableCompositeModification();
520             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort2", dataStore,
521                     TestModel.OUTER_LIST_PATH,
522                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
523                     modification2);
524
525             String transactionID3 = "tx3";
526             MutableCompositeModification modification3 = new MutableCompositeModification();
527             ShardDataTreeCohort cohort3 = setupMockWriteTransaction("cohort3", dataStore,
528                     YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
529                         .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
530                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1),
531                     modification3);
532
533             long timeoutSec = 5;
534             final FiniteDuration duration = FiniteDuration.create(timeoutSec, TimeUnit.SECONDS);
535             final Timeout timeout = new Timeout(duration);
536
537             // Simulate the ForwardedReadyTransaction message for the first Tx that would be sent
538             // by the ShardTransaction.
539
540             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
541                     cohort1, modification1, true, false), getRef());
542             ReadyTransactionReply readyReply = ReadyTransactionReply.fromSerializable(
543                     expectMsgClass(duration, ReadyTransactionReply.class));
544             assertEquals("Cohort path", shard.path().toString(), readyReply.getCohortPath());
545
546             // Send the CanCommitTransaction message for the first Tx.
547
548             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
549             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
550                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
551             assertEquals("Can commit", true, canCommitReply.getCanCommit());
552
553             // Send the ForwardedReadyTransaction for the next 2 Tx's.
554
555             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
556                     cohort2, modification2, true, false), getRef());
557             expectMsgClass(duration, ReadyTransactionReply.class);
558
559             shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
560                     cohort3, modification3, true, false), getRef());
561             expectMsgClass(duration, ReadyTransactionReply.class);
562
563             // Send the CanCommitTransaction message for the next 2 Tx's. These should get queued and
564             // processed after the first Tx completes.
565
566             Future<Object> canCommitFuture1 = Patterns.ask(shard,
567                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
568
569             Future<Object> canCommitFuture2 = Patterns.ask(shard,
570                     new CanCommitTransaction(transactionID3).toSerializable(), timeout);
571
572             // Send the CommitTransaction message for the first Tx. After it completes, it should
573             // trigger the 2nd Tx to proceed which should in turn then trigger the 3rd.
574
575             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
576             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
577
578             // Wait for the next 2 Tx's to complete.
579
580             final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
581             final CountDownLatch commitLatch = new CountDownLatch(2);
582
583             class OnFutureComplete extends OnComplete<Object> {
584                 private final Class<?> expRespType;
585
586                 OnFutureComplete(final Class<?> expRespType) {
587                     this.expRespType = expRespType;
588                 }
589
590                 @Override
591                 public void onComplete(final Throwable error, final Object resp) {
592                     if(error != null) {
593                         caughtEx.set(new AssertionError(getClass().getSimpleName() + " failure", error));
594                     } else {
595                         try {
596                             assertEquals("Commit response type", expRespType, resp.getClass());
597                             onSuccess(resp);
598                         } catch (Exception e) {
599                             caughtEx.set(e);
600                         }
601                     }
602                 }
603
604                 void onSuccess(final Object resp) throws Exception {
605                 }
606             }
607
608             class OnCommitFutureComplete extends OnFutureComplete {
609                 OnCommitFutureComplete() {
610                     super(CommitTransactionReply.SERIALIZABLE_CLASS);
611                 }
612
613                 @Override
614                 public void onComplete(final Throwable error, final Object resp) {
615                     super.onComplete(error, resp);
616                     commitLatch.countDown();
617                 }
618             }
619
620             class OnCanCommitFutureComplete extends OnFutureComplete {
621                 private final String transactionID;
622
623                 OnCanCommitFutureComplete(final String transactionID) {
624                     super(CanCommitTransactionReply.SERIALIZABLE_CLASS);
625                     this.transactionID = transactionID;
626                 }
627
628                 @Override
629                 void onSuccess(final Object resp) throws Exception {
630                     CanCommitTransactionReply canCommitReply =
631                             CanCommitTransactionReply.fromSerializable(resp);
632                     assertEquals("Can commit", true, canCommitReply.getCanCommit());
633
634                     Future<Object> commitFuture = Patterns.ask(shard,
635                             new CommitTransaction(transactionID).toSerializable(), timeout);
636                     commitFuture.onComplete(new OnCommitFutureComplete(), getSystem().dispatcher());
637                 }
638             }
639
640             canCommitFuture1.onComplete(new OnCanCommitFutureComplete(transactionID2),
641                     getSystem().dispatcher());
642
643             canCommitFuture2.onComplete(new OnCanCommitFutureComplete(transactionID3),
644                     getSystem().dispatcher());
645
646             boolean done = commitLatch.await(timeoutSec, TimeUnit.SECONDS);
647
648             if(caughtEx.get() != null) {
649                 throw caughtEx.get();
650             }
651
652             assertEquals("Commits complete", true, done);
653
654             InOrder inOrder = inOrder(cohort1, cohort2, cohort3);
655             inOrder.verify(cohort1).canCommit();
656             inOrder.verify(cohort1).preCommit();
657             inOrder.verify(cohort1).commit();
658             inOrder.verify(cohort2).canCommit();
659             inOrder.verify(cohort2).preCommit();
660             inOrder.verify(cohort2).commit();
661             inOrder.verify(cohort3).canCommit();
662             inOrder.verify(cohort3).preCommit();
663             inOrder.verify(cohort3).commit();
664
665             // Verify data in the data store.
666
667             verifyOuterListEntry(shard, 1);
668
669             verifyLastApplied(shard, 2);
670
671             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
672         }};
673     }
674
675     private static BatchedModifications newBatchedModifications(String transactionID, YangInstanceIdentifier path,
676             NormalizedNode<?, ?> data, boolean ready, boolean doCommitOnReady) {
677         return newBatchedModifications(transactionID, null, path, data, ready, doCommitOnReady);
678     }
679
680     private static BatchedModifications newBatchedModifications(String transactionID, String transactionChainID,
681             YangInstanceIdentifier path, NormalizedNode<?, ?> data, boolean ready, boolean doCommitOnReady) {
682         BatchedModifications batched = new BatchedModifications(transactionID, CURRENT_VERSION, transactionChainID);
683         batched.addModification(new WriteModification(path, data));
684         batched.setReady(ready);
685         batched.setDoCommitOnReady(doCommitOnReady);
686         return batched;
687     }
688
689     @Test
690     public void testBatchedModificationsWithNoCommitOnReady() throws Throwable {
691         new ShardTestKit(getSystem()) {{
692             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
693                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
694                     "testBatchedModificationsWithNoCommitOnReady");
695
696             waitUntilLeader(shard);
697
698             final String transactionID = "tx";
699             FiniteDuration duration = duration("5 seconds");
700
701             final AtomicReference<ShardDataTreeCohort> mockCohort = new AtomicReference<>();
702             ShardCommitCoordinator.CohortDecorator cohortDecorator = new ShardCommitCoordinator.CohortDecorator() {
703                 @Override
704                 public ShardDataTreeCohort decorate(String txID, ShardDataTreeCohort actual) {
705                     if(mockCohort.get() == null) {
706                         mockCohort.set(createDelegatingMockCohort("cohort", actual));
707                     }
708
709                     return mockCohort.get();
710                 }
711             };
712
713             shard.underlyingActor().getCommitCoordinator().setCohortDecorator(cohortDecorator);
714
715             // Send a BatchedModifications to start a transaction.
716
717             shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH,
718                     ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false), getRef());
719             expectMsgClass(duration, BatchedModificationsReply.class);
720
721             // Send a couple more BatchedModifications.
722
723             shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH,
724                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false), getRef());
725             expectMsgClass(duration, BatchedModificationsReply.class);
726
727             shard.tell(newBatchedModifications(transactionID, YangInstanceIdentifier.builder(
728                     TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
729                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, false), getRef());
730             expectMsgClass(duration, ReadyTransactionReply.class);
731
732             // Send the CanCommitTransaction message.
733
734             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
735             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
736                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
737             assertEquals("Can commit", true, canCommitReply.getCanCommit());
738
739             // Send the CanCommitTransaction message.
740
741             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
742             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
743
744             InOrder inOrder = inOrder(mockCohort.get());
745             inOrder.verify(mockCohort.get()).canCommit();
746             inOrder.verify(mockCohort.get()).preCommit();
747             inOrder.verify(mockCohort.get()).commit();
748
749             // Verify data in the data store.
750
751             verifyOuterListEntry(shard, 1);
752
753             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
754         }};
755     }
756
757     @Test
758     public void testBatchedModificationsWithCommitOnReady() throws Throwable {
759         new ShardTestKit(getSystem()) {{
760             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
761                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
762                     "testBatchedModificationsWithCommitOnReady");
763
764             waitUntilLeader(shard);
765
766             final String transactionID = "tx";
767             FiniteDuration duration = duration("5 seconds");
768
769             final AtomicReference<ShardDataTreeCohort> mockCohort = new AtomicReference<>();
770             ShardCommitCoordinator.CohortDecorator cohortDecorator = new ShardCommitCoordinator.CohortDecorator() {
771                 @Override
772                 public ShardDataTreeCohort decorate(String txID, ShardDataTreeCohort actual) {
773                     if(mockCohort.get() == null) {
774                         mockCohort.set(createDelegatingMockCohort("cohort", actual));
775                     }
776
777                     return mockCohort.get();
778                 }
779             };
780
781             shard.underlyingActor().getCommitCoordinator().setCohortDecorator(cohortDecorator);
782
783             // Send a BatchedModifications to start a transaction.
784
785             shard.tell(newBatchedModifications(transactionID, TestModel.TEST_PATH,
786                     ImmutableNodes.containerNode(TestModel.TEST_QNAME), false, false), getRef());
787             expectMsgClass(duration, BatchedModificationsReply.class);
788
789             // Send a couple more BatchedModifications.
790
791             shard.tell(newBatchedModifications(transactionID, TestModel.OUTER_LIST_PATH,
792                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(), false, false), getRef());
793             expectMsgClass(duration, BatchedModificationsReply.class);
794
795             shard.tell(newBatchedModifications(transactionID, YangInstanceIdentifier.builder(
796                     TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
797                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1), true, true), getRef());
798
799             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
800
801             InOrder inOrder = inOrder(mockCohort.get());
802             inOrder.verify(mockCohort.get()).canCommit();
803             inOrder.verify(mockCohort.get()).preCommit();
804             inOrder.verify(mockCohort.get()).commit();
805
806             // Verify data in the data store.
807
808             verifyOuterListEntry(shard, 1);
809
810             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
811         }};
812     }
813
814     @SuppressWarnings("unchecked")
815     private static void verifyOuterListEntry(final TestActorRef<Shard> shard, Object expIDValue) throws Exception {
816         NormalizedNode<?, ?> outerList = readStore(shard, TestModel.OUTER_LIST_PATH);
817         assertNotNull(TestModel.OUTER_LIST_QNAME.getLocalName() + " not found", outerList);
818         assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " value is not Iterable",
819                 outerList.getValue() instanceof Iterable);
820         Object entry = ((Iterable<Object>)outerList.getValue()).iterator().next();
821         assertTrue(TestModel.OUTER_LIST_QNAME.getLocalName() + " entry is not MapEntryNode",
822                 entry instanceof MapEntryNode);
823         MapEntryNode mapEntry = (MapEntryNode)entry;
824         Optional<DataContainerChild<? extends PathArgument, ?>> idLeaf =
825                 mapEntry.getChild(new YangInstanceIdentifier.NodeIdentifier(TestModel.ID_QNAME));
826         assertTrue("Missing leaf " + TestModel.ID_QNAME.getLocalName(), idLeaf.isPresent());
827         assertEquals(TestModel.ID_QNAME.getLocalName() + " value", expIDValue, idLeaf.get().getValue());
828     }
829
830     @Test
831     public void testBatchedModificationsOnTransactionChain() throws Throwable {
832         new ShardTestKit(getSystem()) {{
833             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
834                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
835                     "testBatchedModificationsOnTransactionChain");
836
837             waitUntilLeader(shard);
838
839             String transactionChainID = "txChain";
840             String transactionID1 = "tx1";
841             String transactionID2 = "tx2";
842
843             FiniteDuration duration = duration("5 seconds");
844
845             // Send a BatchedModifications to start a chained write transaction and ready it.
846
847             ContainerNode containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
848             YangInstanceIdentifier path = TestModel.TEST_PATH;
849             shard.tell(newBatchedModifications(transactionID1, transactionChainID, path,
850                     containerNode, true, false), getRef());
851             expectMsgClass(duration, ReadyTransactionReply.class);
852
853             // Create a read Tx on the same chain.
854
855             shard.tell(new CreateTransaction(transactionID2, TransactionProxy.TransactionType.READ_ONLY.ordinal() ,
856                     transactionChainID).toSerializable(), getRef());
857
858             CreateTransactionReply createReply = expectMsgClass(duration("3 seconds"), CreateTransactionReply.class);
859
860             getSystem().actorSelection(createReply.getTransactionActorPath()).tell(new ReadData(path), getRef());
861             ReadDataReply readReply = expectMsgClass(duration("3 seconds"), ReadDataReply.class);
862             assertEquals("Read node", containerNode, readReply.getNormalizedNode());
863
864             // Commit the write transaction.
865
866             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
867             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
868                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
869             assertEquals("Can commit", true, canCommitReply.getCanCommit());
870
871             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
872             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
873
874             // Verify data in the data store.
875
876             NormalizedNode<?, ?> actualNode = readStore(shard, path);
877             assertEquals("Stored node", containerNode, actualNode);
878
879             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
880         }};
881     }
882
883     @Test
884     public void testOnBatchedModificationsWhenNotLeader() {
885         final AtomicBoolean overrideLeaderCalls = new AtomicBoolean();
886         new ShardTestKit(getSystem()) {{
887             Creator<Shard> creator = new Creator<Shard>() {
888                 private static final long serialVersionUID = 1L;
889
890                 @Override
891                 public Shard create() throws Exception {
892                     return new Shard(shardID, Collections.<String,String>emptyMap(),
893                             newDatastoreContext(), SCHEMA_CONTEXT) {
894                         @Override
895                         protected boolean isLeader() {
896                             return overrideLeaderCalls.get() ? false : super.isLeader();
897                         }
898
899                         @Override
900                         protected ActorSelection getLeader() {
901                             return overrideLeaderCalls.get() ? getSystem().actorSelection(getRef().path()) :
902                                 super.getLeader();
903                         }
904                     };
905                 }
906             };
907
908             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
909                     Props.create(new DelegatingShardCreator(creator)), "testOnBatchedModificationsWhenNotLeader");
910
911             waitUntilLeader(shard);
912
913             overrideLeaderCalls.set(true);
914
915             BatchedModifications batched = new BatchedModifications("tx", DataStoreVersions.CURRENT_VERSION, "");
916
917             shard.tell(batched, ActorRef.noSender());
918
919             expectMsgEquals(batched);
920
921             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
922         }};
923     }
924
925     @Test
926     public void testForwardedReadyTransactionWithImmediateCommit() throws Exception{
927         new ShardTestKit(getSystem()) {{
928             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
929                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
930                     "testForwardedReadyTransactionWithImmediateCommit");
931
932             waitUntilLeader(shard);
933
934             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
935
936             String transactionID = "tx1";
937             MutableCompositeModification modification = new MutableCompositeModification();
938             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
939             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort", dataStore,
940                     TestModel.TEST_PATH, containerNode, modification);
941
942             FiniteDuration duration = duration("5 seconds");
943
944             // Simulate the ForwardedReadyTransaction messages that would be sent
945             // by the ShardTransaction.
946
947             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
948                     cohort, modification, true, true), getRef());
949
950             expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
951
952             InOrder inOrder = inOrder(cohort);
953             inOrder.verify(cohort).canCommit();
954             inOrder.verify(cohort).preCommit();
955             inOrder.verify(cohort).commit();
956
957             NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.TEST_PATH);
958             assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode);
959
960             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
961         }};
962     }
963
964     @Test
965     public void testCommitWithPersistenceDisabled() throws Throwable {
966         dataStoreContextBuilder.persistent(false);
967         new ShardTestKit(getSystem()) {{
968             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
969                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
970                     "testCommitWithPersistenceDisabled");
971
972             waitUntilLeader(shard);
973
974             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
975
976             // Setup a simulated transactions with a mock cohort.
977
978             String transactionID = "tx";
979             MutableCompositeModification modification = new MutableCompositeModification();
980             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
981             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort", dataStore,
982                     TestModel.TEST_PATH, containerNode, modification);
983
984             FiniteDuration duration = duration("5 seconds");
985
986             // Simulate the ForwardedReadyTransaction messages that would be sent
987             // by the ShardTransaction.
988
989             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
990                     cohort, modification, true, false), getRef());
991             expectMsgClass(duration, ReadyTransactionReply.class);
992
993             // Send the CanCommitTransaction message.
994
995             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
996             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
997                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
998             assertEquals("Can commit", true, canCommitReply.getCanCommit());
999
1000             // Send the CanCommitTransaction message.
1001
1002             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1003             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1004
1005             InOrder inOrder = inOrder(cohort);
1006             inOrder.verify(cohort).canCommit();
1007             inOrder.verify(cohort).preCommit();
1008             inOrder.verify(cohort).commit();
1009
1010             NormalizedNode<?, ?> actualNode = readStore(shard, TestModel.TEST_PATH);
1011             assertEquals(TestModel.TEST_QNAME.getLocalName(), containerNode, actualNode);
1012
1013             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1014         }};
1015     }
1016
1017     private static DataTreeCandidateTip mockCandidate(final String name) {
1018         DataTreeCandidateTip mockCandidate = mock(DataTreeCandidateTip.class, name);
1019         DataTreeCandidateNode mockCandidateNode = mock(DataTreeCandidateNode.class, name + "-node");
1020         doReturn(ModificationType.WRITE).when(mockCandidateNode).getModificationType();
1021         doReturn(Optional.of(ImmutableNodes.containerNode(CARS_QNAME))).when(mockCandidateNode).getDataAfter();
1022         doReturn(YangInstanceIdentifier.builder().build()).when(mockCandidate).getRootPath();
1023         doReturn(mockCandidateNode).when(mockCandidate).getRootNode();
1024         return mockCandidate;
1025     }
1026
1027     private static DataTreeCandidateTip mockUnmodifiedCandidate(final String name) {
1028         DataTreeCandidateTip mockCandidate = mock(DataTreeCandidateTip.class, name);
1029         DataTreeCandidateNode mockCandidateNode = mock(DataTreeCandidateNode.class, name + "-node");
1030         doReturn(ModificationType.UNMODIFIED).when(mockCandidateNode).getModificationType();
1031         doReturn(YangInstanceIdentifier.builder().build()).when(mockCandidate).getRootPath();
1032         doReturn(mockCandidateNode).when(mockCandidate).getRootNode();
1033         return mockCandidate;
1034     }
1035
1036     @Test
1037     public void testCommitWhenTransactionHasNoModifications(){
1038         // Note that persistence is enabled which would normally result in the entry getting written to the journal
1039         // but here that need not happen
1040         new ShardTestKit(getSystem()) {
1041             {
1042                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1043                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1044                         "testCommitWhenTransactionHasNoModifications");
1045
1046                 waitUntilLeader(shard);
1047
1048                 String transactionID = "tx1";
1049                 MutableCompositeModification modification = new MutableCompositeModification();
1050                 ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1051                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1052                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).preCommit();
1053                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).commit();
1054                 doReturn(mockUnmodifiedCandidate("cohort1-candidate")).when(cohort).getCandidate();
1055
1056                 FiniteDuration duration = duration("5 seconds");
1057
1058                 // Simulate the ForwardedReadyTransaction messages that would be sent
1059                 // by the ShardTransaction.
1060
1061                 shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1062                         cohort, modification, true, false), getRef());
1063                 expectMsgClass(duration, ReadyTransactionReply.class);
1064
1065                 // Send the CanCommitTransaction message.
1066
1067                 shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1068                 CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1069                         expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1070                 assertEquals("Can commit", true, canCommitReply.getCanCommit());
1071
1072                 shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1073                 expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
1074
1075                 InOrder inOrder = inOrder(cohort);
1076                 inOrder.verify(cohort).canCommit();
1077                 inOrder.verify(cohort).preCommit();
1078                 inOrder.verify(cohort).commit();
1079
1080                 // Use MBean for verification
1081                 // Committed transaction count should increase as usual
1082                 assertEquals(1,shard.underlyingActor().getShardMBean().getCommittedTransactionsCount());
1083
1084                 // Commit index should not advance because this does not go into the journal
1085                 assertEquals(-1, shard.underlyingActor().getShardMBean().getCommitIndex());
1086
1087                 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1088
1089             }
1090         };
1091     }
1092
1093     @Test
1094     public void testCommitWhenTransactionHasModifications(){
1095         new ShardTestKit(getSystem()) {
1096             {
1097                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1098                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1099                         "testCommitWhenTransactionHasModifications");
1100
1101                 waitUntilLeader(shard);
1102
1103                 String transactionID = "tx1";
1104                 MutableCompositeModification modification = new MutableCompositeModification();
1105                 modification.addModification(new DeleteModification(YangInstanceIdentifier.builder().build()));
1106                 ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1107                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1108                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).preCommit();
1109                 doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).commit();
1110                 doReturn(mockCandidate("cohort1-candidate")).when(cohort).getCandidate();
1111
1112                 FiniteDuration duration = duration("5 seconds");
1113
1114                 // Simulate the ForwardedReadyTransaction messages that would be sent
1115                 // by the ShardTransaction.
1116
1117                 shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1118                         cohort, modification, true, false), getRef());
1119                 expectMsgClass(duration, ReadyTransactionReply.class);
1120
1121                 // Send the CanCommitTransaction message.
1122
1123                 shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1124                 CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1125                         expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1126                 assertEquals("Can commit", true, canCommitReply.getCanCommit());
1127
1128                 shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1129                 expectMsgClass(duration, ThreePhaseCommitCohortMessages.CommitTransactionReply.class);
1130
1131                 InOrder inOrder = inOrder(cohort);
1132                 inOrder.verify(cohort).canCommit();
1133                 inOrder.verify(cohort).preCommit();
1134                 inOrder.verify(cohort).commit();
1135
1136                 // Use MBean for verification
1137                 // Committed transaction count should increase as usual
1138                 assertEquals(1, shard.underlyingActor().getShardMBean().getCommittedTransactionsCount());
1139
1140                 // Commit index should advance as we do not have an empty modification
1141                 assertEquals(0, shard.underlyingActor().getShardMBean().getCommitIndex());
1142
1143                 shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1144
1145             }
1146         };
1147     }
1148
1149     @Test
1150     public void testCommitPhaseFailure() throws Throwable {
1151         new ShardTestKit(getSystem()) {{
1152             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1153                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1154                     "testCommitPhaseFailure");
1155
1156             waitUntilLeader(shard);
1157
1158             // Setup 2 simulated transactions with mock cohorts. The first one fails in the
1159             // commit phase.
1160
1161             String transactionID1 = "tx1";
1162             MutableCompositeModification modification1 = new MutableCompositeModification();
1163             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1164             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1165             doReturn(Futures.immediateFuture(null)).when(cohort1).preCommit();
1166             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort1).commit();
1167             doReturn(mockCandidate("cohort1-candidate")).when(cohort1).getCandidate();
1168
1169             String transactionID2 = "tx2";
1170             MutableCompositeModification modification2 = new MutableCompositeModification();
1171             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1172             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1173
1174             FiniteDuration duration = duration("5 seconds");
1175             final Timeout timeout = new Timeout(duration);
1176
1177             // Simulate the ForwardedReadyTransaction messages that would be sent
1178             // by the ShardTransaction.
1179
1180             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1181                     cohort1, modification1, true, false), getRef());
1182             expectMsgClass(duration, ReadyTransactionReply.class);
1183
1184             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1185                     cohort2, modification2, true, false), getRef());
1186             expectMsgClass(duration, ReadyTransactionReply.class);
1187
1188             // Send the CanCommitTransaction message for the first Tx.
1189
1190             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1191             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1192                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1193             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1194
1195             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1196             // processed after the first Tx completes.
1197
1198             Future<Object> canCommitFuture = Patterns.ask(shard,
1199                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1200
1201             // Send the CommitTransaction message for the first Tx. This should send back an error
1202             // and trigger the 2nd Tx to proceed.
1203
1204             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1205             expectMsgClass(duration, akka.actor.Status.Failure.class);
1206
1207             // Wait for the 2nd Tx to complete the canCommit phase.
1208
1209             final CountDownLatch latch = new CountDownLatch(1);
1210             canCommitFuture.onComplete(new OnComplete<Object>() {
1211                 @Override
1212                 public void onComplete(final Throwable t, final Object resp) {
1213                     latch.countDown();
1214                 }
1215             }, getSystem().dispatcher());
1216
1217             assertEquals("2nd CanCommit complete", true, latch.await(5, TimeUnit.SECONDS));
1218
1219             InOrder inOrder = inOrder(cohort1, cohort2);
1220             inOrder.verify(cohort1).canCommit();
1221             inOrder.verify(cohort1).preCommit();
1222             inOrder.verify(cohort1).commit();
1223             inOrder.verify(cohort2).canCommit();
1224
1225             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1226         }};
1227     }
1228
1229     @Test
1230     public void testPreCommitPhaseFailure() throws Throwable {
1231         new ShardTestKit(getSystem()) {{
1232             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1233                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1234                     "testPreCommitPhaseFailure");
1235
1236             waitUntilLeader(shard);
1237
1238             String transactionID1 = "tx1";
1239             MutableCompositeModification modification1 = new MutableCompositeModification();
1240             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1241             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1242             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort1).preCommit();
1243
1244             String transactionID2 = "tx2";
1245             MutableCompositeModification modification2 = new MutableCompositeModification();
1246             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1247             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1248
1249             FiniteDuration duration = duration("5 seconds");
1250             final Timeout timeout = new Timeout(duration);
1251
1252             // Simulate the ForwardedReadyTransaction messages that would be sent
1253             // by the ShardTransaction.
1254
1255             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1256                     cohort1, modification1, true, false), getRef());
1257             expectMsgClass(duration, ReadyTransactionReply.class);
1258
1259             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1260                     cohort2, modification2, true, false), getRef());
1261             expectMsgClass(duration, ReadyTransactionReply.class);
1262
1263             // Send the CanCommitTransaction message for the first Tx.
1264
1265             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1266             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1267                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1268             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1269
1270             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1271             // processed after the first Tx completes.
1272
1273             Future<Object> canCommitFuture = Patterns.ask(shard,
1274                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1275
1276             // Send the CommitTransaction message for the first Tx. This should send back an error
1277             // and trigger the 2nd Tx to proceed.
1278
1279             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1280             expectMsgClass(duration, akka.actor.Status.Failure.class);
1281
1282             // Wait for the 2nd Tx to complete the canCommit phase.
1283
1284             final CountDownLatch latch = new CountDownLatch(1);
1285             canCommitFuture.onComplete(new OnComplete<Object>() {
1286                 @Override
1287                 public void onComplete(final Throwable t, final Object resp) {
1288                     latch.countDown();
1289                 }
1290             }, getSystem().dispatcher());
1291
1292             assertEquals("2nd CanCommit complete", true, latch.await(5, TimeUnit.SECONDS));
1293
1294             InOrder inOrder = inOrder(cohort1, cohort2);
1295             inOrder.verify(cohort1).canCommit();
1296             inOrder.verify(cohort1).preCommit();
1297             inOrder.verify(cohort2).canCommit();
1298
1299             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1300         }};
1301     }
1302
1303     @Test
1304     public void testCanCommitPhaseFailure() throws Throwable {
1305         new ShardTestKit(getSystem()) {{
1306             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1307                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1308                     "testCanCommitPhaseFailure");
1309
1310             waitUntilLeader(shard);
1311
1312             final FiniteDuration duration = duration("5 seconds");
1313
1314             String transactionID1 = "tx1";
1315             MutableCompositeModification modification = new MutableCompositeModification();
1316             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1317             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort).canCommit();
1318
1319             // Simulate the ForwardedReadyTransaction messages that would be sent
1320             // by the ShardTransaction.
1321
1322             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1323                     cohort, modification, true, false), getRef());
1324             expectMsgClass(duration, ReadyTransactionReply.class);
1325
1326             // Send the CanCommitTransaction message.
1327
1328             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1329             expectMsgClass(duration, akka.actor.Status.Failure.class);
1330
1331             // Send another can commit to ensure the failed one got cleaned up.
1332
1333             reset(cohort);
1334
1335             String transactionID2 = "tx2";
1336             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1337
1338             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1339                     cohort, modification, true, false), getRef());
1340             expectMsgClass(duration, ReadyTransactionReply.class);
1341
1342             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1343             CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(
1344                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1345             assertEquals("getCanCommit", true, reply.getCanCommit());
1346
1347             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1348         }};
1349     }
1350
1351     @Test
1352     public void testCanCommitPhaseFalseResponse() throws Throwable {
1353         new ShardTestKit(getSystem()) {{
1354             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1355                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1356                     "testCanCommitPhaseFalseResponse");
1357
1358             waitUntilLeader(shard);
1359
1360             final FiniteDuration duration = duration("5 seconds");
1361
1362             String transactionID1 = "tx1";
1363             MutableCompositeModification modification = new MutableCompositeModification();
1364             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1365             doReturn(Futures.immediateFuture(Boolean.FALSE)).when(cohort).canCommit();
1366
1367             // Simulate the ForwardedReadyTransaction messages that would be sent
1368             // by the ShardTransaction.
1369
1370             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1371                     cohort, modification, true, false), getRef());
1372             expectMsgClass(duration, ReadyTransactionReply.class);
1373
1374             // Send the CanCommitTransaction message.
1375
1376             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1377             CanCommitTransactionReply reply = CanCommitTransactionReply.fromSerializable(
1378                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1379             assertEquals("getCanCommit", false, reply.getCanCommit());
1380
1381             // Send another can commit to ensure the failed one got cleaned up.
1382
1383             reset(cohort);
1384
1385             String transactionID2 = "tx2";
1386             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1387
1388             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1389                     cohort, modification, true, false), getRef());
1390             expectMsgClass(duration, ReadyTransactionReply.class);
1391
1392             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1393             reply = CanCommitTransactionReply.fromSerializable(
1394                     expectMsgClass(CanCommitTransactionReply.SERIALIZABLE_CLASS));
1395             assertEquals("getCanCommit", true, reply.getCanCommit());
1396
1397             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1398         }};
1399     }
1400
1401     @Test
1402     public void testImmediateCommitWithCanCommitPhaseFailure() throws Throwable {
1403         new ShardTestKit(getSystem()) {{
1404             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1405                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1406                     "testImmediateCommitWithCanCommitPhaseFailure");
1407
1408             waitUntilLeader(shard);
1409
1410             final FiniteDuration duration = duration("5 seconds");
1411
1412             String transactionID1 = "tx1";
1413             MutableCompositeModification modification = new MutableCompositeModification();
1414             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1415             doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock"))).when(cohort).canCommit();
1416
1417             // Simulate the ForwardedReadyTransaction messages that would be sent
1418             // by the ShardTransaction.
1419
1420             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1421                     cohort, modification, true, true), getRef());
1422
1423             expectMsgClass(duration, akka.actor.Status.Failure.class);
1424
1425             // Send another can commit to ensure the failed one got cleaned up.
1426
1427             reset(cohort);
1428
1429             String transactionID2 = "tx2";
1430             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1431             doReturn(Futures.immediateFuture(null)).when(cohort).preCommit();
1432             doReturn(Futures.immediateFuture(null)).when(cohort).commit();
1433             DataTreeCandidateTip candidate = mock(DataTreeCandidateTip.class);
1434             DataTreeCandidateNode candidateRoot = mock(DataTreeCandidateNode.class);
1435             doReturn(ModificationType.UNMODIFIED).when(candidateRoot).getModificationType();
1436             doReturn(candidateRoot).when(candidate).getRootNode();
1437             doReturn(candidate).when(cohort).getCandidate();
1438
1439             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1440                     cohort, modification, true, true), getRef());
1441
1442             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1443
1444             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1445         }};
1446     }
1447
1448     @Test
1449     public void testImmediateCommitWithCanCommitPhaseFalseResponse() throws Throwable {
1450         new ShardTestKit(getSystem()) {{
1451             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1452                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1453                     "testImmediateCommitWithCanCommitPhaseFalseResponse");
1454
1455             waitUntilLeader(shard);
1456
1457             final FiniteDuration duration = duration("5 seconds");
1458
1459             String transactionID = "tx1";
1460             MutableCompositeModification modification = new MutableCompositeModification();
1461             ShardDataTreeCohort cohort = mock(ShardDataTreeCohort.class, "cohort1");
1462             doReturn(Futures.immediateFuture(Boolean.FALSE)).when(cohort).canCommit();
1463
1464             // Simulate the ForwardedReadyTransaction messages that would be sent
1465             // by the ShardTransaction.
1466
1467             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1468                     cohort, modification, true, true), getRef());
1469
1470             expectMsgClass(duration, akka.actor.Status.Failure.class);
1471
1472             // Send another can commit to ensure the failed one got cleaned up.
1473
1474             reset(cohort);
1475
1476             String transactionID2 = "tx2";
1477             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort).canCommit();
1478             doReturn(Futures.immediateFuture(null)).when(cohort).preCommit();
1479             doReturn(Futures.immediateFuture(null)).when(cohort).commit();
1480             DataTreeCandidateTip candidate = mock(DataTreeCandidateTip.class);
1481             DataTreeCandidateNode candidateRoot = mock(DataTreeCandidateNode.class);
1482             doReturn(ModificationType.UNMODIFIED).when(candidateRoot).getModificationType();
1483             doReturn(candidateRoot).when(candidate).getRootNode();
1484             doReturn(candidate).when(cohort).getCandidate();
1485
1486             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1487                     cohort, modification, true, true), getRef());
1488
1489             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1490
1491             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1492         }};
1493     }
1494
1495     @Test
1496     public void testAbortBeforeFinishCommit() throws Throwable {
1497         new ShardTestKit(getSystem()) {{
1498             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1499                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1500                     "testAbortBeforeFinishCommit");
1501
1502             waitUntilLeader(shard);
1503
1504             final FiniteDuration duration = duration("5 seconds");
1505             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1506
1507             final String transactionID = "tx1";
1508             Function<ShardDataTreeCohort, ListenableFuture<Void>> preCommit =
1509                           new Function<ShardDataTreeCohort, ListenableFuture<Void>>() {
1510                 @Override
1511                 public ListenableFuture<Void> apply(final ShardDataTreeCohort cohort) {
1512                     ListenableFuture<Void> preCommitFuture = cohort.preCommit();
1513
1514                     // Simulate an AbortTransaction message occurring during replication, after
1515                     // persisting and before finishing the commit to the in-memory store.
1516                     // We have no followers so due to optimizations in the RaftActor, it does not
1517                     // attempt replication and thus we can't send an AbortTransaction message b/c
1518                     // it would be processed too late after CommitTransaction completes. So we'll
1519                     // simulate an AbortTransaction message occurring during replication by calling
1520                     // the shard directly.
1521                     //
1522                     shard.underlyingActor().doAbortTransaction(transactionID, null);
1523
1524                     return preCommitFuture;
1525                 }
1526             };
1527
1528             MutableCompositeModification modification = new MutableCompositeModification();
1529             ShardDataTreeCohort cohort = setupMockWriteTransaction("cohort1", dataStore,
1530                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME),
1531                     modification, preCommit);
1532
1533             shard.tell(new ForwardedReadyTransaction(transactionID, CURRENT_VERSION,
1534                     cohort, modification, true, false), getRef());
1535             expectMsgClass(duration, ReadyTransactionReply.class);
1536
1537             shard.tell(new CanCommitTransaction(transactionID).toSerializable(), getRef());
1538             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1539                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1540             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1541
1542             shard.tell(new CommitTransaction(transactionID).toSerializable(), getRef());
1543             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1544
1545             NormalizedNode<?, ?> node = readStore(shard, TestModel.TEST_PATH);
1546
1547             // Since we're simulating an abort occurring during replication and before finish commit,
1548             // the data should still get written to the in-memory store since we've gotten past
1549             // canCommit and preCommit and persisted the data.
1550             assertNotNull(TestModel.TEST_QNAME.getLocalName() + " not found", node);
1551
1552             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1553         }};
1554     }
1555
1556     @Test
1557     public void testTransactionCommitTimeout() throws Throwable {
1558         dataStoreContextBuilder.shardTransactionCommitTimeoutInSeconds(1);
1559
1560         new ShardTestKit(getSystem()) {{
1561             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1562                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1563                     "testTransactionCommitTimeout");
1564
1565             waitUntilLeader(shard);
1566
1567             final FiniteDuration duration = duration("5 seconds");
1568
1569             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1570
1571             writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1572             writeToStore(shard, TestModel.OUTER_LIST_PATH,
1573                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1574
1575             // Create 1st Tx - will timeout
1576
1577             String transactionID1 = "tx1";
1578             MutableCompositeModification modification1 = new MutableCompositeModification();
1579             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
1580                     YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1581                         .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build(),
1582                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1),
1583                     modification1);
1584
1585             // Create 2nd Tx
1586
1587             String transactionID2 = "tx3";
1588             MutableCompositeModification modification2 = new MutableCompositeModification();
1589             YangInstanceIdentifier listNodePath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1590                 .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build();
1591             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort3", dataStore,
1592                     listNodePath,
1593                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2),
1594                     modification2);
1595
1596             // Ready the Tx's
1597
1598             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1599                     cohort1, modification1, true, false), getRef());
1600             expectMsgClass(duration, ReadyTransactionReply.class);
1601
1602             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1603                     cohort2, modification2, true, false), getRef());
1604             expectMsgClass(duration, ReadyTransactionReply.class);
1605
1606             // canCommit 1st Tx. We don't send the commit so it should timeout.
1607
1608             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1609             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1610
1611             // canCommit the 2nd Tx - it should complete after the 1st Tx times out.
1612
1613             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1614             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1615
1616             // Try to commit the 1st Tx - should fail as it's not the current Tx.
1617
1618             shard.tell(new CommitTransaction(transactionID1).toSerializable(), getRef());
1619             expectMsgClass(duration, akka.actor.Status.Failure.class);
1620
1621             // Commit the 2nd Tx.
1622
1623             shard.tell(new CommitTransaction(transactionID2).toSerializable(), getRef());
1624             expectMsgClass(duration, CommitTransactionReply.SERIALIZABLE_CLASS);
1625
1626             NormalizedNode<?, ?> node = readStore(shard, listNodePath);
1627             assertNotNull(listNodePath + " not found", node);
1628
1629             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1630         }};
1631     }
1632
1633     @Test
1634     public void testTransactionCommitQueueCapacityExceeded() throws Throwable {
1635         dataStoreContextBuilder.shardTransactionCommitQueueCapacity(1);
1636
1637         new ShardTestKit(getSystem()) {{
1638             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1639                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1640                     "testTransactionCommitQueueCapacityExceeded");
1641
1642             waitUntilLeader(shard);
1643
1644             final FiniteDuration duration = duration("5 seconds");
1645
1646             ShardDataTree dataStore = shard.underlyingActor().getDataStore();
1647
1648             String transactionID1 = "tx1";
1649             MutableCompositeModification modification1 = new MutableCompositeModification();
1650             ShardDataTreeCohort cohort1 = setupMockWriteTransaction("cohort1", dataStore,
1651                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification1);
1652
1653             String transactionID2 = "tx2";
1654             MutableCompositeModification modification2 = new MutableCompositeModification();
1655             ShardDataTreeCohort cohort2 = setupMockWriteTransaction("cohort2", dataStore,
1656                     TestModel.OUTER_LIST_PATH,
1657                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build(),
1658                     modification2);
1659
1660             String transactionID3 = "tx3";
1661             MutableCompositeModification modification3 = new MutableCompositeModification();
1662             ShardDataTreeCohort cohort3 = setupMockWriteTransaction("cohort3", dataStore,
1663                     TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME), modification3);
1664
1665             // Ready the Tx's
1666
1667             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1668                     cohort1, modification1, true, false), getRef());
1669             expectMsgClass(duration, ReadyTransactionReply.class);
1670
1671             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1672                     cohort2, modification2, true, false), getRef());
1673             expectMsgClass(duration, ReadyTransactionReply.class);
1674
1675             shard.tell(new ForwardedReadyTransaction(transactionID3, CURRENT_VERSION,
1676                     cohort3, modification3, true, false), getRef());
1677             expectMsgClass(duration, ReadyTransactionReply.class);
1678
1679             // canCommit 1st Tx.
1680
1681             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1682             expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS);
1683
1684             // canCommit the 2nd Tx - it should get queued.
1685
1686             shard.tell(new CanCommitTransaction(transactionID2).toSerializable(), getRef());
1687
1688             // canCommit the 3rd Tx - should exceed queue capacity and fail.
1689
1690             shard.tell(new CanCommitTransaction(transactionID3).toSerializable(), getRef());
1691             expectMsgClass(duration, akka.actor.Status.Failure.class);
1692
1693             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1694         }};
1695     }
1696
1697     @Test
1698     public void testCanCommitBeforeReadyFailure() throws Throwable {
1699         new ShardTestKit(getSystem()) {{
1700             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1701                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1702                     "testCanCommitBeforeReadyFailure");
1703
1704             shard.tell(new CanCommitTransaction("tx").toSerializable(), getRef());
1705             expectMsgClass(duration("5 seconds"), akka.actor.Status.Failure.class);
1706
1707             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1708         }};
1709     }
1710
1711     @Test
1712     public void testAbortTransaction() throws Throwable {
1713         new ShardTestKit(getSystem()) {{
1714             final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1715                     newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1716                     "testAbortTransaction");
1717
1718             waitUntilLeader(shard);
1719
1720             // Setup 2 simulated transactions with mock cohorts. The first one will be aborted.
1721
1722             String transactionID1 = "tx1";
1723             MutableCompositeModification modification1 = new MutableCompositeModification();
1724             ShardDataTreeCohort cohort1 = mock(ShardDataTreeCohort.class, "cohort1");
1725             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort1).canCommit();
1726             doReturn(Futures.immediateFuture(null)).when(cohort1).abort();
1727
1728             String transactionID2 = "tx2";
1729             MutableCompositeModification modification2 = new MutableCompositeModification();
1730             ShardDataTreeCohort cohort2 = mock(ShardDataTreeCohort.class, "cohort2");
1731             doReturn(Futures.immediateFuture(Boolean.TRUE)).when(cohort2).canCommit();
1732
1733             FiniteDuration duration = duration("5 seconds");
1734             final Timeout timeout = new Timeout(duration);
1735
1736             // Simulate the ForwardedReadyTransaction messages that would be sent
1737             // by the ShardTransaction.
1738
1739             shard.tell(new ForwardedReadyTransaction(transactionID1, CURRENT_VERSION,
1740                     cohort1, modification1, true, false), getRef());
1741             expectMsgClass(duration, ReadyTransactionReply.class);
1742
1743             shard.tell(new ForwardedReadyTransaction(transactionID2, CURRENT_VERSION,
1744                     cohort2, modification2, true, false), getRef());
1745             expectMsgClass(duration, ReadyTransactionReply.class);
1746
1747             // Send the CanCommitTransaction message for the first Tx.
1748
1749             shard.tell(new CanCommitTransaction(transactionID1).toSerializable(), getRef());
1750             CanCommitTransactionReply canCommitReply = CanCommitTransactionReply.fromSerializable(
1751                     expectMsgClass(duration, CanCommitTransactionReply.SERIALIZABLE_CLASS));
1752             assertEquals("Can commit", true, canCommitReply.getCanCommit());
1753
1754             // Send the CanCommitTransaction message for the 2nd Tx. This should get queued and
1755             // processed after the first Tx completes.
1756
1757             Future<Object> canCommitFuture = Patterns.ask(shard,
1758                     new CanCommitTransaction(transactionID2).toSerializable(), timeout);
1759
1760             // Send the AbortTransaction message for the first Tx. This should trigger the 2nd
1761             // Tx to proceed.
1762
1763             shard.tell(new AbortTransaction(transactionID1).toSerializable(), getRef());
1764             expectMsgClass(duration, AbortTransactionReply.SERIALIZABLE_CLASS);
1765
1766             // Wait for the 2nd Tx to complete the canCommit phase.
1767
1768             Await.ready(canCommitFuture, duration);
1769
1770             InOrder inOrder = inOrder(cohort1, cohort2);
1771             inOrder.verify(cohort1).canCommit();
1772             inOrder.verify(cohort2).canCommit();
1773
1774             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1775         }};
1776     }
1777
1778     @Test
1779     public void testCreateSnapshot() throws Exception {
1780         testCreateSnapshot(true, "testCreateSnapshot");
1781     }
1782
1783     @Test
1784     public void testCreateSnapshotWithNonPersistentData() throws Exception {
1785         testCreateSnapshot(false, "testCreateSnapshotWithNonPersistentData");
1786     }
1787
1788     @SuppressWarnings("serial")
1789     public void testCreateSnapshot(final boolean persistent, final String shardActorName) throws Exception{
1790
1791         final AtomicReference<CountDownLatch> latch = new AtomicReference<>(new CountDownLatch(1));
1792
1793         final AtomicReference<Object> savedSnapshot = new AtomicReference<>();
1794         class TestPersistentDataProvider extends DelegatingPersistentDataProvider {
1795             TestPersistentDataProvider(DataPersistenceProvider delegate) {
1796                 super(delegate);
1797             }
1798
1799             @Override
1800             public void saveSnapshot(Object o) {
1801                 savedSnapshot.set(o);
1802                 super.saveSnapshot(o);
1803             }
1804         }
1805
1806         dataStoreContextBuilder.persistent(persistent);
1807
1808         new ShardTestKit(getSystem()) {{
1809             class TestShard extends Shard {
1810
1811                 protected TestShard(ShardIdentifier name, Map<String, String> peerAddresses,
1812                                     DatastoreContext datastoreContext, SchemaContext schemaContext) {
1813                     super(name, peerAddresses, datastoreContext, schemaContext);
1814                     setPersistence(new TestPersistentDataProvider(super.persistence()));
1815                 }
1816
1817                 @Override
1818                 public void handleCommand(Object message) {
1819                     super.handleCommand(message);
1820
1821                     if (message instanceof SaveSnapshotSuccess || message.equals("commit_snapshot")) {
1822                         latch.get().countDown();
1823                     }
1824                 }
1825
1826                 @Override
1827                 public RaftActorContext getRaftActorContext() {
1828                     return super.getRaftActorContext();
1829                 }
1830             }
1831
1832             Creator<Shard> creator = new Creator<Shard>() {
1833                 @Override
1834                 public Shard create() throws Exception {
1835                     return new TestShard(shardID, Collections.<String,String>emptyMap(),
1836                             newDatastoreContext(), SCHEMA_CONTEXT);
1837                 }
1838             };
1839
1840             TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1841                     Props.create(new DelegatingShardCreator(creator)), shardActorName);
1842
1843             waitUntilLeader(shard);
1844
1845             writeToStore(shard, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1846
1847             NormalizedNode<?,?> expectedRoot = readStore(shard, YangInstanceIdentifier.builder().build());
1848
1849             // Trigger creation of a snapshot by ensuring
1850             RaftActorContext raftActorContext = ((TestShard) shard.underlyingActor()).getRaftActorContext();
1851             raftActorContext.getSnapshotManager().capture(mock(ReplicatedLogEntry.class), -1);
1852
1853             assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
1854
1855             assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
1856                     savedSnapshot.get() instanceof Snapshot);
1857
1858             verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
1859
1860             latch.set(new CountDownLatch(1));
1861             savedSnapshot.set(null);
1862
1863             raftActorContext.getSnapshotManager().capture(mock(ReplicatedLogEntry.class), -1);
1864
1865             assertEquals("Snapshot saved", true, latch.get().await(5, TimeUnit.SECONDS));
1866
1867             assertTrue("Invalid saved snapshot " + savedSnapshot.get(),
1868                     savedSnapshot.get() instanceof Snapshot);
1869
1870             verifySnapshot((Snapshot)savedSnapshot.get(), expectedRoot);
1871
1872             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1873         }
1874
1875         private void verifySnapshot(Snapshot snapshot, NormalizedNode<?,?> expectedRoot) {
1876
1877             NormalizedNode<?, ?> actual = SerializationUtils.deserializeNormalizedNode(snapshot.getState());
1878             assertEquals("Root node", expectedRoot, actual);
1879
1880         }};
1881     }
1882
1883     /**
1884      * This test simply verifies that the applySnapShot logic will work
1885      * @throws ReadFailedException
1886      * @throws DataValidationFailedException
1887      */
1888     @Test
1889     public void testInMemoryDataTreeRestore() throws ReadFailedException, DataValidationFailedException {
1890         DataTree store = InMemoryDataTreeFactory.getInstance().create();
1891         store.setSchemaContext(SCHEMA_CONTEXT);
1892
1893         DataTreeModification putTransaction = store.takeSnapshot().newModification();
1894         putTransaction.write(TestModel.TEST_PATH,
1895             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1896         commitTransaction(store, putTransaction);
1897
1898
1899         NormalizedNode<?, ?> expected = readStore(store, YangInstanceIdentifier.builder().build());
1900
1901         DataTreeModification writeTransaction = store.takeSnapshot().newModification();
1902
1903         writeTransaction.delete(YangInstanceIdentifier.builder().build());
1904         writeTransaction.write(YangInstanceIdentifier.builder().build(), expected);
1905
1906         commitTransaction(store, writeTransaction);
1907
1908         NormalizedNode<?, ?> actual = readStore(store, YangInstanceIdentifier.builder().build());
1909
1910         assertEquals(expected, actual);
1911     }
1912
1913     @Test
1914     public void testRecoveryApplicable(){
1915
1916         final DatastoreContext persistentContext = DatastoreContext.newBuilder().
1917                 shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(true).build();
1918
1919         final Props persistentProps = Shard.props(shardID, Collections.<String, String>emptyMap(),
1920                 persistentContext, SCHEMA_CONTEXT);
1921
1922         final DatastoreContext nonPersistentContext = DatastoreContext.newBuilder().
1923                 shardJournalRecoveryLogBatchSize(3).shardSnapshotBatchCount(5000).persistent(false).build();
1924
1925         final Props nonPersistentProps = Shard.props(shardID, Collections.<String, String>emptyMap(),
1926                 nonPersistentContext, SCHEMA_CONTEXT);
1927
1928         new ShardTestKit(getSystem()) {{
1929             TestActorRef<Shard> shard1 = TestActorRef.create(getSystem(),
1930                     persistentProps, "testPersistence1");
1931
1932             assertTrue("Recovery Applicable", shard1.underlyingActor().persistence().isRecoveryApplicable());
1933
1934             shard1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1935
1936             TestActorRef<Shard> shard2 = TestActorRef.create(getSystem(),
1937                     nonPersistentProps, "testPersistence2");
1938
1939             assertFalse("Recovery Not Applicable", shard2.underlyingActor().persistence().isRecoveryApplicable());
1940
1941             shard2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1942
1943         }};
1944
1945     }
1946
1947     @Test
1948     public void testOnDatastoreContext() {
1949         new ShardTestKit(getSystem()) {{
1950             dataStoreContextBuilder.persistent(true);
1951
1952             TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps(), "testOnDatastoreContext");
1953
1954             assertEquals("isRecoveryApplicable", true,
1955                     shard.underlyingActor().persistence().isRecoveryApplicable());
1956
1957             waitUntilLeader(shard);
1958
1959             shard.tell(dataStoreContextBuilder.persistent(false).build(), ActorRef.noSender());
1960
1961             assertEquals("isRecoveryApplicable", false,
1962                     shard.underlyingActor().persistence().isRecoveryApplicable());
1963
1964             shard.tell(dataStoreContextBuilder.persistent(true).build(), ActorRef.noSender());
1965
1966             assertEquals("isRecoveryApplicable", true,
1967                     shard.underlyingActor().persistence().isRecoveryApplicable());
1968
1969             shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
1970         }};
1971     }
1972
1973     @Test
1974     public void testRegisterRoleChangeListener() throws Exception {
1975         new ShardTestKit(getSystem()) {
1976             {
1977                 final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
1978                         newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
1979                         "testRegisterRoleChangeListener");
1980
1981                 waitUntilLeader(shard);
1982
1983                 TestActorRef<MessageCollectorActor> listener =
1984                         TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class));
1985
1986                 shard.tell(new RegisterRoleChangeListener(), listener);
1987
1988                 // TODO: MessageCollectorActor exists as a test util in both the akka-raft and distributed-datastore
1989                 // projects. Need to move it to commons as a regular utility and then we can get rid of this arbitrary
1990                 // sleep.
1991                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
1992
1993                 List<Object> allMatching = MessageCollectorActor.getAllMatching(listener, RegisterRoleChangeListenerReply.class);
1994
1995                 assertEquals(1, allMatching.size());
1996             }
1997         };
1998     }
1999
2000     @Test
2001     public void testFollowerInitialSyncStatus() throws Exception {
2002         final TestActorRef<Shard> shard = TestActorRef.create(getSystem(),
2003                 newShardProps().withDispatcher(Dispatchers.DefaultDispatcherId()),
2004                 "testFollowerInitialSyncStatus");
2005
2006         shard.underlyingActor().onReceiveCommand(new FollowerInitialSyncUpStatus(false, "member-1-shard-inventory-operational"));
2007
2008         assertEquals(false, shard.underlyingActor().getShardMBean().getFollowerInitialSyncStatus());
2009
2010         shard.underlyingActor().onReceiveCommand(new FollowerInitialSyncUpStatus(true, "member-1-shard-inventory-operational"));
2011
2012         assertEquals(true, shard.underlyingActor().getShardMBean().getFollowerInitialSyncStatus());
2013
2014         shard.tell(PoisonPill.getInstance(), ActorRef.noSender());
2015     }
2016
2017     private static void commitTransaction(DataTree store, final DataTreeModification modification) throws DataValidationFailedException {
2018         modification.ready();
2019         store.validate(modification);
2020         store.commit(store.prepare(modification));
2021     }
2022 }

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