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