Add unit test for FrontedMetadata memory leaks
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / AbstractDistributedDataStoreIntegrationTest.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 package org.opendaylight.controller.cluster.datastore;
9
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertFalse;
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.ArgumentMatchers.any;
16 import static org.mockito.ArgumentMatchers.eq;
17 import static org.mockito.Mockito.timeout;
18 import static org.mockito.Mockito.verify;
19
20 import akka.actor.ActorRef;
21 import akka.actor.ActorSystem;
22 import com.google.common.base.Throwables;
23 import com.google.common.collect.ImmutableMap;
24 import com.google.common.collect.Range;
25 import com.google.common.primitives.UnsignedLong;
26 import com.google.common.util.concurrent.FluentFuture;
27 import com.google.common.util.concurrent.ListenableFuture;
28 import com.google.common.util.concurrent.MoreExecutors;
29 import com.google.common.util.concurrent.Uninterruptibles;
30 import java.math.BigInteger;
31 import java.util.ArrayList;
32 import java.util.Arrays;
33 import java.util.Collection;
34 import java.util.Collections;
35 import java.util.Iterator;
36 import java.util.List;
37 import java.util.Optional;
38 import java.util.Set;
39 import java.util.concurrent.CountDownLatch;
40 import java.util.concurrent.ExecutionException;
41 import java.util.concurrent.TimeUnit;
42 import java.util.concurrent.atomic.AtomicReference;
43 import org.junit.Ignore;
44 import org.junit.Test;
45 import org.junit.runners.Parameterized.Parameter;
46 import org.mockito.Mockito;
47 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
48 import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
49 import org.opendaylight.controller.cluster.datastore.TestShard.RequestFrontendMetadata;
50 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
51 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
52 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
53 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
54 import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata;
55 import org.opendaylight.controller.cluster.datastore.persisted.FrontendShardDataTreeSnapshotMetadata;
56 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
57 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
58 import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
59 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
60 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
61 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
62 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
63 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
64 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
65 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
66 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
67 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
68 import org.opendaylight.mdsal.dom.api.DOMTransactionChainClosedException;
69 import org.opendaylight.mdsal.dom.api.DOMTransactionChainListener;
70 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
71 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
72 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
73 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
74 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
75 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
76 import org.opendaylight.yangtools.concepts.ListenerRegistration;
77 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
78 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
79 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
80 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
81 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
82 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
83 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
84 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
85 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
86 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
87
88 public abstract class AbstractDistributedDataStoreIntegrationTest {
89
90     @Parameter
91     public Class<? extends AbstractDataStore> testParameter;
92
93     protected ActorSystem system;
94
95     protected final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder()
96             .shardHeartbeatIntervalInMillis(100);
97
98     protected ActorSystem getSystem() {
99         return system;
100     }
101
102     @Test
103     public void testWriteTransactionWithSingleShard() throws Exception {
104         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
105         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
106             testParameter, "transactionIntegrationTest", "test-1")) {
107
108             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
109                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
110
111             testKit.testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
112                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
113                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
114                 .build());
115         }
116     }
117
118     @Test
119     public void testWriteTransactionWithMultipleShards() throws Exception {
120         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
121         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
122             testParameter, "testWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
123
124             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
125             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
126
127             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
128             writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
129
130             testKit.doCommit(writeTx.ready());
131
132             writeTx = dataStore.newWriteOnlyTransaction();
133
134             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
135             writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
136
137             testKit.doCommit(writeTx.ready());
138
139             writeTx = dataStore.newWriteOnlyTransaction();
140
141             final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
142             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
143             writeTx.write(carPath, car);
144
145             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
146             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
147             writeTx.write(personPath, person);
148
149             testKit.doCommit(writeTx.ready());
150
151             // Verify the data in the store
152             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
153
154             Optional<NormalizedNode<?, ?>> optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
155             assertTrue("isPresent", optional.isPresent());
156             assertEquals("Data node", car, optional.get());
157
158             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
159             assertTrue("isPresent", optional.isPresent());
160             assertEquals("Data node", person, optional.get());
161         }
162     }
163
164     @Test
165     public void testReadWriteTransactionWithSingleShard() throws Exception {
166         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
167         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
168             testParameter, "testReadWriteTransactionWithSingleShard", "test-1")) {
169
170             // 1. Create a read-write Tx
171             final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
172             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
173
174             // 2. Write some data
175             final YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
176             final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
177             readWriteTx.write(nodePath, nodeToWrite);
178
179             // 3. Read the data from Tx
180             final Boolean exists = readWriteTx.exists(nodePath).get(5, TimeUnit.SECONDS);
181             assertEquals("exists", Boolean.TRUE, exists);
182
183             Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
184             assertTrue("isPresent", optional.isPresent());
185             assertEquals("Data node", nodeToWrite, optional.get());
186
187             // 4. Ready the Tx for commit
188             final DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
189
190             // 5. Commit the Tx
191             testKit.doCommit(cohort);
192
193             // 6. Verify the data in the store
194             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
195
196             optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
197             assertTrue("isPresent", optional.isPresent());
198             assertEquals("Data node", nodeToWrite, optional.get());
199         }
200     }
201
202     @Test
203     public void testReadWriteTransactionWithMultipleShards() throws Exception {
204         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
205         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
206             testParameter, "testReadWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
207
208             DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
209             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
210
211             readWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
212             readWriteTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
213
214             testKit.doCommit(readWriteTx.ready());
215
216             readWriteTx = dataStore.newReadWriteTransaction();
217
218             readWriteTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
219             readWriteTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
220
221             testKit.doCommit(readWriteTx.ready());
222
223             readWriteTx = dataStore.newReadWriteTransaction();
224
225             final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
226             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
227             readWriteTx.write(carPath, car);
228
229             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
230             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
231             readWriteTx.write(personPath, person);
232
233             final Boolean exists = readWriteTx.exists(carPath).get(5, TimeUnit.SECONDS);
234             assertEquals("exists", Boolean.TRUE, exists);
235
236             Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
237             assertTrue("isPresent", optional.isPresent());
238             assertEquals("Data node", car, optional.get());
239
240             testKit.doCommit(readWriteTx.ready());
241
242             // Verify the data in the store
243             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
244
245             optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
246             assertTrue("isPresent", optional.isPresent());
247             assertEquals("Data node", car, optional.get());
248
249             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
250             assertTrue("isPresent", optional.isPresent());
251             assertEquals("Data node", person, optional.get());
252         }
253     }
254
255     @Test
256     @Ignore("Flushes a closed tx leak in single node, needs to be handled separately")
257     public void testSingleTransactionsWritesInQuickSuccession() throws Exception {
258         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
259         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
260             testParameter, "testSingleTransactionsWritesInQuickSuccession", "cars-1")) {
261
262             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
263
264             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
265             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
266             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
267             testKit.doCommit(writeTx.ready());
268
269             int numCars = 5;
270             for (int i = 0; i < numCars; i++) {
271                 writeTx = txChain.newWriteOnlyTransaction();
272                 writeTx.write(CarsModel.newCarPath("car" + i),
273                     CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
274
275                 testKit.doCommit(writeTx.ready());
276
277                 DOMStoreReadTransaction domStoreReadTransaction = txChain.newReadOnlyTransaction();
278                 domStoreReadTransaction.read(CarsModel.BASE_PATH).get();
279
280                 domStoreReadTransaction.close();
281             }
282
283             // verify frontend metadata has no holes in purged transactions causing overtime memory leak
284             Optional<ActorRef> localShard = dataStore.getActorUtils().findLocalShard("cars-1");
285             FrontendShardDataTreeSnapshotMetadata frontendMetadata =
286                     (FrontendShardDataTreeSnapshotMetadata) dataStore.getActorUtils()
287                             .executeOperation(localShard.get(), new RequestFrontendMetadata());
288
289             if (dataStore.getActorUtils().getDatastoreContext().isUseTellBasedProtocol()) {
290                 Iterator<FrontendHistoryMetadata> iterator =
291                         frontendMetadata.getClients().get(0).getCurrentHistories().iterator();
292                 FrontendHistoryMetadata metadata = iterator.next();
293                 while (iterator.hasNext() && metadata.getHistoryId() != 1) {
294                     metadata = iterator.next();
295                 }
296                 Set<Range<UnsignedLong>> ranges = metadata.getPurgedTransactions().asRanges();
297
298                 assertEquals(1, ranges.size());
299             } else {
300                 // ask based should track no metadata
301                 assertTrue(frontendMetadata.getClients().get(0).getCurrentHistories().isEmpty());
302             }
303
304             final Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction()
305                     .read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
306             assertTrue("isPresent", optional.isPresent());
307             assertEquals("# cars", numCars, ((Collection<?>) optional.get().getValue()).size());
308         }
309     }
310
311     @SuppressWarnings("checkstyle:IllegalCatch")
312     private void testTransactionCommitFailureWithNoShardLeader(final boolean writeOnly, final String testName)
313             throws Exception {
314         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
315         final String shardName = "default";
316
317         // We don't want the shard to become the leader so prevent shard
318         // elections.
319         datastoreContextBuilder.customRaftPolicyImplementation(
320                 "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
321
322         // The ShardManager uses the election timeout for FindPrimary so
323         // reset it low so it will timeout quickly.
324         datastoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1)
325         .shardInitializationTimeout(200, TimeUnit.MILLISECONDS).frontendRequestTimeoutInSeconds(2);
326
327         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(testParameter, testName, false, shardName)) {
328
329             final Object result = dataStore.getActorUtils().executeOperation(
330                 dataStore.getActorUtils().getShardManager(), new FindLocalShard(shardName, true));
331             assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
332
333             // Create the write Tx.
334             DOMStoreWriteTransaction writeTxToClose = null;
335             try {
336                 writeTxToClose = writeOnly ? dataStore.newWriteOnlyTransaction()
337                         : dataStore.newReadWriteTransaction();
338                 final DOMStoreWriteTransaction writeTx = writeTxToClose;
339                 assertNotNull("newReadWriteTransaction returned null", writeTx);
340
341                 // Do some modifications and ready the Tx on a separate
342                 // thread.
343                 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
344                 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
345                 final CountDownLatch txReady = new CountDownLatch(1);
346                 final Thread txThread = new Thread(() -> {
347                     try {
348                         writeTx.write(TestModel.JUNK_PATH,
349                             ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
350
351                         txCohort.set(writeTx.ready());
352                     } catch (Exception e) {
353                         caughtEx.set(e);
354                     } finally {
355                         txReady.countDown();
356                     }
357                 });
358
359                 txThread.start();
360
361                 // Wait for the Tx operations to complete.
362                 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
363                 if (caughtEx.get() != null) {
364                     throw caughtEx.get();
365                 }
366
367                 assertTrue("Tx ready", done);
368
369                 // Wait for the commit to complete. Since no shard
370                 // leader was elected in time, the Tx
371                 // should have timed out and throw an appropriate
372                 // exception cause.
373                 try {
374                     txCohort.get().canCommit().get(10, TimeUnit.SECONDS);
375                     fail("Expected NoShardLeaderException");
376                 } catch (final ExecutionException e) {
377                     final String msg = "Unexpected exception: "
378                             + Throwables.getStackTraceAsString(e.getCause());
379                     if (DistributedDataStore.class.isAssignableFrom(testParameter)) {
380                         assertTrue(Throwables.getRootCause(e) instanceof NoShardLeaderException);
381                     } else {
382                         assertTrue(msg, Throwables.getRootCause(e) instanceof RequestTimeoutException);
383                     }
384                 }
385             } finally {
386                 try {
387                     if (writeTxToClose != null) {
388                         writeTxToClose.close();
389                     }
390                 } catch (Exception e) {
391                     // FIXME TransactionProxy.close throws IllegalStateException:
392                     // Transaction is ready, it cannot be closed
393                 }
394             }
395         }
396     }
397
398     @Test
399     public void testWriteOnlyTransactionCommitFailureWithNoShardLeader() throws Exception {
400         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
401         testTransactionCommitFailureWithNoShardLeader(true, "testWriteOnlyTransactionCommitFailureWithNoShardLeader");
402     }
403
404     @Test
405     public void testReadWriteTransactionCommitFailureWithNoShardLeader() throws Exception {
406         testTransactionCommitFailureWithNoShardLeader(false, "testReadWriteTransactionCommitFailureWithNoShardLeader");
407     }
408
409     @Test
410     public void testTransactionAbort() throws Exception {
411         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
412         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
413             testParameter, "transactionAbortIntegrationTest", "test-1")) {
414
415             final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
416             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
417
418             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
419
420             final DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
421
422             cohort.canCommit().get(5, TimeUnit.SECONDS);
423
424             cohort.abort().get(5, TimeUnit.SECONDS);
425
426             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
427                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
428         }
429     }
430
431     @Test
432     @SuppressWarnings("checkstyle:IllegalCatch")
433     public void testTransactionChainWithSingleShard() throws Exception {
434         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
435         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
436             testParameter, "testTransactionChainWithSingleShard", "test-1")) {
437
438             // 1. Create a Tx chain and write-only Tx
439             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
440
441             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
442             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
443
444             // 2. Write some data
445             final NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
446             writeTx.write(TestModel.TEST_PATH, testNode);
447
448             // 3. Ready the Tx for commit
449             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
450
451             // 4. Commit the Tx on another thread that first waits for
452             // the second read Tx.
453             final CountDownLatch continueCommit1 = new CountDownLatch(1);
454             final CountDownLatch commit1Done = new CountDownLatch(1);
455             final AtomicReference<Exception> commit1Error = new AtomicReference<>();
456             new Thread(() -> {
457                 try {
458                     continueCommit1.await();
459                     testKit.doCommit(cohort1);
460                 } catch (Exception e) {
461                     commit1Error.set(e);
462                 } finally {
463                     commit1Done.countDown();
464                 }
465             }).start();
466
467             // 5. Create a new read Tx from the chain to read and verify
468             // the data from the first
469             // Tx is visible after being readied.
470             DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
471             Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
472             assertTrue("isPresent", optional.isPresent());
473             assertEquals("Data node", testNode, optional.get());
474
475             // 6. Create a new RW Tx from the chain, write more data,
476             // and ready it
477             final DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
478             final MapNode outerNode = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
479                     .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
480                     .build();
481             rwTx.write(TestModel.OUTER_LIST_PATH, outerNode);
482
483             final DOMStoreThreePhaseCommitCohort cohort2 = rwTx.ready();
484
485             // 7. Create a new read Tx from the chain to read the data
486             // from the last RW Tx to
487             // verify it is visible.
488             readTx = txChain.newReadWriteTransaction();
489             optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
490             assertTrue("isPresent", optional.isPresent());
491             assertEquals("Data node", outerNode, optional.get());
492
493             // 8. Wait for the 2 commits to complete and close the
494             // chain.
495             continueCommit1.countDown();
496             Uninterruptibles.awaitUninterruptibly(commit1Done, 5, TimeUnit.SECONDS);
497
498             if (commit1Error.get() != null) {
499                 throw commit1Error.get();
500             }
501
502             testKit.doCommit(cohort2);
503
504             txChain.close();
505
506             // 9. Create a new read Tx from the data store and verify
507             // committed data.
508             readTx = dataStore.newReadOnlyTransaction();
509             optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
510             assertTrue("isPresent", optional.isPresent());
511             assertEquals("Data node", outerNode, optional.get());
512         }
513     }
514
515     @Test
516     public void testTransactionChainWithMultipleShards() throws Exception {
517         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
518         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
519             testParameter, "testTransactionChainWithMultipleShards", "cars-1", "people-1")) {
520
521             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
522
523             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
524             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
525
526             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
527             writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
528
529             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
530             writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
531
532             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
533
534             final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
535
536             final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
537             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
538             readWriteTx.write(carPath, car);
539
540             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
541             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
542             readWriteTx.merge(personPath, person);
543
544             Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
545             assertTrue("isPresent", optional.isPresent());
546             assertEquals("Data node", car, optional.get());
547
548             optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
549             assertTrue("isPresent", optional.isPresent());
550             assertEquals("Data node", person, optional.get());
551
552             final DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
553
554             writeTx = txChain.newWriteOnlyTransaction();
555
556             writeTx.delete(carPath);
557
558             final DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
559
560             final ListenableFuture<Boolean> canCommit1 = cohort1.canCommit();
561             final ListenableFuture<Boolean> canCommit2 = cohort2.canCommit();
562
563             testKit.doCommit(canCommit1, cohort1);
564             testKit.doCommit(canCommit2, cohort2);
565             testKit.doCommit(cohort3);
566
567             txChain.close();
568
569             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
570
571             optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
572             assertFalse("isPresent", optional.isPresent());
573
574             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
575             assertTrue("isPresent", optional.isPresent());
576             assertEquals("Data node", person, optional.get());
577         }
578     }
579
580     @Test
581     public void testCreateChainedTransactionsInQuickSuccession() throws Exception {
582         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
583         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
584             testParameter, "testCreateChainedTransactionsInQuickSuccession", "cars-1")) {
585
586             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
587                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
588                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
589                 MoreExecutors.directExecutor());
590
591             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
592             DOMTransactionChain txChain = broker.createTransactionChain(listener);
593
594             final List<ListenableFuture<?>> futures = new ArrayList<>();
595
596             final DOMDataTreeWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
597             writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, CarsModel.emptyContainer());
598             writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
599             futures.add(writeTx.commit());
600
601             int numCars = 100;
602             for (int i = 0; i < numCars; i++) {
603                 final DOMDataTreeReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
604
605                 rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i),
606                     CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
607
608                 futures.add(rwTx.commit());
609             }
610
611             for (final ListenableFuture<?> f : futures) {
612                 f.get(5, TimeUnit.SECONDS);
613             }
614
615             final Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction()
616                     .read(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
617             assertTrue("isPresent", optional.isPresent());
618             assertEquals("# cars", numCars, ((Collection<?>) optional.get().getValue()).size());
619
620             txChain.close();
621
622             broker.close();
623         }
624     }
625
626     @Test
627     public void testCreateChainedTransactionAfterEmptyTxReadied() throws Exception {
628         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
629         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
630             testParameter, "testCreateChainedTransactionAfterEmptyTxReadied", "test-1")) {
631
632             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
633
634             final DOMStoreReadWriteTransaction rwTx1 = txChain.newReadWriteTransaction();
635
636             rwTx1.ready();
637
638             final DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction();
639
640             final Optional<NormalizedNode<?, ?>> optional = rwTx2.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
641             assertFalse("isPresent", optional.isPresent());
642
643             txChain.close();
644         }
645     }
646
647     @Test
648     public void testCreateChainedTransactionWhenPreviousNotReady() throws Exception {
649         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
650         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
651             testParameter, "testCreateChainedTransactionWhenPreviousNotReady", "test-1")) {
652
653             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
654
655             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
656             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
657
658             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
659
660             // Try to create another Tx of each type - each should fail
661             // b/c the previous Tx wasn't
662             // readied.
663             testKit.assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
664         }
665     }
666
667     @Test
668     public void testCreateChainedTransactionAfterClose() throws Exception {
669         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
670         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
671             testParameter, "testCreateChainedTransactionAfterClose", "test-1")) {
672
673             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
674             txChain.close();
675
676             // Try to create another Tx of each type - should fail b/c
677             // the previous Tx was closed.
678             testKit.assertExceptionOnTxChainCreates(txChain, DOMTransactionChainClosedException.class);
679         }
680     }
681
682     @Test
683     public void testChainWithReadOnlyTxAfterPreviousReady() throws Exception {
684         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
685         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
686             testParameter, "testChainWithReadOnlyTxAfterPreviousReady", "test-1")) {
687
688             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
689
690             // Create a write tx and submit.
691             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
692             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
693             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
694
695             // Create read-only tx's and issue a read.
696             FluentFuture<Optional<NormalizedNode<?, ?>>> readFuture1 = txChain
697                     .newReadOnlyTransaction().read(TestModel.TEST_PATH);
698
699             FluentFuture<Optional<NormalizedNode<?, ?>>> readFuture2 = txChain
700                     .newReadOnlyTransaction().read(TestModel.TEST_PATH);
701
702             // Create another write tx and issue the write.
703             DOMStoreWriteTransaction writeTx2 = txChain.newWriteOnlyTransaction();
704             writeTx2.write(TestModel.OUTER_LIST_PATH,
705                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
706                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
707                 .build());
708
709             // Ensure the reads succeed.
710
711             assertTrue("isPresent", readFuture1.get(5, TimeUnit.SECONDS).isPresent());
712             assertTrue("isPresent", readFuture2.get(5, TimeUnit.SECONDS).isPresent());
713
714             // Ensure the writes succeed.
715             DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
716
717             testKit.doCommit(cohort1);
718             testKit.doCommit(cohort2);
719
720             assertTrue("isPresent", txChain.newReadOnlyTransaction().read(TestModel.OUTER_LIST_PATH)
721                 .get(5, TimeUnit.SECONDS).isPresent());
722         }
723     }
724
725     @Test
726     public void testChainedTransactionFailureWithSingleShard() throws Exception {
727         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
728         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
729             testParameter, "testChainedTransactionFailureWithSingleShard", "cars-1")) {
730
731             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
732                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
733                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
734                 MoreExecutors.directExecutor());
735
736             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
737             final DOMTransactionChain txChain = broker.createTransactionChain(listener);
738
739             final DOMDataTreeReadWriteTransaction writeTx = txChain.newReadWriteTransaction();
740
741             writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
742                 PeopleModel.emptyContainer());
743
744             final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
745                     .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
746                     .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
747
748             writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
749
750             try {
751                 writeTx.commit().get(5, TimeUnit.SECONDS);
752                 fail("Expected TransactionCommitFailedException");
753             } catch (final ExecutionException e) {
754                 // Expected
755             }
756
757             verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
758                 any(Throwable.class));
759
760             txChain.close();
761             broker.close();
762         }
763     }
764
765     @Test
766     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
767         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
768         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
769             testParameter, "testChainedTransactionFailureWithMultipleShards", "cars-1", "people-1")) {
770
771             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
772                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
773                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
774                 MoreExecutors.directExecutor());
775
776             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
777             final DOMTransactionChain txChain = broker.createTransactionChain(listener);
778
779             final DOMDataTreeWriteTransaction writeTx = txChain.newReadWriteTransaction();
780
781             writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
782                 PeopleModel.emptyContainer());
783
784             final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
785                     .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
786                     .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
787
788             writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
789
790             // Note that merge will validate the data and fail but put
791             // succeeds b/c deep validation is not
792             // done for put for performance reasons.
793             try {
794                 writeTx.commit().get(5, TimeUnit.SECONDS);
795                 fail("Expected TransactionCommitFailedException");
796             } catch (final ExecutionException e) {
797                 // Expected
798             }
799
800             verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
801                 any(Throwable.class));
802
803             txChain.close();
804             broker.close();
805         }
806     }
807
808     @Test
809     public void testDataTreeChangeListenerRegistration() throws Exception {
810         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
811         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
812             testParameter, "testDataTreeChangeListenerRegistration", "test-1")) {
813
814             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
815                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
816
817             final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
818
819             ListenerRegistration<MockDataTreeChangeListener> listenerReg = dataStore
820                     .registerTreeChangeListener(TestModel.TEST_PATH, listener);
821
822             assertNotNull("registerTreeChangeListener returned null", listenerReg);
823
824             IntegrationTestKit.verifyShardState(dataStore, "test-1",
825                 state -> assertEquals("getTreeChangeListenerActors", 1,
826                         state.getTreeChangeListenerActors().size()));
827
828             // Wait for the initial notification
829             listener.waitForChangeEvents(TestModel.TEST_PATH);
830             listener.reset(2);
831
832             // Write 2 updates.
833             testKit.testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
834                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
835                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
836                 .build());
837
838             YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
839                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
840             testKit.testWriteTransaction(dataStore, listPath,
841                 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
842
843             // Wait for the 2 updates.
844             listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
845             listenerReg.close();
846
847             IntegrationTestKit.verifyShardState(dataStore, "test-1",
848                 state -> assertEquals("getTreeChangeListenerActors", 0,
849                     state.getTreeChangeListenerActors().size()));
850
851             testKit.testWriteTransaction(dataStore,
852                 YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
853                 .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
854                 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
855
856             listener.expectNoMoreChanges("Received unexpected change after close");
857         }
858     }
859
860     @Test
861     public void testRestoreFromDatastoreSnapshot() throws Exception {
862         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
863         final String name = "transactionIntegrationTest";
864
865         final ContainerNode carsNode = CarsModel.newCarsNode(
866             CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)),
867                 CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L))));
868
869         DataTree dataTree = new InMemoryDataTreeFactory().create(
870             DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full());
871         AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
872         NormalizedNode<?, ?> root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
873
874         final Snapshot carsSnapshot = Snapshot.create(
875             new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
876             Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
877
878         dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL,
879             SchemaContextHelper.full());
880
881         final NormalizedNode<?, ?> peopleNode = PeopleModel.create();
882         AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode);
883
884         root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
885
886         final Snapshot peopleSnapshot = Snapshot.create(
887             new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
888             Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
889
890         testKit.restoreFromSnapshot = new DatastoreSnapshot(name, null, Arrays.asList(
891             new DatastoreSnapshot.ShardSnapshot("cars", carsSnapshot),
892             new DatastoreSnapshot.ShardSnapshot("people", peopleSnapshot)));
893
894         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
895             testParameter, name, "module-shards-member1.conf", true, "cars", "people")) {
896
897             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
898
899             // two reads
900             Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
901             assertTrue("isPresent", optional.isPresent());
902             assertEquals("Data node", carsNode, optional.get());
903
904             optional = readTx.read(PeopleModel.BASE_PATH).get(5, TimeUnit.SECONDS);
905             assertTrue("isPresent", optional.isPresent());
906             assertEquals("Data node", peopleNode, optional.get());
907         }
908     }
909 }

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