Convert mdsal submit() calls to commit()
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / DistributedDataStoreIntegrationTest.java
1 /*
2  * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8
9 package org.opendaylight.controller.cluster.datastore;
10
11 import static org.junit.Assert.assertEquals;
12 import static org.junit.Assert.assertNotNull;
13 import static org.junit.Assert.assertTrue;
14 import static org.junit.Assert.fail;
15 import static org.mockito.Matchers.any;
16 import static org.mockito.Matchers.eq;
17 import static org.mockito.Mockito.timeout;
18 import static org.mockito.Mockito.verify;
19
20 import akka.actor.ActorSystem;
21 import akka.actor.Address;
22 import akka.actor.AddressFromURIString;
23 import akka.cluster.Cluster;
24 import akka.testkit.javadsl.TestKit;
25 import com.google.common.base.Optional;
26 import com.google.common.base.Throwables;
27 import com.google.common.collect.ImmutableMap;
28 import com.google.common.util.concurrent.CheckedFuture;
29 import com.google.common.util.concurrent.ListenableFuture;
30 import com.google.common.util.concurrent.MoreExecutors;
31 import com.google.common.util.concurrent.Uninterruptibles;
32 import com.typesafe.config.ConfigFactory;
33 import java.io.IOException;
34 import java.math.BigInteger;
35 import java.util.ArrayList;
36 import java.util.Arrays;
37 import java.util.Collection;
38 import java.util.Collections;
39 import java.util.List;
40 import java.util.concurrent.CountDownLatch;
41 import java.util.concurrent.ExecutionException;
42 import java.util.concurrent.TimeUnit;
43 import java.util.concurrent.atomic.AtomicReference;
44 import org.junit.After;
45 import org.junit.Before;
46 import org.junit.Test;
47 import org.junit.runner.RunWith;
48 import org.junit.runners.Parameterized;
49 import org.junit.runners.Parameterized.Parameter;
50 import org.junit.runners.Parameterized.Parameters;
51 import org.mockito.Mockito;
52 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
53 import org.opendaylight.controller.cluster.databroker.ClientBackedDataStore;
54 import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
55 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
56 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
57 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
58 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
59 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
60 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
61 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
62 import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
63 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
64 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
65 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
66 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
67 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
68 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
69 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
70 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
71 import org.opendaylight.mdsal.common.api.ReadFailedException;
72 import org.opendaylight.mdsal.common.api.TransactionChainClosedException;
73 import org.opendaylight.mdsal.common.api.TransactionChainListener;
74 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
75 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
76 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
77 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
78 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
79 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
80 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
81 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
82 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
83 import org.opendaylight.yangtools.concepts.ListenerRegistration;
84 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
85 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
86 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
87 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
88 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
89 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
90 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
91 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
92 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
93 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
94
95 @RunWith(Parameterized.class)
96 public class DistributedDataStoreIntegrationTest {
97
98     @Parameters(name = "{0}")
99     public static Collection<Object[]> data() {
100         return Arrays.asList(new Object[][] {
101                 { DistributedDataStore.class }, { ClientBackedDataStore.class }
102         });
103     }
104
105     @Parameter
106     public Class<? extends AbstractDataStore> testParameter;
107
108     private ActorSystem system;
109
110     private final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder()
111             .shardHeartbeatIntervalInMillis(100);
112
113     @Before
114     public void setUp() throws IOException {
115         InMemorySnapshotStore.clear();
116         InMemoryJournal.clear();
117         system = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
118         Address member1Address = AddressFromURIString.parse("akka://cluster-test@127.0.0.1:2558");
119         Cluster.get(system).join(member1Address);
120     }
121
122     @After
123     public void tearDown() throws IOException {
124         TestKit.shutdownActorSystem(system, Boolean.TRUE);
125         system = null;
126     }
127
128     protected ActorSystem getSystem() {
129         return system;
130     }
131
132     @Test
133     public void testWriteTransactionWithSingleShard() throws Exception {
134         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
135             {
136                 try (AbstractDataStore dataStore = setupAbstractDataStore(
137                         testParameter, "transactionIntegrationTest", "test-1")) {
138
139                     testWriteTransaction(dataStore, TestModel.TEST_PATH,
140                             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
141
142                     testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
143                             ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
144                 }
145             }
146         };
147     }
148
149     @Test
150     public void testWriteTransactionWithMultipleShards() throws Exception {
151         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
152             {
153                 try (AbstractDataStore dataStore = setupAbstractDataStore(
154                         testParameter, "testWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
155
156                     DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
157                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
158
159                     writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
160                     writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
161
162                     doCommit(writeTx.ready());
163
164                     writeTx = dataStore.newWriteOnlyTransaction();
165
166                     writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
167                     writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
168
169                     doCommit(writeTx.ready());
170
171                     writeTx = dataStore.newWriteOnlyTransaction();
172
173                     final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
174                     final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
175                     writeTx.write(carPath, car);
176
177                     final MapEntryNode person = PeopleModel.newPersonEntry("jack");
178                     final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
179                     writeTx.write(personPath, person);
180
181                     doCommit(writeTx.ready());
182
183                     // Verify the data in the store
184                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
185
186                     Optional<NormalizedNode<?, ?>> optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
187                     assertEquals("isPresent", true, optional.isPresent());
188                     assertEquals("Data node", car, optional.get());
189
190                     optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
191                     assertEquals("isPresent", true, optional.isPresent());
192                     assertEquals("Data node", person, optional.get());
193                 }
194             }
195         };
196     }
197
198     @Test
199     public void testReadWriteTransactionWithSingleShard() throws Exception {
200         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
201             {
202                 try (AbstractDataStore dataStore = setupAbstractDataStore(
203                         testParameter, "testReadWriteTransactionWithSingleShard", "test-1")) {
204
205                     // 1. Create a read-write Tx
206                     final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
207                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
208
209                     // 2. Write some data
210                     final YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
211                     final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
212                     readWriteTx.write(nodePath, nodeToWrite);
213
214                     // 3. Read the data from Tx
215                     final Boolean exists = readWriteTx.exists(nodePath).checkedGet(5, TimeUnit.SECONDS);
216                     assertEquals("exists", true, exists);
217
218                     Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
219                     assertEquals("isPresent", true, optional.isPresent());
220                     assertEquals("Data node", nodeToWrite, optional.get());
221
222                     // 4. Ready the Tx for commit
223                     final DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
224
225                     // 5. Commit the Tx
226                     doCommit(cohort);
227
228                     // 6. Verify the data in the store
229                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
230
231                     optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
232                     assertEquals("isPresent", true, optional.isPresent());
233                     assertEquals("Data node", nodeToWrite, optional.get());
234                 }
235             }
236         };
237     }
238
239     @Test
240     public void testReadWriteTransactionWithMultipleShards() throws Exception {
241         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
242             {
243                 try (AbstractDataStore dataStore = setupAbstractDataStore(
244                         testParameter, "testReadWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
245
246                     DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
247                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
248
249                     readWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
250                     readWriteTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
251
252                     doCommit(readWriteTx.ready());
253
254                     readWriteTx = dataStore.newReadWriteTransaction();
255
256                     readWriteTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
257                     readWriteTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
258
259                     doCommit(readWriteTx.ready());
260
261                     readWriteTx = dataStore.newReadWriteTransaction();
262
263                     final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
264                     final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
265                     readWriteTx.write(carPath, car);
266
267                     final MapEntryNode person = PeopleModel.newPersonEntry("jack");
268                     final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
269                     readWriteTx.write(personPath, person);
270
271                     final Boolean exists = readWriteTx.exists(carPath).checkedGet(5, TimeUnit.SECONDS);
272                     assertEquals("exists", true, exists);
273
274                     Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
275                     assertEquals("isPresent", true, optional.isPresent());
276                     assertEquals("Data node", car, optional.get());
277
278                     doCommit(readWriteTx.ready());
279
280                     // Verify the data in the store
281                     DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
282
283                     optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
284                     assertEquals("isPresent", true, optional.isPresent());
285                     assertEquals("Data node", car, optional.get());
286
287                     optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
288                     assertEquals("isPresent", true, optional.isPresent());
289                     assertEquals("Data node", person, optional.get());
290
291                 }
292             }
293         };
294     }
295
296     @Test
297     public void testSingleTransactionsWritesInQuickSuccession() throws Exception {
298         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
299             {
300                 try (AbstractDataStore dataStore = setupAbstractDataStore(
301                         testParameter, "testSingleTransactionsWritesInQuickSuccession", "cars-1")) {
302
303                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
304
305                     DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
306                     writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
307                     writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
308                     doCommit(writeTx.ready());
309
310                     writeTx = txChain.newWriteOnlyTransaction();
311
312                     int numCars = 5;
313                     for (int i = 0; i < numCars; i++) {
314                         writeTx.write(CarsModel.newCarPath("car" + i),
315                                 CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
316                     }
317
318                     doCommit(writeTx.ready());
319
320                     final Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction()
321                             .read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
322                     assertEquals("isPresent", true, optional.isPresent());
323                     assertEquals("# cars", numCars, ((Collection<?>) optional.get().getValue()).size());
324                 }
325             }
326         };
327     }
328
329     @SuppressWarnings("checkstyle:IllegalCatch")
330     private void testTransactionWritesWithShardNotInitiallyReady(final String testName, final boolean writeOnly)
331             throws Exception {
332         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
333             {
334                 final String shardName = "test-1";
335
336                 // Setup the InMemoryJournal to block shard recovery to ensure
337                 // the shard isn't
338                 // initialized until we create and submit the write the Tx.
339                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
340                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
341                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
342
343                 try (AbstractDataStore dataStore = setupAbstractDataStore(
344                         testParameter, testName, false, shardName)) {
345
346                     // Create the write Tx
347                     final DOMStoreWriteTransaction writeTx = writeOnly ? dataStore.newWriteOnlyTransaction()
348                             : dataStore.newReadWriteTransaction();
349                     assertNotNull("newReadWriteTransaction returned null", writeTx);
350
351                     // Do some modification operations and ready the Tx on a
352                     // separate thread.
353                     final YangInstanceIdentifier listEntryPath = YangInstanceIdentifier
354                             .builder(TestModel.OUTER_LIST_PATH)
355                             .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
356
357                     final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
358                     final AtomicReference<Exception> caughtEx = new AtomicReference<>();
359                     final CountDownLatch txReady = new CountDownLatch(1);
360                     final Thread txThread = new Thread(() -> {
361                         try {
362                             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
363
364                             writeTx.merge(TestModel.OUTER_LIST_PATH,
365                                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
366
367                             writeTx.write(listEntryPath,
368                                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
369
370                             writeTx.delete(listEntryPath);
371
372                             txCohort.set(writeTx.ready());
373                         } catch (Exception e) {
374                             caughtEx.set(e);
375                         } finally {
376                             txReady.countDown();
377                         }
378                     });
379
380                     txThread.start();
381
382                     // Wait for the Tx operations to complete.
383                     final boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
384                     if (caughtEx.get() != null) {
385                         throw caughtEx.get();
386                     }
387
388                     assertEquals("Tx ready", true, done);
389
390                     // At this point the Tx operations should be waiting for the
391                     // shard to initialize so
392                     // trigger the latch to let the shard recovery to continue.
393                     blockRecoveryLatch.countDown();
394
395                     // Wait for the Tx commit to complete.
396                     doCommit(txCohort.get());
397
398                     // Verify the data in the store
399                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
400
401                     Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
402                     assertEquals("isPresent", true, optional.isPresent());
403
404                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
405                     assertEquals("isPresent", true, optional.isPresent());
406
407                     optional = readTx.read(listEntryPath).get(5, TimeUnit.SECONDS);
408                     assertEquals("isPresent", false, optional.isPresent());
409                 }
410             }
411         };
412     }
413
414     @Test
415     public void testWriteOnlyTransactionWithShardNotInitiallyReady() throws Exception {
416         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
417         testTransactionWritesWithShardNotInitiallyReady("testWriteOnlyTransactionWithShardNotInitiallyReady", true);
418     }
419
420     @Test
421     public void testReadWriteTransactionWithShardNotInitiallyReady() throws Exception {
422         testTransactionWritesWithShardNotInitiallyReady("testReadWriteTransactionWithShardNotInitiallyReady", false);
423     }
424
425     @Test
426     @SuppressWarnings("checkstyle:IllegalCatch")
427     public void testTransactionReadsWithShardNotInitiallyReady() throws Exception {
428         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
429             {
430                 final String testName = "testTransactionReadsWithShardNotInitiallyReady";
431                 final String shardName = "test-1";
432
433                 // Setup the InMemoryJournal to block shard recovery to ensure
434                 // the shard isn't
435                 // initialized until we create the Tx.
436                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
437                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
438                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
439
440                 try (AbstractDataStore dataStore = setupAbstractDataStore(
441                         testParameter, testName, false, shardName)) {
442
443                     // Create the read-write Tx
444                     final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
445                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
446
447                     // Do some reads on the Tx on a separate thread.
448                     final AtomicReference<CheckedFuture<Boolean, ReadFailedException>> txExistsFuture =
449                             new AtomicReference<>();
450                     final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
451                             txReadFuture = new AtomicReference<>();
452                     final AtomicReference<Exception> caughtEx = new AtomicReference<>();
453                     final CountDownLatch txReadsDone = new CountDownLatch(1);
454                     final Thread txThread = new Thread(() -> {
455                         try {
456                             readWriteTx.write(TestModel.TEST_PATH,
457                                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
458
459                             txExistsFuture.set(readWriteTx.exists(TestModel.TEST_PATH));
460
461                             txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
462                         } catch (Exception e) {
463                             caughtEx.set(e);
464                         } finally {
465                             txReadsDone.countDown();
466                         }
467                     });
468
469                     txThread.start();
470
471                     // Wait for the Tx operations to complete.
472                     boolean done = Uninterruptibles.awaitUninterruptibly(txReadsDone, 5, TimeUnit.SECONDS);
473                     if (caughtEx.get() != null) {
474                         throw caughtEx.get();
475                     }
476
477                     assertEquals("Tx reads done", true, done);
478
479                     // At this point the Tx operations should be waiting for the
480                     // shard to initialize so
481                     // trigger the latch to let the shard recovery to continue.
482                     blockRecoveryLatch.countDown();
483
484                     // Wait for the reads to complete and verify.
485                     assertEquals("exists", true, txExistsFuture.get().checkedGet(5, TimeUnit.SECONDS));
486                     assertEquals("read", true, txReadFuture.get().checkedGet(5, TimeUnit.SECONDS).isPresent());
487
488                     readWriteTx.close();
489                 }
490             }
491         };
492     }
493
494     @Test(expected = NotInitializedException.class)
495     @SuppressWarnings("checkstyle:IllegalCatch")
496     public void testTransactionCommitFailureWithShardNotInitialized() throws Exception {
497         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
498             {
499                 final String testName = "testTransactionCommitFailureWithShardNotInitialized";
500                 final String shardName = "test-1";
501
502                 // Set the shard initialization timeout low for the test.
503                 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
504
505                 // Setup the InMemoryJournal to block shard recovery
506                 // indefinitely.
507                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
508                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
509                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
510
511                 InMemoryJournal.addEntry(persistentID, 1, "Dummy data so akka will read from persistence");
512
513                 final AbstractDataStore dataStore =
514                         setupAbstractDataStore(testParameter, testName, false, shardName);
515
516                 // Create the write Tx
517                 final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
518                 assertNotNull("newReadWriteTransaction returned null", writeTx);
519
520                 // Do some modifications and ready the Tx on a separate
521                 // thread.
522                 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
523                 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
524                 final CountDownLatch txReady = new CountDownLatch(1);
525                 final Thread txThread = new Thread(() -> {
526                     try {
527                         writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
528
529                         txCohort.set(writeTx.ready());
530                     } catch (Exception e) {
531                         caughtEx.set(e);
532                     } finally {
533                         txReady.countDown();
534                     }
535                 });
536
537                 txThread.start();
538
539                 // Wait for the Tx operations to complete.
540                 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
541                 if (caughtEx.get() != null) {
542                     throw caughtEx.get();
543                 }
544
545                 assertEquals("Tx ready", true, done);
546
547                 // Wait for the commit to complete. Since the shard never
548                 // initialized, the Tx should
549                 // have timed out and throw an appropriate exception cause.
550                 try {
551                     txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
552                     fail("Expected NotInitializedException");
553                 } catch (final Exception e) {
554                     final Throwable root = Throwables.getRootCause(e);
555                     Throwables.throwIfUnchecked(root);
556                     throw new RuntimeException(root);
557                 } finally {
558                     blockRecoveryLatch.countDown();
559                 }
560             }
561         };
562     }
563
564     @Test(expected = NotInitializedException.class)
565     @SuppressWarnings("checkstyle:IllegalCatch")
566     public void testTransactionReadFailureWithShardNotInitialized() throws Exception {
567         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
568             {
569                 final String testName = "testTransactionReadFailureWithShardNotInitialized";
570                 final String shardName = "test-1";
571
572                 // Set the shard initialization timeout low for the test.
573                 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
574
575                 // Setup the InMemoryJournal to block shard recovery
576                 // indefinitely.
577                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
578                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
579                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
580
581                 InMemoryJournal.addEntry(persistentID, 1, "Dummy data so akka will read from persistence");
582
583                 try (AbstractDataStore dataStore = setupAbstractDataStore(
584                         testParameter, testName, false, shardName)) {
585
586                     // Create the read-write Tx
587                     final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
588                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
589
590                     // Do a read on the Tx on a separate thread.
591                     final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
592                             txReadFuture = new AtomicReference<>();
593                     final AtomicReference<Exception> caughtEx = new AtomicReference<>();
594                     final CountDownLatch txReadDone = new CountDownLatch(1);
595                     final Thread txThread = new Thread(() -> {
596                         try {
597                             readWriteTx.write(TestModel.TEST_PATH,
598                                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
599
600                             txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
601
602                             readWriteTx.close();
603                         } catch (Exception e) {
604                             caughtEx.set(e);
605                         } finally {
606                             txReadDone.countDown();
607                         }
608                     });
609
610                     txThread.start();
611
612                     // Wait for the Tx operations to complete.
613                     boolean done = Uninterruptibles.awaitUninterruptibly(txReadDone, 5, TimeUnit.SECONDS);
614                     if (caughtEx.get() != null) {
615                         throw caughtEx.get();
616                     }
617
618                     assertEquals("Tx read done", true, done);
619
620                     // Wait for the read to complete. Since the shard never
621                     // initialized, the Tx should
622                     // have timed out and throw an appropriate exception cause.
623                     try {
624                         txReadFuture.get().checkedGet(5, TimeUnit.SECONDS);
625                         fail("Expected NotInitializedException");
626                     } catch (final ReadFailedException e) {
627                         final Throwable root = Throwables.getRootCause(e);
628                         Throwables.throwIfUnchecked(root);
629                         throw new RuntimeException(root);
630                     } finally {
631                         blockRecoveryLatch.countDown();
632                     }
633                 }
634             }
635         };
636     }
637
638     @SuppressWarnings("checkstyle:IllegalCatch")
639     private void testTransactionCommitFailureWithNoShardLeader(final boolean writeOnly, final String testName)
640             throws Exception {
641         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
642             {
643                 final String shardName = "default";
644
645                 // We don't want the shard to become the leader so prevent shard
646                 // elections.
647                 datastoreContextBuilder.customRaftPolicyImplementation(
648                         "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
649
650                 // The ShardManager uses the election timeout for FindPrimary so
651                 // reset it low so it will timeout quickly.
652                 datastoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1)
653                         .shardInitializationTimeout(200, TimeUnit.MILLISECONDS).frontendRequestTimeoutInSeconds(2);
654
655                 try (AbstractDataStore dataStore = setupAbstractDataStore(
656                         testParameter, testName, false, shardName)) {
657
658                     final Object result = dataStore.getActorContext().executeOperation(
659                             dataStore.getActorContext().getShardManager(), new FindLocalShard(shardName, true));
660                     assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
661
662                     // Create the write Tx.
663                     DOMStoreWriteTransaction writeTxToClose = null;
664                     try {
665                         writeTxToClose = writeOnly ? dataStore.newWriteOnlyTransaction()
666                                 : dataStore.newReadWriteTransaction();
667                         final DOMStoreWriteTransaction writeTx = writeTxToClose;
668                         assertNotNull("newReadWriteTransaction returned null", writeTx);
669
670                         // Do some modifications and ready the Tx on a separate
671                         // thread.
672                         final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
673                         final AtomicReference<Exception> caughtEx = new AtomicReference<>();
674                         final CountDownLatch txReady = new CountDownLatch(1);
675                         final Thread txThread = new Thread(() -> {
676                             try {
677                                 writeTx.write(TestModel.JUNK_PATH,
678                                         ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
679
680                                 txCohort.set(writeTx.ready());
681                             } catch (Exception e) {
682                                 caughtEx.set(e);
683                             } finally {
684                                 txReady.countDown();
685                             }
686                         });
687
688                         txThread.start();
689
690                         // Wait for the Tx operations to complete.
691                         boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
692                         if (caughtEx.get() != null) {
693                             throw caughtEx.get();
694                         }
695
696                         assertEquals("Tx ready", true, done);
697
698                         // Wait for the commit to complete. Since no shard
699                         // leader was elected in time, the Tx
700                         // should have timed out and throw an appropriate
701                         // exception cause.
702                         try {
703                             txCohort.get().canCommit().get(10, TimeUnit.SECONDS);
704                             fail("Expected NoShardLeaderException");
705                         } catch (final ExecutionException e) {
706                             final String msg = "Unexpected exception: "
707                                     + Throwables.getStackTraceAsString(e.getCause());
708                             if (DistributedDataStore.class.equals(testParameter)) {
709                                 assertTrue(Throwables.getRootCause(e) instanceof NoShardLeaderException);
710                             } else {
711                                 assertTrue(msg, Throwables.getRootCause(e) instanceof RequestTimeoutException);
712                             }
713                         }
714                     } finally {
715                         try {
716                             if (writeTxToClose != null) {
717                                 writeTxToClose.close();
718                             }
719                         } catch (Exception e) {
720                             // FIXME TransactionProxy.close throws IllegalStateException:
721                             // Transaction is ready, it cannot be closed
722                         }
723                     }
724                 }
725             }
726         };
727     }
728
729     @Test
730     public void testWriteOnlyTransactionCommitFailureWithNoShardLeader() throws Exception {
731         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
732         testTransactionCommitFailureWithNoShardLeader(true, "testWriteOnlyTransactionCommitFailureWithNoShardLeader");
733     }
734
735     @Test
736     public void testReadWriteTransactionCommitFailureWithNoShardLeader() throws Exception {
737         testTransactionCommitFailureWithNoShardLeader(false, "testReadWriteTransactionCommitFailureWithNoShardLeader");
738     }
739
740     @Test
741     public void testTransactionAbort() throws Exception {
742         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
743             {
744                 try (AbstractDataStore dataStore = setupAbstractDataStore(
745                         testParameter, "transactionAbortIntegrationTest", "test-1")) {
746
747                     final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
748                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
749
750                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
751
752                     final DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
753
754                     cohort.canCommit().get(5, TimeUnit.SECONDS);
755
756                     cohort.abort().get(5, TimeUnit.SECONDS);
757
758                     testWriteTransaction(dataStore, TestModel.TEST_PATH,
759                             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
760                 }
761             }
762         };
763     }
764
765     @Test
766     @SuppressWarnings("checkstyle:IllegalCatch")
767     public void testTransactionChainWithSingleShard() throws Exception {
768         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
769             {
770                 try (AbstractDataStore dataStore = setupAbstractDataStore(
771                         testParameter, "testTransactionChainWithSingleShard", "test-1")) {
772
773                     // 1. Create a Tx chain and write-only Tx
774                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
775
776                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
777                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
778
779                     // 2. Write some data
780                     final NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
781                     writeTx.write(TestModel.TEST_PATH, testNode);
782
783                     // 3. Ready the Tx for commit
784                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
785
786                     // 4. Commit the Tx on another thread that first waits for
787                     // the second read Tx.
788                     final CountDownLatch continueCommit1 = new CountDownLatch(1);
789                     final CountDownLatch commit1Done = new CountDownLatch(1);
790                     final AtomicReference<Exception> commit1Error = new AtomicReference<>();
791                     new Thread(() -> {
792                         try {
793                             continueCommit1.await();
794                             doCommit(cohort1);
795                         } catch (Exception e) {
796                             commit1Error.set(e);
797                         } finally {
798                             commit1Done.countDown();
799                         }
800                     }).start();
801
802                     // 5. Create a new read Tx from the chain to read and verify
803                     // the data from the first
804                     // Tx is visible after being readied.
805                     DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
806                     Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
807                     assertEquals("isPresent", true, optional.isPresent());
808                     assertEquals("Data node", testNode, optional.get());
809
810                     // 6. Create a new RW Tx from the chain, write more data,
811                     // and ready it
812                     final DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
813                     final MapNode outerNode = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build();
814                     rwTx.write(TestModel.OUTER_LIST_PATH, outerNode);
815
816                     final DOMStoreThreePhaseCommitCohort cohort2 = rwTx.ready();
817
818                     // 7. Create a new read Tx from the chain to read the data
819                     // from the last RW Tx to
820                     // verify it is visible.
821                     readTx = txChain.newReadWriteTransaction();
822                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
823                     assertEquals("isPresent", true, optional.isPresent());
824                     assertEquals("Data node", outerNode, optional.get());
825
826                     // 8. Wait for the 2 commits to complete and close the
827                     // chain.
828                     continueCommit1.countDown();
829                     Uninterruptibles.awaitUninterruptibly(commit1Done, 5, TimeUnit.SECONDS);
830
831                     if (commit1Error.get() != null) {
832                         throw commit1Error.get();
833                     }
834
835                     doCommit(cohort2);
836
837                     txChain.close();
838
839                     // 9. Create a new read Tx from the data store and verify
840                     // committed data.
841                     readTx = dataStore.newReadOnlyTransaction();
842                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
843                     assertEquals("isPresent", true, optional.isPresent());
844                     assertEquals("Data node", outerNode, optional.get());
845                 }
846             }
847         };
848     }
849
850     @Test
851     public void testTransactionChainWithMultipleShards() throws Exception {
852         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
853             {
854                 try (AbstractDataStore dataStore = setupAbstractDataStore(
855                         testParameter, "testTransactionChainWithMultipleShards", "cars-1", "people-1")) {
856
857                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
858
859                     DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
860                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
861
862                     writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
863                     writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
864
865                     writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
866                     writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
867
868                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
869
870                     final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
871
872                     final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
873                     final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
874                     readWriteTx.write(carPath, car);
875
876                     final MapEntryNode person = PeopleModel.newPersonEntry("jack");
877                     final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
878                     readWriteTx.merge(personPath, person);
879
880                     Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
881                     assertEquals("isPresent", true, optional.isPresent());
882                     assertEquals("Data node", car, optional.get());
883
884                     optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
885                     assertEquals("isPresent", true, optional.isPresent());
886                     assertEquals("Data node", person, optional.get());
887
888                     final DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
889
890                     writeTx = txChain.newWriteOnlyTransaction();
891
892                     writeTx.delete(carPath);
893
894                     final DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
895
896                     final ListenableFuture<Boolean> canCommit1 = cohort1.canCommit();
897                     final ListenableFuture<Boolean> canCommit2 = cohort2.canCommit();
898
899                     doCommit(canCommit1, cohort1);
900                     doCommit(canCommit2, cohort2);
901                     doCommit(cohort3);
902
903                     txChain.close();
904
905                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
906
907                     optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
908                     assertEquals("isPresent", false, optional.isPresent());
909
910                     optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
911                     assertEquals("isPresent", true, optional.isPresent());
912                     assertEquals("Data node", person, optional.get());
913                 }
914             }
915         };
916     }
917
918     @Test
919     public void testCreateChainedTransactionsInQuickSuccession() throws Exception {
920         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
921             {
922                 try (AbstractDataStore dataStore = setupAbstractDataStore(
923                         testParameter, "testCreateChainedTransactionsInQuickSuccession", "cars-1")) {
924
925                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
926                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
927                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
928                             MoreExecutors.directExecutor());
929
930                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
931                     DOMTransactionChain txChain = broker.createTransactionChain(listener);
932
933                     final List<ListenableFuture<?>> futures = new ArrayList<>();
934
935                     final DOMDataTreeWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
936                     writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, CarsModel.emptyContainer());
937                     writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
938                     futures.add(writeTx.commit());
939
940                     int numCars = 100;
941                     for (int i = 0; i < numCars; i++) {
942                         final DOMDataTreeReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
943
944                         rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i),
945                                 CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
946
947                         futures.add(rwTx.commit());
948                     }
949
950                     for (final ListenableFuture<?> f : futures) {
951                         f.get(5, TimeUnit.SECONDS);
952                     }
953
954                     final Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction()
955                             .read(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
956                     assertEquals("isPresent", true, optional.isPresent());
957                     assertEquals("# cars", numCars, ((Collection<?>) optional.get().getValue()).size());
958
959                     txChain.close();
960
961                     broker.close();
962                 }
963             }
964         };
965     }
966
967     @Test
968     public void testCreateChainedTransactionAfterEmptyTxReadied() throws Exception {
969         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
970             {
971                 try (AbstractDataStore dataStore = setupAbstractDataStore(
972                         testParameter, "testCreateChainedTransactionAfterEmptyTxReadied", "test-1")) {
973
974                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
975
976                     final DOMStoreReadWriteTransaction rwTx1 = txChain.newReadWriteTransaction();
977
978                     rwTx1.ready();
979
980                     final DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction();
981
982                     final Optional<NormalizedNode<?, ?>> optional = rwTx2.read(TestModel.TEST_PATH).get(
983                             5, TimeUnit.SECONDS);
984                     assertEquals("isPresent", false, optional.isPresent());
985
986                     txChain.close();
987                 }
988             }
989         };
990     }
991
992     @Test
993     public void testCreateChainedTransactionWhenPreviousNotReady() throws Exception {
994         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
995             {
996                 try (AbstractDataStore dataStore = setupAbstractDataStore(
997                         testParameter, "testCreateChainedTransactionWhenPreviousNotReady", "test-1")) {
998
999                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
1000
1001                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
1002                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
1003
1004                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1005
1006                     // Try to create another Tx of each type - each should fail
1007                     // b/c the previous Tx wasn't
1008                     // readied.
1009                     assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
1010                 }
1011             }
1012         };
1013     }
1014
1015     @Test
1016     public void testCreateChainedTransactionAfterClose() throws Exception {
1017         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1018             {
1019                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1020                         testParameter, "testCreateChainedTransactionAfterClose", "test-1")) {
1021
1022                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
1023                     txChain.close();
1024
1025                     // Try to create another Tx of each type - should fail b/c
1026                     // the previous Tx was closed.
1027                     assertExceptionOnTxChainCreates(txChain, TransactionChainClosedException.class);
1028                 }
1029             }
1030         };
1031     }
1032
1033     @Test
1034     public void testChainWithReadOnlyTxAfterPreviousReady() throws Exception {
1035         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1036             {
1037                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1038                         testParameter, "testChainWithReadOnlyTxAfterPreviousReady", "test-1")) {
1039
1040                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
1041
1042                     // Create a write tx and submit.
1043                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
1044                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1045                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
1046
1047                     // Create read-only tx's and issue a read.
1048                     CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture1 = txChain
1049                             .newReadOnlyTransaction().read(TestModel.TEST_PATH);
1050
1051                     CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture2 = txChain
1052                             .newReadOnlyTransaction().read(TestModel.TEST_PATH);
1053
1054                     // Create another write tx and issue the write.
1055                     DOMStoreWriteTransaction writeTx2 = txChain.newWriteOnlyTransaction();
1056                     writeTx2.write(TestModel.OUTER_LIST_PATH,
1057                             ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1058
1059                     // Ensure the reads succeed.
1060
1061                     assertEquals("isPresent", true, readFuture1.checkedGet(5, TimeUnit.SECONDS).isPresent());
1062                     assertEquals("isPresent", true, readFuture2.checkedGet(5, TimeUnit.SECONDS).isPresent());
1063
1064                     // Ensure the writes succeed.
1065                     DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
1066
1067                     doCommit(cohort1);
1068                     doCommit(cohort2);
1069
1070                     assertEquals("isPresent", true, txChain.newReadOnlyTransaction().read(TestModel.OUTER_LIST_PATH)
1071                             .checkedGet(5, TimeUnit.SECONDS).isPresent());
1072                 }
1073             }
1074         };
1075     }
1076
1077     @Test
1078     public void testChainedTransactionFailureWithSingleShard() throws Exception {
1079         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1080             {
1081                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1082                         testParameter, "testChainedTransactionFailureWithSingleShard", "cars-1")) {
1083
1084                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
1085                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
1086                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
1087                             MoreExecutors.directExecutor());
1088
1089                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
1090                     final DOMTransactionChain txChain = broker.createTransactionChain(listener);
1091
1092                     final DOMDataTreeReadWriteTransaction writeTx = txChain.newReadWriteTransaction();
1093
1094                     writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
1095                             PeopleModel.emptyContainer());
1096
1097                     final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
1098                             .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
1099                             .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
1100
1101                     writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
1102
1103                     try {
1104                         writeTx.commit().get(5, TimeUnit.SECONDS);
1105                         fail("Expected TransactionCommitFailedException");
1106                     } catch (final ExecutionException e) {
1107                         // Expected
1108                     }
1109
1110                     verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
1111                             any(Throwable.class));
1112
1113                     txChain.close();
1114                     broker.close();
1115                 }
1116             }
1117         };
1118     }
1119
1120     @Test
1121     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
1122         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1123             {
1124                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1125                         testParameter, "testChainedTransactionFailureWithMultipleShards", "cars-1", "people-1")) {
1126
1127                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
1128                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
1129                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
1130                             MoreExecutors.directExecutor());
1131
1132                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
1133                     final DOMTransactionChain txChain = broker.createTransactionChain(listener);
1134
1135                     final DOMDataTreeWriteTransaction writeTx = txChain.newReadWriteTransaction();
1136
1137                     writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
1138                             PeopleModel.emptyContainer());
1139
1140                     final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
1141                             .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
1142                             .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
1143
1144                     writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
1145
1146                     // Note that merge will validate the data and fail but put
1147                     // succeeds b/c deep validation is not
1148                     // done for put for performance reasons.
1149                     try {
1150                         writeTx.commit().get(5, TimeUnit.SECONDS);
1151                         fail("Expected TransactionCommitFailedException");
1152                     } catch (final ExecutionException e) {
1153                         // Expected
1154                     }
1155
1156                     verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
1157                             any(Throwable.class));
1158
1159                     txChain.close();
1160                     broker.close();
1161                 }
1162             }
1163         };
1164     }
1165
1166     @Test
1167     public void testDataTreeChangeListenerRegistration() throws Exception {
1168         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1169             {
1170                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1171                         testParameter, "testDataTreeChangeListenerRegistration", "test-1")) {
1172
1173                     testWriteTransaction(dataStore, TestModel.TEST_PATH,
1174                             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1175
1176                     final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
1177
1178                     ListenerRegistration<MockDataTreeChangeListener> listenerReg = dataStore
1179                             .registerTreeChangeListener(TestModel.TEST_PATH, listener);
1180
1181                     assertNotNull("registerTreeChangeListener returned null", listenerReg);
1182
1183                     IntegrationTestKit.verifyShardState(dataStore, "test-1",
1184                         state -> assertEquals("getTreeChangeListenerActors", 1,
1185                                 state.getTreeChangeListenerActors().size()));
1186
1187                     // Wait for the initial notification
1188                     listener.waitForChangeEvents(TestModel.TEST_PATH);
1189                     listener.reset(2);
1190
1191                     // Write 2 updates.
1192                     testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
1193                             ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1194
1195                     YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1196                             .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
1197                     testWriteTransaction(dataStore, listPath,
1198                             ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
1199
1200                     // Wait for the 2 updates.
1201                     listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
1202                     listenerReg.close();
1203
1204                     IntegrationTestKit.verifyShardState(dataStore, "test-1",
1205                         state -> assertEquals("getTreeChangeListenerActors", 0,
1206                                 state.getTreeChangeListenerActors().size()));
1207
1208                     testWriteTransaction(dataStore,
1209                             YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1210                                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
1211                             ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
1212
1213                     listener.expectNoMoreChanges("Received unexpected change after close");
1214                 }
1215             }
1216         };
1217     }
1218
1219     @Test
1220     public void testRestoreFromDatastoreSnapshot() throws Exception {
1221         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1222             {
1223                 final String name = "transactionIntegrationTest";
1224
1225                 final ContainerNode carsNode = CarsModel.newCarsNode(
1226                         CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)),
1227                                 CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L))));
1228
1229                 DataTree dataTree = new InMemoryDataTreeFactory().create(
1230                     DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full());
1231                 AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
1232                 NormalizedNode<?, ?> root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
1233
1234                 final Snapshot carsSnapshot = Snapshot.create(
1235                         new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
1236                         Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
1237
1238                 dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL,
1239                     SchemaContextHelper.full());
1240
1241                 final NormalizedNode<?, ?> peopleNode = PeopleModel.create();
1242                 AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode);
1243
1244                 root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
1245
1246                 final Snapshot peopleSnapshot = Snapshot.create(
1247                         new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
1248                         Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
1249
1250                 restoreFromSnapshot = new DatastoreSnapshot(name, null, Arrays.asList(
1251                         new DatastoreSnapshot.ShardSnapshot("cars", carsSnapshot),
1252                         new DatastoreSnapshot.ShardSnapshot("people", peopleSnapshot)));
1253
1254                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1255                         testParameter, name, "module-shards-member1.conf", true, "cars", "people")) {
1256
1257                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
1258
1259                     // two reads
1260                     Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1261                     assertEquals("isPresent", true, optional.isPresent());
1262                     assertEquals("Data node", carsNode, optional.get());
1263
1264                     optional = readTx.read(PeopleModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1265                     assertEquals("isPresent", true, optional.isPresent());
1266                     assertEquals("Data node", peopleNode, optional.get());
1267                 }
1268             }
1269         };
1270     }
1271 }

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