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

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import akka.actor.ActorRef;
   4 import akka.actor.ActorSystem;
   5 import akka.actor.PoisonPill;
   6 import com.google.common.base.Optional;
   7 import com.google.common.util.concurrent.CheckedFuture;
   8 import com.google.common.util.concurrent.Uninterruptibles;
   9 import static org.junit.Assert.assertEquals;
  10 import static org.junit.Assert.assertNotNull;
  11 import org.junit.Test;
  12 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  13 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  14 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  15 import org.opendaylight.controller.cluster.datastore.utils.InMemoryJournal;
  16 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
  17 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
  18 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  19 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  20 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  21 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  23 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  24 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  25 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  26 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  29 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  30 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  31 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  32 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  33 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  34 import java.util.concurrent.CountDownLatch;
  35 import java.util.concurrent.ExecutionException;
  36 import java.util.concurrent.TimeUnit;
  37 import java.util.concurrent.atomic.AtomicReference;
  38
  39 public class DistributedDataStoreIntegrationTest extends AbstractActorTest {
  40
  41     private final DatastoreContext.Builder datastoreContextBuilder =
  42             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100);
  43
  44     @Test
  45     public void testWriteTransactionWithSingleShard() throws Exception{
  46         System.setProperty("shard.persistent", "true");
  47         new IntegrationTestKit(getSystem()) {{
  48             DistributedDataStore dataStore =
  49                     setupDistributedDataStore("transactionIntegrationTest", "test-1");
  50
  51             testWriteTransaction(dataStore, TestModel.TEST_PATH,
  52                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
  53
  54             testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
  55                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
  56
  57             cleanup(dataStore);
  58         }};
  59     }
  60
  61     @Test
  62     public void testWriteTransactionWithMultipleShards() throws Exception{
  63         System.setProperty("shard.persistent", "true");
  64         new IntegrationTestKit(getSystem()) {{
  65             DistributedDataStore dataStore =
  66                     setupDistributedDataStore("testWriteTransactionWithMultipleShards", "cars-1", "people-1");
  67
  68             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
  69             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
  70
  71             YangInstanceIdentifier nodePath1 = CarsModel.BASE_PATH;
  72             NormalizedNode<?, ?> nodeToWrite1 = CarsModel.emptyContainer();
  73             writeTx.write(nodePath1, nodeToWrite1);
  74
  75             YangInstanceIdentifier nodePath2 = PeopleModel.BASE_PATH;
  76             NormalizedNode<?, ?> nodeToWrite2 = PeopleModel.emptyContainer();
  77             writeTx.write(nodePath2, nodeToWrite2);
  78
  79             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
  80
  81             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
  82             assertEquals("canCommit", true, canCommit);
  83             cohort.preCommit().get(5, TimeUnit.SECONDS);
  84             cohort.commit().get(5, TimeUnit.SECONDS);
  85
  86             // Verify the data in the store
  87
  88             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
  89
  90             Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath1).get();
  91             assertEquals("isPresent", true, optional.isPresent());
  92             assertEquals("Data node", nodeToWrite1, optional.get());
  93
  94             optional = readTx.read(nodePath2).get();
  95             assertEquals("isPresent", true, optional.isPresent());
  96             assertEquals("Data node", nodeToWrite2, optional.get());
  97
  98             cleanup(dataStore);
  99         }};
 100     }
 101
 102     @Test
 103     public void testReadWriteTransaction() throws Exception{
 104         System.setProperty("shard.persistent", "true");
 105         new IntegrationTestKit(getSystem()) {{
 106             DistributedDataStore dataStore =
 107                     setupDistributedDataStore("testReadWriteTransaction", "test-1");
 108
 109             // 1. Create a read-write Tx
 110
 111             DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 112             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 113
 114             // 2. Write some data
 115
 116             YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
 117             NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 118             readWriteTx.write(nodePath, nodeToWrite );
 119
 120             // 3. Read the data from Tx
 121
 122             Boolean exists = readWriteTx.exists(nodePath).checkedGet(5, TimeUnit.SECONDS);
 123             assertEquals("exists", true, exists);
 124
 125             Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
 126             assertEquals("isPresent", true, optional.isPresent());
 127             assertEquals("Data node", nodeToWrite, optional.get());
 128
 129             // 4. Ready the Tx for commit
 130
 131             DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
 132
 133             // 5. Commit the Tx
 134
 135             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 136             assertEquals("canCommit", true, canCommit);
 137             cohort.preCommit().get(5, TimeUnit.SECONDS);
 138             cohort.commit().get(5, TimeUnit.SECONDS);
 139
 140             // 6. Verify the data in the store
 141
 142             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 143
 144             optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
 145             assertEquals("isPresent", true, optional.isPresent());
 146             assertEquals("Data node", nodeToWrite, optional.get());
 147
 148             cleanup(dataStore);
 149         }};
 150     }
 151
 152     @Test
 153     public void testTransactionWritesWithShardNotInitiallyReady() throws Exception{
 154         new IntegrationTestKit(getSystem()) {{
 155             String testName = "testTransactionWritesWithShardNotInitiallyReady";
 156             String shardName = "test-1";
 157
 158             // Setup the InMemoryJournal to block shard recovery to ensure the shard isn't
 159             // initialized until we create and submit the write the Tx.
 160             String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 161             CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 162             InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 163
 164             DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
 165
 166             // Create the write Tx
 167
 168             final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 169             assertNotNull("newReadWriteTransaction returned null", writeTx);
 170
 171             // Do some modification operations and ready the Tx on a separate thread.
 172
 173             final YangInstanceIdentifier listEntryPath = YangInstanceIdentifier.builder(
 174                     TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME,
 175                             TestModel.ID_QNAME, 1).build();
 176
 177             final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 178             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 179             final CountDownLatch txReady = new CountDownLatch(1);
 180             Thread txThread = new Thread() {
 181                 @Override
 182                 public void run() {
 183                     try {
 184                         writeTx.write(TestModel.TEST_PATH,
 185                                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 186
 187                         writeTx.merge(TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(
 188                                 TestModel.OUTER_LIST_QNAME).build());
 189
 190                         writeTx.write(listEntryPath, ImmutableNodes.mapEntry(
 191                                 TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
 192
 193                         writeTx.delete(listEntryPath);
 194
 195                         txCohort.set(writeTx.ready());
 196                     } catch(Exception e) {
 197                         caughtEx.set(e);
 198                         return;
 199                     } finally {
 200                         txReady.countDown();
 201                     }
 202                 }
 203             };
 204
 205             txThread.start();
 206
 207             // Wait for the Tx operations to complete.
 208
 209             boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 210             if(caughtEx.get() != null) {
 211                 throw caughtEx.get();
 212             }
 213
 214             assertEquals("Tx ready", true, done);
 215
 216             // At this point the Tx operations should be waiting for the shard to initialize so
 217             // trigger the latch to let the shard recovery to continue.
 218
 219             blockRecoveryLatch.countDown();
 220
 221             // Wait for the Tx commit to complete.
 222
 223             assertEquals("canCommit", true, txCohort.get().canCommit().get(5, TimeUnit.SECONDS));
 224             txCohort.get().preCommit().get(5, TimeUnit.SECONDS);
 225             txCohort.get().commit().get(5, TimeUnit.SECONDS);
 226
 227             // Verify the data in the store
 228
 229             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 230
 231             Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).
 232                     get(5, TimeUnit.SECONDS);
 233             assertEquals("isPresent", true, optional.isPresent());
 234
 235             optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 236             assertEquals("isPresent", true, optional.isPresent());
 237
 238             optional = readTx.read(listEntryPath).get(5, TimeUnit.SECONDS);
 239             assertEquals("isPresent", false, optional.isPresent());
 240
 241             cleanup(dataStore);
 242         }};
 243     }
 244
 245     @Test
 246     public void testTransactionReadsWithShardNotInitiallyReady() throws Exception{
 247         new IntegrationTestKit(getSystem()) {{
 248             String testName = "testTransactionReadsWithShardNotInitiallyReady";
 249             String shardName = "test-1";
 250
 251             // Setup the InMemoryJournal to block shard recovery to ensure the shard isn't
 252             // initialized until we create the Tx.
 253             String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 254             CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 255             InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 256
 257             DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
 258
 259             // Create the read-write Tx
 260
 261             final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 262             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 263
 264             // Do some reads on the Tx on a separate thread.
 265
 266             final AtomicReference<CheckedFuture<Boolean, ReadFailedException>> txExistsFuture =
 267                     new AtomicReference<>();
 268             final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
 269                     txReadFuture = new AtomicReference<>();
 270             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 271             final CountDownLatch txReadsDone = new CountDownLatch(1);
 272             Thread txThread = new Thread() {
 273                 @Override
 274                 public void run() {
 275                     try {
 276                         readWriteTx.write(TestModel.TEST_PATH,
 277                                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 278
 279                         txExistsFuture.set(readWriteTx.exists(TestModel.TEST_PATH));
 280
 281                         txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
 282                     } catch(Exception e) {
 283                         caughtEx.set(e);
 284                         return;
 285                     } finally {
 286                         txReadsDone.countDown();
 287                     }
 288                 }
 289             };
 290
 291             txThread.start();
 292
 293             // Wait for the Tx operations to complete.
 294
 295             boolean done = Uninterruptibles.awaitUninterruptibly(txReadsDone, 5, TimeUnit.SECONDS);
 296             if(caughtEx.get() != null) {
 297                 throw caughtEx.get();
 298             }
 299
 300             assertEquals("Tx reads done", true, done);
 301
 302             // At this point the Tx operations should be waiting for the shard to initialize so
 303             // trigger the latch to let the shard recovery to continue.
 304
 305             blockRecoveryLatch.countDown();
 306
 307             // Wait for the reads to complete and verify.
 308
 309             assertEquals("exists", true, txExistsFuture.get().checkedGet(5, TimeUnit.SECONDS));
 310             assertEquals("read", true, txReadFuture.get().checkedGet(5, TimeUnit.SECONDS).isPresent());
 311
 312             readWriteTx.close();
 313
 314             cleanup(dataStore);
 315         }};
 316     }
 317
 318     @Test(expected=NotInitializedException.class)
 319     public void testTransactionCommitFailureWithShardNotInitialized() throws Throwable{
 320         new IntegrationTestKit(getSystem()) {{
 321             String testName = "testTransactionCommitFailureWithShardNotInitialized";
 322             String shardName = "test-1";
 323
 324             // Set the shard initialization timeout low for the test.
 325
 326             datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
 327
 328             // Setup the InMemoryJournal to block shard recovery indefinitely.
 329
 330             String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 331             CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 332             InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 333
 334             DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
 335
 336             // Create the write Tx
 337
 338             final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 339             assertNotNull("newReadWriteTransaction returned null", writeTx);
 340
 341             // Do some modifications and ready the Tx on a separate thread.
 342
 343             final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 344             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 345             final CountDownLatch txReady = new CountDownLatch(1);
 346             Thread txThread = new Thread() {
 347                 @Override
 348                 public void run() {
 349                     try {
 350                         writeTx.write(TestModel.TEST_PATH,
 351                                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 352
 353                         txCohort.set(writeTx.ready());
 354                     } catch(Exception e) {
 355                         caughtEx.set(e);
 356                         return;
 357                     } finally {
 358                         txReady.countDown();
 359                     }
 360                 }
 361             };
 362
 363             txThread.start();
 364
 365             // Wait for the Tx operations to complete.
 366
 367             boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 368             if(caughtEx.get() != null) {
 369                 throw caughtEx.get();
 370             }
 371
 372             assertEquals("Tx ready", true, done);
 373
 374             // Wait for the commit to complete. Since the shard never initialized, the Tx should
 375             // have timed out and throw an appropriate exception cause.
 376
 377             try {
 378                 txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
 379             } catch(ExecutionException e) {
 380                 throw e.getCause();
 381             } finally {
 382                 blockRecoveryLatch.countDown();
 383                 cleanup(dataStore);
 384             }
 385         }};
 386     }
 387
 388     @Test(expected=NotInitializedException.class)
 389     public void testTransactionReadFailureWithShardNotInitialized() throws Throwable{
 390         new IntegrationTestKit(getSystem()) {{
 391             String testName = "testTransactionReadFailureWithShardNotInitialized";
 392             String shardName = "test-1";
 393
 394             // Set the shard initialization timeout low for the test.
 395
 396             datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
 397
 398             // Setup the InMemoryJournal to block shard recovery indefinitely.
 399
 400             String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
 401             CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
 402             InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
 403
 404             DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
 405
 406             // Create the read-write Tx
 407
 408             final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 409             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 410
 411             // Do a read on the Tx on a separate thread.
 412
 413             final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
 414                     txReadFuture = new AtomicReference<>();
 415             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 416             final CountDownLatch txReadDone = new CountDownLatch(1);
 417             Thread txThread = new Thread() {
 418                 @Override
 419                 public void run() {
 420                     try {
 421                         readWriteTx.write(TestModel.TEST_PATH,
 422                                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 423
 424                         txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
 425
 426                         readWriteTx.close();
 427                     } catch(Exception e) {
 428                         caughtEx.set(e);
 429                         return;
 430                     } finally {
 431                         txReadDone.countDown();
 432                     }
 433                 }
 434             };
 435
 436             txThread.start();
 437
 438             // Wait for the Tx operations to complete.
 439
 440             boolean done = Uninterruptibles.awaitUninterruptibly(txReadDone, 5, TimeUnit.SECONDS);
 441             if(caughtEx.get() != null) {
 442                 throw caughtEx.get();
 443             }
 444
 445             assertEquals("Tx read done", true, done);
 446
 447             // Wait for the read to complete. Since the shard never initialized, the Tx should
 448             // have timed out and throw an appropriate exception cause.
 449
 450             try {
 451                 txReadFuture.get().checkedGet(5, TimeUnit.SECONDS);
 452             } catch(ReadFailedException e) {
 453                 throw e.getCause();
 454             } finally {
 455                 blockRecoveryLatch.countDown();
 456                 cleanup(dataStore);
 457             }
 458         }};
 459     }
 460
 461     @Test(expected=NoShardLeaderException.class)
 462     public void testTransactionCommitFailureWithNoShardLeader() throws Throwable{
 463         new IntegrationTestKit(getSystem()) {{
 464             String testName = "testTransactionCommitFailureWithNoShardLeader";
 465             String shardName = "test-1";
 466
 467             // We don't want the shard to become the leader so prevent shard election from completing
 468             // by setting the election timeout, which is based on the heartbeat interval, really high.
 469
 470             datastoreContextBuilder.shardHeartbeatIntervalInMillis(30000);
 471
 472             // Set the leader election timeout low for the test.
 473
 474             datastoreContextBuilder.shardLeaderElectionTimeout(1, TimeUnit.MILLISECONDS);
 475
 476             DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
 477
 478             // Create the write Tx.
 479
 480             final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 481             assertNotNull("newReadWriteTransaction returned null", writeTx);
 482
 483             // Do some modifications and ready the Tx on a separate thread.
 484
 485             final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 486             final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 487             final CountDownLatch txReady = new CountDownLatch(1);
 488             Thread txThread = new Thread() {
 489                 @Override
 490                 public void run() {
 491                     try {
 492                         writeTx.write(TestModel.TEST_PATH,
 493                                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 494
 495                         txCohort.set(writeTx.ready());
 496                     } catch(Exception e) {
 497                         caughtEx.set(e);
 498                         return;
 499                     } finally {
 500                         txReady.countDown();
 501                     }
 502                 }
 503             };
 504
 505             txThread.start();
 506
 507             // Wait for the Tx operations to complete.
 508
 509             boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 510             if(caughtEx.get() != null) {
 511                 throw caughtEx.get();
 512             }
 513
 514             assertEquals("Tx ready", true, done);
 515
 516             // Wait for the commit to complete. Since no shard leader was elected in time, the Tx
 517             // should have timed out and throw an appropriate exception cause.
 518
 519             try {
 520                 txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
 521             } catch(ExecutionException e) {
 522                 throw e.getCause();
 523             } finally {
 524                 cleanup(dataStore);
 525             }
 526         }};
 527     }
 528
 529     @Test
 530     public void testTransactionAbort() throws Exception{
 531         System.setProperty("shard.persistent", "true");
 532         new IntegrationTestKit(getSystem()) {{
 533             DistributedDataStore dataStore =
 534                     setupDistributedDataStore("transactionAbortIntegrationTest", "test-1");
 535
 536             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 537             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 538
 539             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 540
 541             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 542
 543             cohort.canCommit().get(5, TimeUnit.SECONDS);
 544
 545             cohort.abort().get(5, TimeUnit.SECONDS);
 546
 547             testWriteTransaction(dataStore, TestModel.TEST_PATH,
 548                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 549
 550             cleanup(dataStore);
 551         }};
 552     }
 553
 554     @Test
 555     public void testTransactionChain() throws Exception{
 556         System.setProperty("shard.persistent", "true");
 557         new IntegrationTestKit(getSystem()) {{
 558             DistributedDataStore dataStore =
 559                     setupDistributedDataStore("transactionChainIntegrationTest", "test-1");
 560
 561             // 1. Create a Tx chain and write-only Tx
 562
 563             DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 564
 565             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 566             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 567
 568             // 2. Write some data
 569
 570             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 571             writeTx.write(TestModel.TEST_PATH, containerNode);
 572
 573             // 3. Ready the Tx for commit
 574
 575             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 576
 577             // 4. Commit the Tx
 578
 579             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 580             assertEquals("canCommit", true, canCommit);
 581             cohort.preCommit().get(5, TimeUnit.SECONDS);
 582             cohort.commit().get(5, TimeUnit.SECONDS);
 583
 584             // 5. Verify the data in the store
 585
 586             DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
 587
 588             Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 589             assertEquals("isPresent", true, optional.isPresent());
 590             assertEquals("Data node", containerNode, optional.get());
 591
 592             txChain.close();
 593
 594             cleanup(dataStore);
 595         }};
 596     }
 597
 598     @Test
 599     public void testChangeListenerRegistration() throws Exception{
 600         new IntegrationTestKit(getSystem()) {{
 601             DistributedDataStore dataStore =
 602                     setupDistributedDataStore("testChangeListenerRegistration", "test-1");
 603
 604             MockDataChangeListener listener = new MockDataChangeListener(3);
 605
 606             ListenerRegistration<MockDataChangeListener>
 607                     listenerReg = dataStore.registerChangeListener(TestModel.TEST_PATH, listener,
 608                             DataChangeScope.SUBTREE);
 609
 610             assertNotNull("registerChangeListener returned null", listenerReg);
 611
 612             testWriteTransaction(dataStore, TestModel.TEST_PATH,
 613                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 614
 615             testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
 616                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
 617
 618             YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
 619                     nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
 620             testWriteTransaction(dataStore, listPath,
 621                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
 622
 623             listener.waitForChangeEvents(TestModel.TEST_PATH, TestModel.OUTER_LIST_PATH, listPath );
 624
 625             listenerReg.close();
 626
 627             testWriteTransaction(dataStore, YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
 628                     nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
 629                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
 630
 631             listener.expectNoMoreChanges("Received unexpected change after close");
 632
 633             cleanup(dataStore);
 634         }};
 635     }
 636
 637     class IntegrationTestKit extends ShardTestKit {
 638
 639         IntegrationTestKit(ActorSystem actorSystem) {
 640             super(actorSystem);
 641         }
 642
 643         DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) {
 644             return setupDistributedDataStore(typeName, true, shardNames);
 645         }
 646
 647         DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader,
 648                 String... shardNames) {
 649             MockClusterWrapper cluster = new MockClusterWrapper();
 650             Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
 651             ShardStrategyFactory.setConfiguration(config);
 652
 653             DatastoreContext datastoreContext = datastoreContextBuilder.build();
 654             DistributedDataStore dataStore = new DistributedDataStore(getSystem(), typeName, cluster,
 655                     config, datastoreContext);
 656
 657             SchemaContext schemaContext = SchemaContextHelper.full();
 658             dataStore.onGlobalContextUpdated(schemaContext);
 659
 660             if(waitUntilLeader) {
 661                 for(String shardName: shardNames) {
 662                     ActorRef shard = null;
 663                     for(int i = 0; i < 20 * 5 && shard == null; i++) {
 664                         Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 665                         Optional<ActorRef> shardReply = dataStore.getActorContext().findLocalShard(shardName);
 666                         if(shardReply.isPresent()) {
 667                             shard = shardReply.get();
 668                         }
 669                     }
 670
 671                     assertNotNull("Shard was not created", shard);
 672
 673                     waitUntilLeader(shard);
 674                 }
 675             }
 676
 677             return dataStore;
 678         }
 679
 680         void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath,
 681                 NormalizedNode<?, ?> nodeToWrite) throws Exception {
 682
 683             // 1. Create a write-only Tx
 684
 685             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 686             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 687
 688             // 2. Write some data
 689
 690             writeTx.write(nodePath, nodeToWrite);
 691
 692             // 3. Ready the Tx for commit
 693
 694             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 695
 696             // 4. Commit the Tx
 697
 698             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 699             assertEquals("canCommit", true, canCommit);
 700             cohort.preCommit().get(5, TimeUnit.SECONDS);
 701             cohort.commit().get(5, TimeUnit.SECONDS);
 702
 703             // 5. Verify the data in the store
 704
 705             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 706
 707             Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
 708             assertEquals("isPresent", true, optional.isPresent());
 709             assertEquals("Data node", nodeToWrite, optional.get());
 710         }
 711
 712         void cleanup(DistributedDataStore dataStore) {
 713             dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), null);
 714         }
 715     }
 716
 717 }