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

   1 /*
   2  * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static org.awaitility.Awaitility.await;
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertFalse;
  13 import static org.junit.Assert.assertNotNull;
  14 import static org.junit.Assert.assertTrue;
  15 import static org.junit.Assert.fail;
  16 import static org.mockito.ArgumentMatchers.any;
  17 import static org.mockito.ArgumentMatchers.eq;
  18 import static org.mockito.Mockito.timeout;
  19 import static org.mockito.Mockito.verify;
  20
  21 import akka.actor.ActorRef;
  22 import akka.actor.ActorSystem;
  23 import com.google.common.base.Throwables;
  24 import com.google.common.collect.ImmutableMap;
  25 import com.google.common.util.concurrent.FluentFuture;
  26 import com.google.common.util.concurrent.ListenableFuture;
  27 import com.google.common.util.concurrent.MoreExecutors;
  28 import com.google.common.util.concurrent.Uninterruptibles;
  29 import java.util.ArrayList;
  30 import java.util.Arrays;
  31 import java.util.Collection;
  32 import java.util.Collections;
  33 import java.util.Iterator;
  34 import java.util.List;
  35 import java.util.Optional;
  36 import java.util.concurrent.CountDownLatch;
  37 import java.util.concurrent.ExecutionException;
  38 import java.util.concurrent.TimeUnit;
  39 import java.util.concurrent.atomic.AtomicReference;
  40 import org.junit.Ignore;
  41 import org.junit.Test;
  42 import org.junit.runners.Parameterized.Parameter;
  43 import org.mockito.Mockito;
  44 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
  45 import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
  46 import org.opendaylight.controller.cluster.datastore.TestShard.RequestFrontendMetadata;
  47 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  48 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  49 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  50 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
  51 import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata;
  52 import org.opendaylight.controller.cluster.datastore.persisted.FrontendShardDataTreeSnapshotMetadata;
  53 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
  54 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
  55 import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
  56 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  57 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  58 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  59 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  60 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  61 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  62 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  63 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  64 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  65 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
  66 import org.opendaylight.mdsal.dom.api.DOMTransactionChainClosedException;
  67 import org.opendaylight.mdsal.dom.api.DOMTransactionChainListener;
  68 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
  69 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
  70 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
  71 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  72 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
  73 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
  74 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  75 import org.opendaylight.yangtools.yang.common.Uint64;
  76 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  77 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  78 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  79 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  80 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  81 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  82 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
  83 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  84 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  85 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
  86 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  87
  88 public abstract class AbstractDistributedDataStoreIntegrationTest {
  89
  90     @Parameter
  91     public Class<? extends AbstractDataStore> testParameter;
  92
  93     protected ActorSystem system;
  94
  95     protected final DatastoreContext.Builder datastoreContextBuilder = DatastoreContext.newBuilder()
  96             .shardHeartbeatIntervalInMillis(100);
  97
  98     protected ActorSystem getSystem() {
  99         return system;
 100     }
 101
 102     @Test
 103     public void testWriteTransactionWithSingleShard() throws Exception {
 104         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 105         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 106             testParameter, "transactionIntegrationTest", "test-1")) {
 107
 108             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
 109                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 110
 111             testKit.testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
 112                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
 113                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
 114                 .build());
 115         }
 116     }
 117
 118     @Test
 119     public void testWriteTransactionWithMultipleShards() throws Exception {
 120         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 121         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 122             testParameter, "testWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
 123
 124             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 125             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 126
 127             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 128             writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 129
 130             testKit.doCommit(writeTx.ready());
 131
 132             writeTx = dataStore.newWriteOnlyTransaction();
 133
 134             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 135             writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 136
 137             testKit.doCommit(writeTx.ready());
 138
 139             writeTx = dataStore.newWriteOnlyTransaction();
 140
 141             final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 142             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 143             writeTx.write(carPath, car);
 144
 145             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
 146             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 147             writeTx.write(personPath, person);
 148
 149             testKit.doCommit(writeTx.ready());
 150
 151             // Verify the data in the store
 152             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 153
 154             Optional<NormalizedNode> optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
 155             assertTrue("isPresent", optional.isPresent());
 156             assertEquals("Data node", car, optional.get());
 157
 158             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
 159             assertTrue("isPresent", optional.isPresent());
 160             assertEquals("Data node", person, optional.get());
 161         }
 162     }
 163
 164     @Test
 165     public void testReadWriteTransactionWithSingleShard() throws Exception {
 166         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 167         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 168             testParameter, "testReadWriteTransactionWithSingleShard", "test-1")) {
 169
 170             // 1. Create a read-write Tx
 171             final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 172             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 173
 174             // 2. Write some data
 175             final YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
 176             final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 177             readWriteTx.write(nodePath, nodeToWrite);
 178
 179             // 3. Read the data from Tx
 180             final Boolean exists = readWriteTx.exists(nodePath).get(5, TimeUnit.SECONDS);
 181             assertEquals("exists", Boolean.TRUE, exists);
 182
 183             Optional<NormalizedNode> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
 184             assertTrue("isPresent", optional.isPresent());
 185             assertEquals("Data node", nodeToWrite, optional.get());
 186
 187             // 4. Ready the Tx for commit
 188             final DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
 189
 190             // 5. Commit the Tx
 191             testKit.doCommit(cohort);
 192
 193             // 6. Verify the data in the store
 194             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 195
 196             optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
 197             assertTrue("isPresent", optional.isPresent());
 198             assertEquals("Data node", nodeToWrite, optional.get());
 199         }
 200     }
 201
 202     @Test
 203     public void testReadWriteTransactionWithMultipleShards() throws Exception {
 204         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 205         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 206             testParameter, "testReadWriteTransactionWithMultipleShards", "cars-1", "people-1")) {
 207
 208             DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 209             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 210
 211             readWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 212             readWriteTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 213
 214             testKit.doCommit(readWriteTx.ready());
 215
 216             readWriteTx = dataStore.newReadWriteTransaction();
 217
 218             readWriteTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 219             readWriteTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 220
 221             testKit.doCommit(readWriteTx.ready());
 222
 223             readWriteTx = dataStore.newReadWriteTransaction();
 224
 225             final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 226             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 227             readWriteTx.write(carPath, car);
 228
 229             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
 230             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 231             readWriteTx.write(personPath, person);
 232
 233             final Boolean exists = readWriteTx.exists(carPath).get(5, TimeUnit.SECONDS);
 234             assertEquals("exists", Boolean.TRUE, exists);
 235
 236             Optional<NormalizedNode> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
 237             assertTrue("isPresent", optional.isPresent());
 238             assertEquals("Data node", car, optional.get());
 239
 240             testKit.doCommit(readWriteTx.ready());
 241
 242             // Verify the data in the store
 243             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 244
 245             optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
 246             assertTrue("isPresent", optional.isPresent());
 247             assertEquals("Data node", car, optional.get());
 248
 249             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
 250             assertTrue("isPresent", optional.isPresent());
 251             assertEquals("Data node", person, optional.get());
 252         }
 253     }
 254
 255     @Test
 256     @Ignore("Flushes a closed tx leak in single node, needs to be handled separately")
 257     public void testSingleTransactionsWritesInQuickSuccession() throws Exception {
 258         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 259         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 260             testParameter, "testSingleTransactionsWritesInQuickSuccession", "cars-1")) {
 261
 262             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 263
 264             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 265             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 266             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 267             testKit.doCommit(writeTx.ready());
 268
 269             int numCars = 5;
 270             for (int i = 0; i < numCars; i++) {
 271                 writeTx = txChain.newWriteOnlyTransaction();
 272                 writeTx.write(CarsModel.newCarPath("car" + i),
 273                     CarsModel.newCarEntry("car" + i, Uint64.valueOf(20000)));
 274
 275                 testKit.doCommit(writeTx.ready());
 276
 277                 DOMStoreReadTransaction domStoreReadTransaction = txChain.newReadOnlyTransaction();
 278                 domStoreReadTransaction.read(CarsModel.BASE_PATH).get();
 279
 280                 domStoreReadTransaction.close();
 281             }
 282
 283             // verify frontend metadata has no holes in purged transactions causing overtime memory leak
 284             Optional<ActorRef> localShard = dataStore.getActorUtils().findLocalShard("cars-1");
 285             FrontendShardDataTreeSnapshotMetadata frontendMetadata =
 286                     (FrontendShardDataTreeSnapshotMetadata) dataStore.getActorUtils()
 287                             .executeOperation(localShard.get(), new RequestFrontendMetadata());
 288
 289             if (dataStore.getActorUtils().getDatastoreContext().isUseTellBasedProtocol()) {
 290                 Iterator<FrontendHistoryMetadata> iterator =
 291                         frontendMetadata.getClients().get(0).getCurrentHistories().iterator();
 292                 FrontendHistoryMetadata metadata = iterator.next();
 293                 while (iterator.hasNext() && metadata.getHistoryId() != 1) {
 294                     metadata = iterator.next();
 295                 }
 296                 assertEquals(1, metadata.getPurgedTransactions().size());
 297             } else {
 298                 // ask based should track no metadata
 299                 assertTrue(frontendMetadata.getClients().get(0).getCurrentHistories().isEmpty());
 300             }
 301
 302             final Optional<NormalizedNode> optional = txChain.newReadOnlyTransaction()
 303                     .read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
 304             assertTrue("isPresent", optional.isPresent());
 305             assertEquals("# cars", numCars, ((Collection<?>) optional.get().body()).size());
 306         }
 307     }
 308
 309     @SuppressWarnings("checkstyle:IllegalCatch")
 310     private void testTransactionCommitFailureWithNoShardLeader(final boolean writeOnly, final String testName)
 311             throws Exception {
 312         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 313         final String shardName = "default";
 314
 315         // We don't want the shard to become the leader so prevent shard
 316         // elections.
 317         datastoreContextBuilder.customRaftPolicyImplementation(
 318                 "org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy");
 319
 320         // The ShardManager uses the election timeout for FindPrimary so
 321         // reset it low so it will timeout quickly.
 322         datastoreContextBuilder.shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1)
 323         .shardInitializationTimeout(200, TimeUnit.MILLISECONDS).frontendRequestTimeoutInSeconds(2);
 324
 325         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(testParameter, testName, false, shardName)) {
 326
 327             final Object result = dataStore.getActorUtils().executeOperation(
 328                 dataStore.getActorUtils().getShardManager(), new FindLocalShard(shardName, true));
 329             assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
 330
 331             // Create the write Tx.
 332             DOMStoreWriteTransaction writeTxToClose = null;
 333             try {
 334                 writeTxToClose = writeOnly ? dataStore.newWriteOnlyTransaction()
 335                         : dataStore.newReadWriteTransaction();
 336                 final DOMStoreWriteTransaction writeTx = writeTxToClose;
 337                 assertNotNull("newReadWriteTransaction returned null", writeTx);
 338
 339                 // Do some modifications and ready the Tx on a separate
 340                 // thread.
 341                 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
 342                 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
 343                 final CountDownLatch txReady = new CountDownLatch(1);
 344                 final Thread txThread = new Thread(() -> {
 345                     try {
 346                         writeTx.write(TestModel.JUNK_PATH,
 347                             ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 348
 349                         txCohort.set(writeTx.ready());
 350                     } catch (Exception e) {
 351                         caughtEx.set(e);
 352                     } finally {
 353                         txReady.countDown();
 354                     }
 355                 });
 356
 357                 txThread.start();
 358
 359                 // Wait for the Tx operations to complete.
 360                 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
 361                 if (caughtEx.get() != null) {
 362                     throw caughtEx.get();
 363                 }
 364
 365                 assertTrue("Tx ready", done);
 366
 367                 // Wait for the commit to complete. Since no shard
 368                 // leader was elected in time, the Tx
 369                 // should have timed out and throw an appropriate
 370                 // exception cause.
 371                 try {
 372                     txCohort.get().canCommit().get(10, TimeUnit.SECONDS);
 373                     fail("Expected NoShardLeaderException");
 374                 } catch (final ExecutionException e) {
 375                     final String msg = "Unexpected exception: "
 376                             + Throwables.getStackTraceAsString(e.getCause());
 377                     if (DistributedDataStore.class.isAssignableFrom(testParameter)) {
 378                         assertTrue(Throwables.getRootCause(e) instanceof NoShardLeaderException);
 379                     } else {
 380                         assertTrue(msg, Throwables.getRootCause(e) instanceof RequestTimeoutException);
 381                     }
 382                 }
 383             } finally {
 384                 try {
 385                     if (writeTxToClose != null) {
 386                         writeTxToClose.close();
 387                     }
 388                 } catch (Exception e) {
 389                     // FIXME TransactionProxy.close throws IllegalStateException:
 390                     // Transaction is ready, it cannot be closed
 391                 }
 392             }
 393         }
 394     }
 395
 396     @Test
 397     public void testWriteOnlyTransactionCommitFailureWithNoShardLeader() throws Exception {
 398         datastoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 399         testTransactionCommitFailureWithNoShardLeader(true, "testWriteOnlyTransactionCommitFailureWithNoShardLeader");
 400     }
 401
 402     @Test
 403     public void testReadWriteTransactionCommitFailureWithNoShardLeader() throws Exception {
 404         testTransactionCommitFailureWithNoShardLeader(false, "testReadWriteTransactionCommitFailureWithNoShardLeader");
 405     }
 406
 407     @Test
 408     public void testTransactionAbort() throws Exception {
 409         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 410         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 411             testParameter, "transactionAbortIntegrationTest", "test-1")) {
 412
 413             final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 414             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 415
 416             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 417
 418             final DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 419
 420             cohort.canCommit().get(5, TimeUnit.SECONDS);
 421
 422             cohort.abort().get(5, TimeUnit.SECONDS);
 423
 424             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
 425                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 426         }
 427     }
 428
 429     @Test
 430     @SuppressWarnings("checkstyle:IllegalCatch")
 431     public void testTransactionChainWithSingleShard() throws Exception {
 432         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 433         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 434             testParameter, "testTransactionChainWithSingleShard", "test-1")) {
 435
 436             // 1. Create a Tx chain and write-only Tx
 437             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 438
 439             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 440             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 441
 442             // 2. Write some data
 443             final NormalizedNode testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 444             writeTx.write(TestModel.TEST_PATH, testNode);
 445
 446             // 3. Ready the Tx for commit
 447             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 448
 449             // 4. Commit the Tx on another thread that first waits for
 450             // the second read Tx.
 451             final CountDownLatch continueCommit1 = new CountDownLatch(1);
 452             final CountDownLatch commit1Done = new CountDownLatch(1);
 453             final AtomicReference<Exception> commit1Error = new AtomicReference<>();
 454             new Thread(() -> {
 455                 try {
 456                     continueCommit1.await();
 457                     testKit.doCommit(cohort1);
 458                 } catch (Exception e) {
 459                     commit1Error.set(e);
 460                 } finally {
 461                     commit1Done.countDown();
 462                 }
 463             }).start();
 464
 465             // 5. Create a new read Tx from the chain to read and verify
 466             // the data from the first
 467             // Tx is visible after being readied.
 468             DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
 469             Optional<NormalizedNode> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 470             assertTrue("isPresent", optional.isPresent());
 471             assertEquals("Data node", testNode, optional.get());
 472
 473             // 6. Create a new RW Tx from the chain, write more data,
 474             // and ready it
 475             final DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 476             final MapNode outerNode = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
 477                     .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
 478                     .build();
 479             rwTx.write(TestModel.OUTER_LIST_PATH, outerNode);
 480
 481             final DOMStoreThreePhaseCommitCohort cohort2 = rwTx.ready();
 482
 483             // 7. Create a new read Tx from the chain to read the data
 484             // from the last RW Tx to
 485             // verify it is visible.
 486             readTx = txChain.newReadWriteTransaction();
 487             optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 488             assertTrue("isPresent", optional.isPresent());
 489             assertEquals("Data node", outerNode, optional.get());
 490
 491             // 8. Wait for the 2 commits to complete and close the
 492             // chain.
 493             continueCommit1.countDown();
 494             Uninterruptibles.awaitUninterruptibly(commit1Done, 5, TimeUnit.SECONDS);
 495
 496             if (commit1Error.get() != null) {
 497                 throw commit1Error.get();
 498             }
 499
 500             testKit.doCommit(cohort2);
 501
 502             txChain.close();
 503
 504             // 9. Create a new read Tx from the data store and verify
 505             // committed data.
 506             readTx = dataStore.newReadOnlyTransaction();
 507             optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
 508             assertTrue("isPresent", optional.isPresent());
 509             assertEquals("Data node", outerNode, optional.get());
 510         }
 511     }
 512
 513     @Test
 514     public void testTransactionChainWithMultipleShards() throws Exception {
 515         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 516         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 517             testParameter, "testTransactionChainWithMultipleShards", "cars-1", "people-1")) {
 518
 519             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 520
 521             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 522             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 523
 524             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 525             writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 526
 527             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 528             writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 529
 530             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 531
 532             final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
 533
 534             final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 535             final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 536             readWriteTx.write(carPath, car);
 537
 538             final MapEntryNode person = PeopleModel.newPersonEntry("jack");
 539             final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 540             readWriteTx.merge(personPath, person);
 541
 542             Optional<NormalizedNode> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
 543             assertTrue("isPresent", optional.isPresent());
 544             assertEquals("Data node", car, optional.get());
 545
 546             optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
 547             assertTrue("isPresent", optional.isPresent());
 548             assertEquals("Data node", person, optional.get());
 549
 550             final DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
 551
 552             writeTx = txChain.newWriteOnlyTransaction();
 553
 554             writeTx.delete(carPath);
 555
 556             final DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
 557
 558             final ListenableFuture<Boolean> canCommit1 = cohort1.canCommit();
 559             final ListenableFuture<Boolean> canCommit2 = cohort2.canCommit();
 560
 561             testKit.doCommit(canCommit1, cohort1);
 562             testKit.doCommit(canCommit2, cohort2);
 563             testKit.doCommit(cohort3);
 564
 565             txChain.close();
 566
 567             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 568
 569             optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
 570             assertFalse("isPresent", optional.isPresent());
 571
 572             optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
 573             assertTrue("isPresent", optional.isPresent());
 574             assertEquals("Data node", person, optional.get());
 575         }
 576     }
 577
 578     @Test
 579     public void testCreateChainedTransactionsInQuickSuccession() throws Exception {
 580         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 581         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 582             testParameter, "testCreateChainedTransactionsInQuickSuccession", "cars-1")) {
 583
 584             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 585                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
 586                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
 587                 MoreExecutors.directExecutor());
 588
 589             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
 590             DOMTransactionChain txChain = broker.createTransactionChain(listener);
 591
 592             final List<ListenableFuture<?>> futures = new ArrayList<>();
 593
 594             final DOMDataTreeWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 595             writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, CarsModel.emptyContainer());
 596             writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 597             futures.add(writeTx.commit());
 598
 599             int numCars = 100;
 600             for (int i = 0; i < numCars; i++) {
 601                 final DOMDataTreeReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 602
 603                 rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i),
 604                     CarsModel.newCarEntry("car" + i, Uint64.valueOf(20000)));
 605
 606                 futures.add(rwTx.commit());
 607             }
 608
 609             for (final ListenableFuture<?> f : futures) {
 610                 f.get(5, TimeUnit.SECONDS);
 611             }
 612
 613             final Optional<NormalizedNode> optional = txChain.newReadOnlyTransaction()
 614                     .read(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
 615             assertTrue("isPresent", optional.isPresent());
 616             assertEquals("# cars", numCars, ((Collection<?>) optional.get().body()).size());
 617
 618             txChain.close();
 619
 620             broker.close();
 621         }
 622     }
 623
 624     @Test
 625     public void testCreateChainedTransactionAfterEmptyTxReadied() throws Exception {
 626         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 627         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 628             testParameter, "testCreateChainedTransactionAfterEmptyTxReadied", "test-1")) {
 629
 630             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 631
 632             final DOMStoreReadWriteTransaction rwTx1 = txChain.newReadWriteTransaction();
 633
 634             rwTx1.ready();
 635
 636             final DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction();
 637
 638             final Optional<NormalizedNode> optional = rwTx2.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 639             assertFalse("isPresent", optional.isPresent());
 640
 641             txChain.close();
 642         }
 643     }
 644
 645     @Test
 646     public void testCreateChainedTransactionWhenPreviousNotReady() throws Exception {
 647         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 648         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 649             testParameter, "testCreateChainedTransactionWhenPreviousNotReady", "test-1")) {
 650
 651             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 652
 653             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 654             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 655
 656             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 657
 658             // Try to create another Tx of each type - each should fail
 659             // b/c the previous Tx wasn't
 660             // readied.
 661             testKit.assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
 662         }
 663     }
 664
 665     @Test
 666     public void testCreateChainedTransactionAfterClose() throws Exception {
 667         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 668         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 669             testParameter, "testCreateChainedTransactionAfterClose", "test-1")) {
 670
 671             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 672             txChain.close();
 673
 674             // Try to create another Tx of each type - should fail b/c
 675             // the previous Tx was closed.
 676             testKit.assertExceptionOnTxChainCreates(txChain, DOMTransactionChainClosedException.class);
 677         }
 678     }
 679
 680     @Test
 681     public void testChainWithReadOnlyTxAfterPreviousReady() throws Exception {
 682         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 683         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 684             testParameter, "testChainWithReadOnlyTxAfterPreviousReady", "test-1")) {
 685
 686             final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 687
 688             // Create a write tx and submit.
 689             final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 690             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 691             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 692
 693             // Create read-only tx's and issue a read.
 694             FluentFuture<Optional<NormalizedNode>> readFuture1 = txChain
 695                     .newReadOnlyTransaction().read(TestModel.TEST_PATH);
 696
 697             FluentFuture<Optional<NormalizedNode>> readFuture2 = txChain
 698                     .newReadOnlyTransaction().read(TestModel.TEST_PATH);
 699
 700             // Create another write tx and issue the write.
 701             DOMStoreWriteTransaction writeTx2 = txChain.newWriteOnlyTransaction();
 702             writeTx2.write(TestModel.OUTER_LIST_PATH,
 703                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
 704                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
 705                 .build());
 706
 707             // Ensure the reads succeed.
 708
 709             assertTrue("isPresent", readFuture1.get(5, TimeUnit.SECONDS).isPresent());
 710             assertTrue("isPresent", readFuture2.get(5, TimeUnit.SECONDS).isPresent());
 711
 712             // Ensure the writes succeed.
 713             DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
 714
 715             testKit.doCommit(cohort1);
 716             testKit.doCommit(cohort2);
 717
 718             assertTrue("isPresent", txChain.newReadOnlyTransaction().read(TestModel.OUTER_LIST_PATH)
 719                 .get(5, TimeUnit.SECONDS).isPresent());
 720         }
 721     }
 722
 723     @Test
 724     public void testChainedTransactionFailureWithSingleShard() throws Exception {
 725         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 726         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 727             testParameter, "testChainedTransactionFailureWithSingleShard", "cars-1")) {
 728
 729             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 730                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
 731                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
 732                 MoreExecutors.directExecutor());
 733
 734             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
 735             final DOMTransactionChain txChain = broker.createTransactionChain(listener);
 736
 737             final DOMDataTreeReadWriteTransaction writeTx = txChain.newReadWriteTransaction();
 738
 739             writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
 740                 PeopleModel.emptyContainer());
 741
 742             final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
 743                     .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
 744                     .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 745
 746             writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
 747
 748             try {
 749                 writeTx.commit().get(5, TimeUnit.SECONDS);
 750                 fail("Expected TransactionCommitFailedException");
 751             } catch (final ExecutionException e) {
 752                 // Expected
 753             }
 754
 755             verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
 756                 any(Throwable.class));
 757
 758             txChain.close();
 759             broker.close();
 760         }
 761     }
 762
 763     @Test
 764     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
 765         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 766         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 767             testParameter, "testChainedTransactionFailureWithMultipleShards", "cars-1", "people-1")) {
 768
 769             final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 770                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder()
 771                 .put(LogicalDatastoreType.CONFIGURATION, dataStore).build(),
 772                 MoreExecutors.directExecutor());
 773
 774             final DOMTransactionChainListener listener = Mockito.mock(DOMTransactionChainListener.class);
 775             final DOMTransactionChain txChain = broker.createTransactionChain(listener);
 776
 777             final DOMDataTreeWriteTransaction writeTx = txChain.newReadWriteTransaction();
 778
 779             writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH,
 780                 PeopleModel.emptyContainer());
 781
 782             final ContainerNode invalidData = ImmutableContainerNodeBuilder.create()
 783                     .withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
 784                     .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 785
 786             writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
 787
 788             // Note that merge will validate the data and fail but put
 789             // succeeds b/c deep validation is not
 790             // done for put for performance reasons.
 791             try {
 792                 writeTx.commit().get(5, TimeUnit.SECONDS);
 793                 fail("Expected TransactionCommitFailedException");
 794             } catch (final ExecutionException e) {
 795                 // Expected
 796             }
 797
 798             verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx),
 799                 any(Throwable.class));
 800
 801             txChain.close();
 802             broker.close();
 803         }
 804     }
 805
 806     @Test
 807     public void testDataTreeChangeListenerRegistration() throws Exception {
 808         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 809         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 810             testParameter, "testDataTreeChangeListenerRegistration", "test-1")) {
 811
 812             testKit.testWriteTransaction(dataStore, TestModel.TEST_PATH,
 813                 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 814
 815             final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
 816
 817             ListenerRegistration<MockDataTreeChangeListener> listenerReg = dataStore
 818                     .registerTreeChangeListener(TestModel.TEST_PATH, listener);
 819
 820             assertNotNull("registerTreeChangeListener returned null", listenerReg);
 821
 822             IntegrationTestKit.verifyShardState(dataStore, "test-1",
 823                 state -> assertEquals("getTreeChangeListenerActors", 1,
 824                         state.getTreeChangeListenerActors().size()));
 825
 826             // Wait for the initial notification
 827             listener.waitForChangeEvents(TestModel.TEST_PATH);
 828             listener.reset(2);
 829
 830             // Write 2 updates.
 831             testKit.testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
 832                 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME)
 833                 .withChild(ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 42))
 834                 .build());
 835
 836             YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
 837                     .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
 838             testKit.testWriteTransaction(dataStore, listPath,
 839                 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
 840
 841             // Wait for the 2 updates.
 842             listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
 843             listenerReg.close();
 844
 845             IntegrationTestKit.verifyShardState(dataStore, "test-1",
 846                 state -> assertEquals("getTreeChangeListenerActors", 0,
 847                     state.getTreeChangeListenerActors().size()));
 848
 849             testKit.testWriteTransaction(dataStore,
 850                 YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH)
 851                 .nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
 852                 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
 853
 854             listener.expectNoMoreChanges("Received unexpected change after close");
 855         }
 856     }
 857
 858     @Test
 859     public void testRestoreFromDatastoreSnapshot() throws Exception {
 860         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder);
 861         final String name = "transactionIntegrationTest";
 862
 863         final ContainerNode carsNode = CarsModel.newCarsNode(
 864             CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", Uint64.valueOf(20000)),
 865                 CarsModel.newCarEntry("sportage", Uint64.valueOf(30000))));
 866
 867         DataTree dataTree = new InMemoryDataTreeFactory().create(
 868             DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full());
 869         AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
 870         NormalizedNode root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.empty());
 871
 872         final Snapshot carsSnapshot = Snapshot.create(
 873             new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
 874             Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
 875
 876         dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL,
 877             SchemaContextHelper.full());
 878
 879         final NormalizedNode peopleNode = PeopleModel.create();
 880         AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode);
 881
 882         root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.empty());
 883
 884         final Snapshot peopleSnapshot = Snapshot.create(
 885             new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)),
 886             Collections.emptyList(), 2, 1, 2, 1, 1, "member-1", null);
 887
 888         testKit.restoreFromSnapshot = new DatastoreSnapshot(name, null, Arrays.asList(
 889             new DatastoreSnapshot.ShardSnapshot("cars", carsSnapshot),
 890             new DatastoreSnapshot.ShardSnapshot("people", peopleSnapshot)));
 891
 892         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 893             testParameter, name, "module-shards-member1.conf", true, "cars", "people")) {
 894
 895             final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 896
 897             // two reads
 898             Optional<NormalizedNode> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
 899             assertTrue("isPresent", optional.isPresent());
 900             assertEquals("Data node", carsNode, optional.get());
 901
 902             optional = readTx.read(PeopleModel.BASE_PATH).get(5, TimeUnit.SECONDS);
 903             assertTrue("isPresent", optional.isPresent());
 904             assertEquals("Data node", peopleNode, optional.get());
 905         }
 906     }
 907
 908     @Test
 909     public void testSnapshotOnRootOverwrite() throws Exception {
 910         if (!DistributedDataStore.class.isAssignableFrom(testParameter)) {
 911             // FIXME: ClientBackedDatastore does not have stable indexes/term, the snapshot index seems to fluctuate
 912             return;
 913         }
 914
 915         final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(),
 916                 datastoreContextBuilder.snapshotOnRootOverwrite(true));
 917         try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(
 918                 testParameter, "testRootOverwrite", "module-shards-default-cars-member1.conf",
 919                 true, "cars", "default")) {
 920
 921             ContainerNode rootNode = ImmutableContainerNodeBuilder.create()
 922                     .withNodeIdentifier(YangInstanceIdentifier.NodeIdentifier.create(SchemaContext.NAME))
 923                     .withChild(CarsModel.create())
 924                     .build();
 925
 926             testKit.testWriteTransaction(dataStore, YangInstanceIdentifier.empty(), rootNode);
 927             IntegrationTestKit.verifyShardState(dataStore, "cars",
 928                 state -> assertEquals(1, state.getSnapshotIndex()));
 929
 930             // root has been written expect snapshot at index 0
 931             verifySnapshot("member-1-shard-cars-testRootOverwrite", 1, 1);
 932
 933             for (int i = 0; i < 10; i++) {
 934                 testKit.testWriteTransaction(dataStore, CarsModel.newCarPath("car " + i),
 935                     CarsModel.newCarEntry("car " + i, Uint64.ONE));
 936             }
 937
 938             // fake snapshot causes the snapshotIndex to move
 939             IntegrationTestKit.verifyShardState(dataStore, "cars",
 940                 state -> assertEquals(10, state.getSnapshotIndex()));
 941
 942             // however the real snapshot still has not changed and was taken at index 0
 943             verifySnapshot("member-1-shard-cars-testRootOverwrite", 1, 1);
 944
 945             // root overwrite so expect a snapshot
 946             testKit.testWriteTransaction(dataStore, YangInstanceIdentifier.empty(), rootNode);
 947
 948             // this was a real snapshot so everything should be in it(1 + 10 + 1)
 949             IntegrationTestKit.verifyShardState(dataStore, "cars",
 950                 state -> assertEquals(12, state.getSnapshotIndex()));
 951
 952             verifySnapshot("member-1-shard-cars-testRootOverwrite", 12, 1);
 953         }
 954     }
 955
 956     private static void verifySnapshot(final String persistenceId, final long lastAppliedIndex,
 957             final long lastAppliedTerm) {
 958         await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
 959                 List<Snapshot> snap = InMemorySnapshotStore.getSnapshots(persistenceId, Snapshot.class);
 960                 assertEquals(1, snap.size());
 961                 assertEquals(lastAppliedIndex, snap.get(0).getLastAppliedIndex());
 962                 assertEquals(lastAppliedTerm, snap.get(0).getLastAppliedTerm());
 963             }
 964         );
 965     }
 966 }