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