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

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