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