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