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

   1 /*
   2  * Copyright (c) 2014, 2015 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.ByteArrayOutputStream;
  34 import java.io.DataOutputStream;
  35 import java.io.IOException;
  36 import java.io.ObjectOutputStream;
  37 import java.math.BigInteger;
  38 import java.util.ArrayList;
  39 import java.util.Arrays;
  40 import java.util.Collection;
  41 import java.util.Collections;
  42 import java.util.List;
  43 import java.util.concurrent.CountDownLatch;
  44 import java.util.concurrent.ExecutionException;
  45 import java.util.concurrent.TimeUnit;
  46 import java.util.concurrent.atomic.AtomicReference;
  47 import org.junit.After;
  48 import org.junit.Before;
  49 import org.junit.Test;
  50 import org.junit.runner.RunWith;
  51 import org.junit.runners.Parameterized;
  52 import org.junit.runners.Parameterized.Parameter;
  53 import org.junit.runners.Parameterized.Parameters;
  54 import org.mockito.Mockito;
  55 import org.opendaylight.controller.cluster.databroker.ClientBackedDataStore;
  56 import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
  57 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  58 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  59 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  60 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  61 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
  62 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
  63 import org.opendaylight.controller.cluster.datastore.persisted.PayloadVersion;
  64 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
  65 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
  66 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  67 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  68 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  69 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  70 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  71 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  72 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  73 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  74 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  75 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  76 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  77 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  78 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  79 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  80 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  81 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
  82 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  83 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  84 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  85 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  86 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  87 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  88 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  89 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  90 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  91 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  92 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  93 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  94 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  95 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
  96 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  97 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  98 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  99
 100 @RunWith(Parameterized.class)
 101 public class DistributedDataStoreIntegrationTest {
 102
 103     @Parameters(name = "{0}")
 104     public static Collection<Object[]> data() {
 105         return Arrays.asList(new Object[][] {
 106                 { DistributedDataStore.class }, { ClientBackedDataStore.class }
 107         });
 108     }
 109
 110     @Parameter
 111     public Class<? extends AbstractDataStore> testParameter;
 112
 113     private ActorSystem system;
 114
 115     private final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder()
 116             .shardHeartbeatIntervalInMillis(100);
 117
 118     @Before
 119     public void setUp() throws IOException {
 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 (final 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 (final 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 (final 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 (final 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 (final 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 (final 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                         @Override
 365                         public void run() {
 366                             try {
 367                                 writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 368
 369                                 writeTx.merge(TestModel.OUTER_LIST_PATH,
 370                                         ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
 371
 372                                 writeTx.write(listEntryPath,
 373                                         ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
 374
 375                                 writeTx.delete(listEntryPath);
 376
 377                                 txCohort.set(writeTx.ready());
 378                             } catch (Exception e) {
 379                                 caughtEx.set(e);
 380                             } finally {
 381                                 txReady.countDown();
 382                             }
 383                         }
 384                     };
 385
 386                     txThread.start();
 387
 388                     // Wait for the Tx operations to complete.
 389                     final boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 390                     if (caughtEx.get() != null) {
 391                         throw caughtEx.get();
 392                     }
 393
 394                     assertEquals("Tx ready", true, done);
 395
 396                     // At this point the Tx operations should be waiting for the
 397                     // shard to initialize so
 398                     // trigger the latch to let the shard recovery to continue.
 399                     blockRecoveryLatch.countDown();
 400
 401                     // Wait for the Tx commit to complete.
 402                     doCommit(txCohort.get());
 403
 404                     // Verify the data in the store
 405                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 406
 407                     Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 408                     assertEquals("isPresent", true, optional.isPresent());
 409
 410                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 411                     assertEquals("isPresent", true, optional.isPresent());
 412
 413                     optional = readTx.read(listEntryPath).get(5, TimeUnit.SECONDS);
 414                     assertEquals("isPresent", false, optional.isPresent());
 415                 }
 416             }
 417         };
 418     }
 419
 420     @Test
 421     public void testWriteOnlyTransactionWithShardNotInitiallyReady() throws Exception {
 422         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 423         testTransactionWritesWithShardNotInitiallyReady("testWriteOnlyTransactionWithShardNotInitiallyReady", true);
 424     }
 425
 426     @Test
 427     public void testReadWriteTransactionWithShardNotInitiallyReady() throws Exception {
 428         testTransactionWritesWithShardNotInitiallyReady("testReadWriteTransactionWithShardNotInitiallyReady", false);
 429     }
 430
 431     @Test
 432     @SuppressWarnings("checkstyle:IllegalCatch")
 433     public void testTransactionReadsWithShardNotInitiallyReady() throws Exception {
 434         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 435             {
 436                 final String testName = "testTransactionReadsWithShardNotInitiallyReady";
 437                 final String shardName = "test-1";
 438
 439                 // Setup the InMemoryJournal to block shard recovery to ensure
 440                 // the shard isn't
 441                 // initialized until we create the Tx.
 442                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 443                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 444                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 445
 446                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 447                         testParameter, testName, false, shardName)) {
 448
 449                     // Create the read-write Tx
 450                     final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 451                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 452
 453                     // Do some reads on the Tx on a separate thread.
 454                     final AtomicReference<CheckedFuture<Boolean, ReadFailedException>> txExistsFuture =
 455                             new AtomicReference<>();
 456                     final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
 457                             txReadFuture = new AtomicReference<>();
 458                     final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 459                     final CountDownLatch txReadsDone = new CountDownLatch(1);
 460                     final Thread txThread = new Thread() {
 461                         @Override
 462                         public void run() {
 463                             try {
 464                                 readWriteTx.write(TestModel.TEST_PATH,
 465                                         ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 466
 467                                 txExistsFuture.set(readWriteTx.exists(TestModel.TEST_PATH));
 468
 469                                 txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
 470                             } catch (Exception e) {
 471                                 caughtEx.set(e);
 472                             } finally {
 473                                 txReadsDone.countDown();
 474                             }
 475                         }
 476                     };
 477
 478                     txThread.start();
 479
 480                     // Wait for the Tx operations to complete.
 481                     boolean done = Uninterruptibles.awaitUninterruptibly(txReadsDone, 5, TimeUnit.SECONDS);
 482                     if (caughtEx.get() != null) {
 483                         throw caughtEx.get();
 484                     }
 485
 486                     assertEquals("Tx reads done", true, done);
 487
 488                     // At this point the Tx operations should be waiting for the
 489                     // shard to initialize so
 490                     // trigger the latch to let the shard recovery to continue.
 491                     blockRecoveryLatch.countDown();
 492
 493                     // Wait for the reads to complete and verify.
 494                     assertEquals("exists", true, txExistsFuture.get().checkedGet(5, TimeUnit.SECONDS));
 495                     assertEquals("read", true, txReadFuture.get().checkedGet(5, TimeUnit.SECONDS).isPresent());
 496
 497                     readWriteTx.close();
 498                 }
 499             }
 500         };
 501     }
 502
 503     @Test(expected = NotInitializedException.class)
 504     @SuppressWarnings("checkstyle:IllegalCatch")
 505     public void testTransactionCommitFailureWithShardNotInitialized() throws Exception {
 506         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 507             {
 508                 final String testName = "testTransactionCommitFailureWithShardNotInitialized";
 509                 final String shardName = "test-1";
 510
 511                 // Set the shard initialization timeout low for the test.
 512                 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
 513
 514                 // Setup the InMemoryJournal to block shard recovery
 515                 // indefinitely.
 516                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 517                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 518                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 519
 520                 InMemoryJournal.addEntry(persistentID, 1, "Dummy data so akka will read from persistence");
 521
 522                 final AbstractDataStore dataStore =
 523                         setupAbstractDataStore(testParameter, testName, false, shardName);
 524
 525                 // Create the write Tx
 526                 final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 527                 assertNotNull("newReadWriteTransaction returned null", writeTx);
 528
 529                 // Do some modifications and ready the Tx on a separate
 530                 // thread.
 531                 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 532                 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 533                 final CountDownLatch txReady = new CountDownLatch(1);
 534                 final Thread txThread = new Thread(() -> {
 535                     try {
 536                         writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 537
 538                         txCohort.set(writeTx.ready());
 539                     } catch (Exception e) {
 540                         caughtEx.set(e);
 541                     } finally {
 542                         txReady.countDown();
 543                     }
 544                 });
 545
 546                 txThread.start();
 547
 548                 // Wait for the Tx operations to complete.
 549                 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 550                 if (caughtEx.get() != null) {
 551                     throw caughtEx.get();
 552                 }
 553
 554                 assertEquals("Tx ready", true, done);
 555
 556                 // Wait for the commit to complete. Since the shard never
 557                 // initialized, the Tx should
 558                 // have timed out and throw an appropriate exception cause.
 559                 try {
 560                     txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
 561                     fail("Expected NotInitializedException");
 562                 } catch (final Exception e) {
 563                     Throwables.propagate(Throwables.getRootCause(e));
 564                 } finally {
 565                     blockRecoveryLatch.countDown();
 566                 }
 567             }
 568         };
 569     }
 570
 571     @Test(expected = NotInitializedException.class)
 572     @SuppressWarnings("checkstyle:IllegalCatch")
 573     public void testTransactionReadFailureWithShardNotInitialized() throws Exception {
 574         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 575             {
 576                 final String testName = "testTransactionReadFailureWithShardNotInitialized";
 577                 final String shardName = "test-1";
 578
 579                 // Set the shard initialization timeout low for the test.
 580                 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
 581
 582                 // Setup the InMemoryJournal to block shard recovery
 583                 // indefinitely.
 584                 final String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 585                 final CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 586                 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 587
 588                 InMemoryJournal.addEntry(persistentID, 1, "Dummy data so akka will read from persistence");
 589
 590                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 591                         testParameter, testName, false, shardName)) {
 592
 593                     // Create the read-write Tx
 594                     final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 595                     assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 596
 597                     // Do a read on the Tx on a separate thread.
 598                     final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
 599                             txReadFuture = new AtomicReference<>();
 600                     final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 601                     final CountDownLatch txReadDone = new CountDownLatch(1);
 602                     final Thread txThread = new Thread(() -> {
 603                         try {
 604                             readWriteTx.write(TestModel.TEST_PATH,
 605                                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 606
 607                             txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
 608
 609                             readWriteTx.close();
 610                         } catch (Exception e) {
 611                             caughtEx.set(e);
 612                         } finally {
 613                             txReadDone.countDown();
 614                         }
 615                     });
 616
 617                     txThread.start();
 618
 619                     // Wait for the Tx operations to complete.
 620                     boolean done = Uninterruptibles.awaitUninterruptibly(txReadDone, 5, TimeUnit.SECONDS);
 621                     if (caughtEx.get() != null) {
 622                         throw caughtEx.get();
 623                     }
 624
 625                     assertEquals("Tx read done", true, done);
 626
 627                     // Wait for the read to complete. Since the shard never
 628                     // initialized, the Tx should
 629                     // have timed out and throw an appropriate exception cause.
 630                     try {
 631                         txReadFuture.get().checkedGet(5, TimeUnit.SECONDS);
 632                         fail("Expected NotInitializedException");
 633                     } catch (final ReadFailedException e) {
 634                         Throwables.propagate(Throwables.getRootCause(e));
 635                     } finally {
 636                         blockRecoveryLatch.countDown();
 637                     }
 638                 }
 639             }
 640         };
 641     }
 642
 643     @SuppressWarnings("checkstyle:IllegalCatch")
 644     private void testTransactionCommitFailureWithNoShardLeader(final boolean writeOnly, final String testName)
 645             throws Exception {
 646         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 647             {
 648                 final String shardName = "default";
 649
 650                 // We don't want the shard to become the leader so prevent shard
 651                 // elections.
 652                 datastoreContextBuilder.customRaftPolicyImplementation(
 653                         "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
 654
 655                 // The ShardManager uses the election timeout for FindPrimary so
 656                 // reset it low so it will timeout quickly.
 657                 datastoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1)
 658                         .shardInitializationTimeout(200, TimeUnit.MILLISECONDS);
 659
 660                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 661                         testParameter, testName, false, shardName)) {
 662
 663                     final Object result = dataStore.getActorContext().executeOperation(
 664                             dataStore.getActorContext().getShardManager(), new FindLocalShard(shardName, true));
 665                     assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
 666
 667                     // Create the write Tx.
 668                     try (final DOMStoreWriteTransaction writeTx = writeOnly ? dataStore.newWriteOnlyTransaction()
 669                             : dataStore.newReadWriteTransaction()) {
 670                         assertNotNull("newReadWriteTransaction returned null", writeTx);
 671
 672                         // Do some modifications and ready the Tx on a separate
 673                         // thread.
 674                         final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 675                         final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 676                         final CountDownLatch txReady = new CountDownLatch(1);
 677                         final Thread txThread = new Thread(() -> {
 678                             try {
 679                                 writeTx.write(TestModel.JUNK_PATH,
 680                                         ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 681
 682                                 txCohort.set(writeTx.ready());
 683                             } catch (Exception e) {
 684                                 caughtEx.set(e);
 685                             } finally {
 686                                 txReady.countDown();
 687                             }
 688                         });
 689
 690                         txThread.start();
 691
 692                         // Wait for the Tx operations to complete.
 693                         boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 694                         if (caughtEx.get() != null) {
 695                             throw caughtEx.get();
 696                         }
 697
 698                         assertEquals("Tx ready", true, done);
 699
 700                         // Wait for the commit to complete. Since no shard
 701                         // leader was elected in time, the Tx
 702                         // should have timed out and throw an appropriate
 703                         // exception cause.
 704                         try {
 705                             txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
 706                             fail("Expected NoShardLeaderException");
 707                         } catch (final ExecutionException e) {
 708                             Throwables.propagate(Throwables.getRootCause(e));
 709                         }
 710                     }
 711                 }
 712             }
 713         };
 714     }
 715
 716     @Test(expected = NoShardLeaderException.class)
 717     public void testWriteOnlyTransactionCommitFailureWithNoShardLeader() throws Exception {
 718         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 719         testTransactionCommitFailureWithNoShardLeader(true, "testWriteOnlyTransactionCommitFailureWithNoShardLeader");
 720     }
 721
 722     @Test(expected = NoShardLeaderException.class)
 723     public void testReadWriteTransactionCommitFailureWithNoShardLeader() throws Exception {
 724         testTransactionCommitFailureWithNoShardLeader(false, "testReadWriteTransactionCommitFailureWithNoShardLeader");
 725     }
 726
 727     @Test
 728     public void testTransactionAbort() throws Exception {
 729         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 730             {
 731                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 732                         testParameter, "transactionAbortIntegrationTest", "test-1")) {
 733
 734                     final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 735                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 736
 737                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 738
 739                     final DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 740
 741                     cohort.canCommit().get(5, TimeUnit.SECONDS);
 742
 743                     cohort.abort().get(5, TimeUnit.SECONDS);
 744
 745                     testWriteTransaction(dataStore, TestModel.TEST_PATH,
 746                             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 747                 }
 748             }
 749         };
 750     }
 751
 752     @Test
 753     @SuppressWarnings("checkstyle:IllegalCatch")
 754     public void testTransactionChainWithSingleShard() throws Exception {
 755         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 756             {
 757                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 758                         testParameter, "testTransactionChainWithSingleShard", "test-1")) {
 759
 760                     // 1. Create a Tx chain and write-only Tx
 761                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 762
 763                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 764                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 765
 766                     // 2. Write some data
 767                     final NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 768                     writeTx.write(TestModel.TEST_PATH, testNode);
 769
 770                     // 3. Ready the Tx for commit
 771                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 772
 773                     // 4. Commit the Tx on another thread that first waits for
 774                     // the second read Tx.
 775                     final CountDownLatch continueCommit1 = new CountDownLatch(1);
 776                     final CountDownLatch commit1Done = new CountDownLatch(1);
 777                     final AtomicReference<Exception> commit1Error = new AtomicReference<>();
 778                     new Thread(() -> {
 779                         try {
 780                             continueCommit1.await();
 781                             doCommit(cohort1);
 782                         } catch (Exception e) {
 783                             commit1Error.set(e);
 784                         } finally {
 785                             commit1Done.countDown();
 786                         }
 787                     }).start();
 788
 789                     // 5. Create a new read Tx from the chain to read and verify
 790                     // the data from the first
 791                     // Tx is visible after being readied.
 792                     DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
 793                     Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 794                     assertEquals("isPresent", true, optional.isPresent());
 795                     assertEquals("Data node", testNode, optional.get());
 796
 797                     // 6. Create a new RW Tx from the chain, write more data,
 798                     // and ready it
 799                     final DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 800                     final MapNode outerNode = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build();
 801                     rwTx.write(TestModel.OUTER_LIST_PATH, outerNode);
 802
 803                     final DOMStoreThreePhaseCommitCohort cohort2 = rwTx.ready();
 804
 805                     // 7. Create a new read Tx from the chain to read the data
 806                     // from the last RW Tx to
 807                     // verify it is visible.
 808                     readTx = txChain.newReadWriteTransaction();
 809                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 810                     assertEquals("isPresent", true, optional.isPresent());
 811                     assertEquals("Data node", outerNode, optional.get());
 812
 813                     // 8. Wait for the 2 commits to complete and close the
 814                     // chain.
 815                     continueCommit1.countDown();
 816                     Uninterruptibles.awaitUninterruptibly(commit1Done, 5, TimeUnit.SECONDS);
 817
 818                     if (commit1Error.get() != null) {
 819                         throw commit1Error.get();
 820                     }
 821
 822                     doCommit(cohort2);
 823
 824                     txChain.close();
 825
 826                     // 9. Create a new read Tx from the data store and verify
 827                     // committed data.
 828                     readTx = dataStore.newReadOnlyTransaction();
 829                     optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 830                     assertEquals("isPresent", true, optional.isPresent());
 831                     assertEquals("Data node", outerNode, optional.get());
 832                 }
 833             }
 834         };
 835     }
 836
 837     @Test
 838     public void testTransactionChainWithMultipleShards() throws Exception {
 839         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 840             {
 841                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 842                         testParameter, "testTransactionChainWithMultipleShards", "cars-1", "people-1")) {
 843
 844                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 845
 846                     DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 847                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 848
 849                     writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 850                     writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 851
 852                     writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 853                     writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 854
 855                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 856
 857                     final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
 858
 859                     final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
 860                     final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 861                     readWriteTx.write(carPath, car);
 862
 863                     final MapEntryNode person = PeopleModel.newPersonEntry("jack");
 864                     final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 865                     readWriteTx.merge(personPath, person);
 866
 867                     Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
 868                     assertEquals("isPresent", true, optional.isPresent());
 869                     assertEquals("Data node", car, optional.get());
 870
 871                     optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
 872                     assertEquals("isPresent", true, optional.isPresent());
 873                     assertEquals("Data node", person, optional.get());
 874
 875                     final DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
 876
 877                     writeTx = txChain.newWriteOnlyTransaction();
 878
 879                     writeTx.delete(carPath);
 880
 881                     final DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
 882
 883                     final ListenableFuture<Boolean> canCommit1 = cohort1.canCommit();
 884                     final ListenableFuture<Boolean> canCommit2 = cohort2.canCommit();
 885
 886                     doCommit(canCommit1, cohort1);
 887                     doCommit(canCommit2, cohort2);
 888                     doCommit(cohort3);
 889
 890                     txChain.close();
 891
 892                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 893
 894                     optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
 895                     assertEquals("isPresent", false, optional.isPresent());
 896
 897                     optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
 898                     assertEquals("isPresent", true, optional.isPresent());
 899                     assertEquals("Data node", person, optional.get());
 900                 }
 901             }
 902         };
 903     }
 904
 905     @Test
 906     public void testCreateChainedTransactionsInQuickSuccession() throws Exception {
 907         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 908             {
 909                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 910                         testParameter, "testCreateChainedTransactionsInQuickSuccession", "cars-1")) {
 911
 912                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 913                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
 914                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
 915                             MoreExecutors.directExecutor());
 916
 917                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
 918                     DOMTransactionChain txChain = broker.createTransactionChain(listener);
 919
 920                     final List<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
 921
 922                     final DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 923                     writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, CarsModel.emptyContainer());
 924                     writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 925                     futures.add(writeTx.submit());
 926
 927                     int numCars = 100;
 928                     for (int i = 0; i < numCars; i++) {
 929                         final DOMDataReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 930
 931                         rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i),
 932                                 CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
 933
 934                         futures.add(rwTx.submit());
 935                     }
 936
 937                     for (final CheckedFuture<Void, TransactionCommitFailedException> f : futures) {
 938                         f.checkedGet();
 939                     }
 940
 941                     final Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction()
 942                             .read(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
 943                     assertEquals("isPresent", true, optional.isPresent());
 944                     assertEquals("# cars", numCars, ((Collection<?>) optional.get().getValue()).size());
 945
 946                     txChain.close();
 947
 948                     broker.close();
 949                 }
 950             }
 951         };
 952     }
 953
 954     @Test
 955     public void testCreateChainedTransactionAfterEmptyTxReadied() throws Exception {
 956         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 957             {
 958                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 959                         testParameter, "testCreateChainedTransactionAfterEmptyTxReadied", "test-1")) {
 960
 961                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 962
 963                     final DOMStoreReadWriteTransaction rwTx1 = txChain.newReadWriteTransaction();
 964
 965                     rwTx1.ready();
 966
 967                     final DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction();
 968
 969                     final Optional<NormalizedNode<?, ?>> optional = rwTx2.read(TestModel.TEST_PATH).get(
 970                             5, TimeUnit.SECONDS);
 971                     assertEquals("isPresent", false, optional.isPresent());
 972
 973                     txChain.close();
 974                 }
 975             }
 976         };
 977     }
 978
 979     @Test
 980     public void testCreateChainedTransactionWhenPreviousNotReady() throws Exception {
 981         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
 982             {
 983                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
 984                         testParameter, "testCreateChainedTransactionWhenPreviousNotReady", "test-1")) {
 985
 986                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 987
 988                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 989                     assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 990
 991                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 992
 993                     // Try to create another Tx of each type - each should fail
 994                     // b/c the previous Tx wasn't
 995                     // readied.
 996                     assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
 997                 }
 998             }
 999         };
1000     }
1001
1002     @Test
1003     public void testCreateChainedTransactionAfterClose() throws Exception {
1004         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1005             {
1006                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1007                         testParameter, "testCreateChainedTransactionAfterClose", "test-1")) {
1008
1009                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
1010                     txChain.close();
1011
1012                     // Try to create another Tx of each type - should fail b/c
1013                     // the previous Tx was closed.
1014                     assertExceptionOnTxChainCreates(txChain, TransactionChainClosedException.class);
1015                 }
1016             }
1017         };
1018     }
1019
1020     @Test
1021     public void testChainWithReadOnlyTxAfterPreviousReady() throws Exception {
1022         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1023             {
1024                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
1025                         testParameter, "testChainWithReadOnlyTxAfterPreviousReady", "test-1")) {
1026
1027                     final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
1028
1029                     // Create a write tx and submit.
1030                     final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
1031                     writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1032                     final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
1033
1034                     // Create read-only tx's and issue a read.
1035                     CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture1 = txChain
1036                             .newReadOnlyTransaction().read(TestModel.TEST_PATH);
1037
1038                     CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture2 = txChain
1039                             .newReadOnlyTransaction().read(TestModel.TEST_PATH);
1040
1041                     // Create another write tx and issue the write.
1042                     DOMStoreWriteTransaction writeTx2 = txChain.newWriteOnlyTransaction();
1043                     writeTx2.write(TestModel.OUTER_LIST_PATH,
1044                             ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1045
1046                     // Ensure the reads succeed.
1047
1048                     assertEquals("isPresent", true, readFuture1.checkedGet(5, TimeUnit.SECONDS).isPresent());
1049                     assertEquals("isPresent", true, readFuture2.checkedGet(5, TimeUnit.SECONDS).isPresent());
1050
1051                     // Ensure the writes succeed.
1052                     DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
1053
1054                     doCommit(cohort1);
1055                     doCommit(cohort2);
1056
1057                     assertEquals("isPresent", true, txChain.newReadOnlyTransaction().read(TestModel.OUTER_LIST_PATH)
1058                             .checkedGet(5, TimeUnit.SECONDS).isPresent());
1059                 }
1060             }
1061         };
1062     }
1063
1064     @Test
1065     public void testChainedTransactionFailureWithSingleShard() throws Exception {
1066         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1067             {
1068                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
1069                         testParameter, "testChainedTransactionFailureWithSingleShard", "cars-1")) {
1070
1071                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
1072                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
1073                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
1074                             MoreExecutors.directExecutor());
1075
1076                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
1077                     final DOMTransactionChain txChain = broker.createTransactionChain(listener);
1078
1079                     final DOMDataReadWriteTransaction writeTx = txChain.newReadWriteTransaction();
1080
1081                     writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
1082                             PeopleModel.emptyContainer());
1083
1084                     final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
1085                             .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
1086                             .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
1087
1088                     writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
1089
1090                     try {
1091                         writeTx.submit().checkedGet(5, TimeUnit.SECONDS);
1092                         fail("Expected TransactionCommitFailedException");
1093                     } catch (final TransactionCommitFailedException e) {
1094                         // Expected
1095                     }
1096
1097                     verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
1098                             any(Throwable.class));
1099
1100                     txChain.close();
1101                     broker.close();
1102                 }
1103             }
1104         };
1105     }
1106
1107     @Test
1108     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
1109         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1110             {
1111                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
1112                         testParameter, "testChainedTransactionFailureWithMultipleShards", "cars-1", "people-1")) {
1113
1114                     final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
1115                             ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
1116                                     .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
1117                             MoreExecutors.directExecutor());
1118
1119                     final TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
1120                     final DOMTransactionChain txChain = broker.createTransactionChain(listener);
1121
1122                     final DOMDataReadWriteTransaction writeTx = txChain.newReadWriteTransaction();
1123
1124                     writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
1125                             PeopleModel.emptyContainer());
1126
1127                     final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
1128                             .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
1129                             .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
1130
1131                     writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
1132
1133                     // Note that merge will validate the data and fail but put
1134                     // succeeds b/c deep validation is not
1135                     // done for put for performance reasons.
1136                     try {
1137                         writeTx.submit().checkedGet(5, TimeUnit.SECONDS);
1138                         fail("Expected TransactionCommitFailedException");
1139                     } catch (final TransactionCommitFailedException e) {
1140                         // Expected
1141                     }
1142
1143                     verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
1144                             any(Throwable.class));
1145
1146                     txChain.close();
1147                     broker.close();
1148                 }
1149             }
1150         };
1151     }
1152
1153     @Test
1154     public void testChangeListenerRegistration() throws Exception {
1155         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1156             {
1157                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
1158                         testParameter, "testChangeListenerRegistration", "test-1")) {
1159
1160                     testWriteTransaction(dataStore, TestModel.TEST_PATH,
1161                             ImmutableNodes.containerNode(TestModel.TEST_QNAME));
1162
1163                     final MockDataChangeListener listener = new MockDataChangeListener(1);
1164
1165                     final ListenerRegistration<MockDataChangeListener> listenerReg = dataStore
1166                             .registerChangeListener(TestModel.TEST_PATH, listener, DataChangeScope.SUBTREE);
1167
1168                     assertNotNull("registerChangeListener returned null", listenerReg);
1169
1170                     // Wait for the initial notification
1171                     listener.waitForChangeEvents(TestModel.TEST_PATH);
1172                     listener.reset(2);
1173
1174                     // Write 2 updates.
1175                     testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
1176                             ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
1177
1178                     YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1179                             .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
1180                     testWriteTransaction(dataStore, listPath,
1181                             ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
1182
1183                     // Wait for the 2 updates.
1184                     listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
1185                     listenerReg.close();
1186
1187                     testWriteTransaction(dataStore,
1188                             YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
1189                                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
1190                             ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
1191
1192                     listener.expectNoMoreChanges("Received unexpected change after close");
1193                 }
1194             }
1195         };
1196     }
1197
1198     @Test
1199     public void testRestoreFromDatastoreSnapshot() throws Exception {
1200         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1201             {
1202                 final String name = "transactionIntegrationTest";
1203
1204                 final ContainerNode carsNode = CarsModel.newCarsNode(
1205                         CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)),
1206                                 CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L))));
1207
1208                 DataTree dataTree = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL);
1209                 dataTree.setSchemaContext(SchemaContextHelper.full());
1210                 AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
1211                 NormalizedNode<?, ?> root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
1212
1213                 final Snapshot carsSnapshot = Snapshot.create(
1214                         new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
1215                         Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
1216
1217                 dataTree = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL);
1218                 dataTree.setSchemaContext(SchemaContextHelper.full());
1219
1220                 final NormalizedNode<?, ?> peopleNode = PeopleModel.create();
1221                 AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode);
1222
1223                 root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
1224
1225                 final Snapshot peopleSnapshot = Snapshot.create(
1226                         new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
1227                         Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
1228
1229                 restoreFromSnapshot = new DatastoreSnapshot(name, null, Arrays.asList(
1230                         new DatastoreSnapshot.ShardSnapshot("cars", carsSnapshot),
1231                         new DatastoreSnapshot.ShardSnapshot("people", peopleSnapshot)));
1232
1233                 try (final AbstractDataStore dataStore = setupAbstractDataStore(
1234                         testParameter, name, "module-shards-member1.conf", true, "cars", "people")) {
1235
1236                     final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
1237
1238                     // two reads
1239                     Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1240                     assertEquals("isPresent", true, optional.isPresent());
1241                     assertEquals("Data node", carsNode, optional.get());
1242
1243                     optional = readTx.read(PeopleModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1244                     assertEquals("isPresent", true, optional.isPresent());
1245                     assertEquals("Data node", peopleNode, optional.get());
1246                 }
1247             }
1248         };
1249     }
1250
1251     @Test
1252     @Deprecated
1253     public void testRecoveryFromPreCarbonSnapshot() throws Exception {
1254         new IntegrationTestKit(getSystem(), datastoreContextBuilder) {
1255             {
1256                 final String name = "testRecoveryFromPreCarbonSnapshot";
1257
1258                 final ContainerNode carsNode = CarsModel.newCarsNode(
1259                         CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)),
1260                                 CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L))));
1261
1262                 DataTree dataTree = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL);
1263                 dataTree.setSchemaContext(SchemaContextHelper.full());
1264                 AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
1265                 NormalizedNode<?, ?> root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.EMPTY);
1266
1267                 MetadataShardDataTreeSnapshot shardSnapshot = new MetadataShardDataTreeSnapshot(root);
1268                 final ByteArrayOutputStream bos = new ByteArrayOutputStream();
1269                 try (final DataOutputStream dos = new DataOutputStream(bos)) {
1270                     PayloadVersion.BORON.writeTo(dos);
1271                     try (ObjectOutputStream oos = new ObjectOutputStream(dos)) {
1272                         oos.writeObject(shardSnapshot);
1273                     }
1274                 }
1275
1276                 final org.opendaylight.controller.cluster.raft.Snapshot snapshot =
1277                         org.opendaylight.controller.cluster.raft.Snapshot.create(bos.toByteArray(),
1278                                 Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
1279
1280                 InMemorySnapshotStore.addSnapshot("member-1-shard-cars-" + name, snapshot);
1281
1282                 try (AbstractDataStore dataStore = setupAbstractDataStore(
1283                         testParameter, name, "module-shards-member1.conf", true, "cars")) {
1284
1285                     DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
1286
1287                     Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1288                     assertEquals("isPresent", true, optional.isPresent());
1289                     assertEquals("Data node", carsNode, optional.get());
1290                 }
1291             }
1292         };
1293     }
1294 }