opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java

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