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

   1 /*
   2  * Copyright (c) 2015, 2017 Brocade Communications 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.hamcrest.Matchers.equalTo;
  14 import static org.junit.Assert.assertEquals;
  15 import static org.junit.Assert.assertFalse;
  16 import static org.junit.Assert.assertNotNull;
  17 import static org.junit.Assert.assertThrows;
  18 import static org.junit.Assert.assertTrue;
  19 import static org.junit.Assume.assumeTrue;
  20 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  21 import static org.mockito.ArgumentMatchers.any;
  22 import static org.mockito.ArgumentMatchers.anyString;
  23 import static org.mockito.Mockito.doAnswer;
  24 import static org.mockito.Mockito.mock;
  25 import static org.mockito.Mockito.timeout;
  26 import static org.mockito.Mockito.verify;
  27
  28 import akka.actor.ActorRef;
  29 import akka.actor.ActorSelection;
  30 import akka.actor.ActorSystem;
  31 import akka.actor.Address;
  32 import akka.actor.AddressFromURIString;
  33 import akka.cluster.Cluster;
  34 import akka.cluster.Member;
  35 import akka.dispatch.Futures;
  36 import akka.pattern.Patterns;
  37 import akka.testkit.javadsl.TestKit;
  38 import com.google.common.base.Stopwatch;
  39 import com.google.common.base.Throwables;
  40 import com.google.common.collect.ImmutableMap;
  41 import com.google.common.util.concurrent.FutureCallback;
  42 import com.google.common.util.concurrent.ListenableFuture;
  43 import com.google.common.util.concurrent.MoreExecutors;
  44 import com.google.common.util.concurrent.Uninterruptibles;
  45 import com.typesafe.config.ConfigFactory;
  46 import java.util.Arrays;
  47 import java.util.Collection;
  48 import java.util.Collections;
  49 import java.util.LinkedList;
  50 import java.util.List;
  51 import java.util.Map;
  52 import java.util.Optional;
  53 import java.util.concurrent.ExecutionException;
  54 import java.util.concurrent.ExecutorService;
  55 import java.util.concurrent.Executors;
  56 import java.util.concurrent.TimeUnit;
  57 import java.util.concurrent.atomic.AtomicBoolean;
  58 import java.util.concurrent.atomic.AtomicLong;
  59 import org.junit.After;
  60 import org.junit.Before;
  61 import org.junit.Test;
  62 import org.junit.runner.RunWith;
  63 import org.junit.runners.Parameterized;
  64 import org.junit.runners.Parameterized.Parameter;
  65 import org.junit.runners.Parameterized.Parameters;
  66 import org.mockito.stubbing.Answer;
  67 import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
  68 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  69 import org.opendaylight.controller.cluster.databroker.ClientBackedDataStore;
  70 import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
  71 import org.opendaylight.controller.cluster.databroker.TestClientBackedDataStore;
  72 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  73 import org.opendaylight.controller.cluster.datastore.TestShard.RequestFrontendMetadata;
  74 import org.opendaylight.controller.cluster.datastore.TestShard.StartDropMessages;
  75 import org.opendaylight.controller.cluster.datastore.TestShard.StopDropMessages;
  76 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  77 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
  78 import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
  79 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
  80 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
  81 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  82 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  83 import org.opendaylight.controller.cluster.datastore.persisted.FrontendClientMetadata;
  84 import org.opendaylight.controller.cluster.datastore.persisted.FrontendShardDataTreeSnapshotMetadata;
  85 import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
  86 import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
  87 import org.opendaylight.controller.cluster.datastore.utils.UnsignedLongBitmap;
  88 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
  89 import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
  90 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
  91 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
  92 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  93 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  94 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
  95 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  96 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
  97 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  98 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  99 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
 100 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
 101 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
 102 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
 103 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
 104 import org.opendaylight.mdsal.common.api.OptimisticLockFailedException;
 105 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
 106 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
 107 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
 108 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
 109 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
 110 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
 111 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
 112 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
 113 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
 114 import org.opendaylight.yangtools.yang.common.Uint64;
 115 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 116 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
 117 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 118 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
 119 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 120 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
 121 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
 122 import org.opendaylight.yangtools.yang.data.tree.api.ConflictingModificationAppliedException;
 123 import org.opendaylight.yangtools.yang.data.tree.api.DataTree;
 124 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeConfiguration;
 125 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
 126 import org.opendaylight.yangtools.yang.data.tree.impl.di.InMemoryDataTreeFactory;
 127 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
 128 import scala.collection.Set;
 129 import scala.concurrent.Await;
 130 import scala.concurrent.Future;
 131 import scala.concurrent.duration.FiniteDuration;
 132
 133 /**
 134  * End-to-end distributed data store tests that exercise remote shards and transactions.
 135  *
 136  * @author Thomas Pantelis
 137  */
 138 @RunWith(Parameterized.class)
 139 public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest {
 140
 141     @Parameters(name = "{0}")
 142     public static Collection<Object[]> data() {
 143         return Arrays.asList(new Object[][] {
 144                 { TestClientBackedDataStore.class, 12 }
 145         });
 146     }
 147
 148     @Parameter(0)
 149     public Class<? extends ClientBackedDataStore> testParameter;
 150     @Parameter(1)
 151     public int commitTimeout;
 152
 153     private static final String[] CARS_AND_PEOPLE = {"cars", "people"};
 154     private static final String[] CARS = {"cars"};
 155
 156     private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse(
 157             "akka://cluster-test@127.0.0.1:2558");
 158     private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse(
 159             "akka://cluster-test@127.0.0.1:2559");
 160
 161     private static final String MODULE_SHARDS_CARS_ONLY_1_2 = "module-shards-cars-member-1-and-2.conf";
 162     private static final String MODULE_SHARDS_CARS_PEOPLE_1_2 = "module-shards-member1-and-2.conf";
 163     private static final String MODULE_SHARDS_CARS_PEOPLE_1_2_3 = "module-shards-member1-and-2-and-3.conf";
 164     private static final String MODULE_SHARDS_CARS_1_2_3 = "module-shards-cars-member-1-and-2-and-3.conf";
 165
 166     private ActorSystem leaderSystem;
 167     private ActorSystem followerSystem;
 168     private ActorSystem follower2System;
 169
 170     private final DatastoreContext.Builder leaderDatastoreContextBuilder =
 171             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
 172
 173     private final DatastoreContext.Builder followerDatastoreContextBuilder =
 174             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5)
 175                 .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
 176     private final TransactionIdentifier tx1 = nextTransactionId();
 177     private final TransactionIdentifier tx2 = nextTransactionId();
 178
 179     private ClientBackedDataStore followerDistributedDataStore;
 180     private ClientBackedDataStore leaderDistributedDataStore;
 181     private IntegrationTestKit followerTestKit;
 182     private IntegrationTestKit leaderTestKit;
 183
 184     @Before
 185     public void setUp() {
 186         InMemoryJournal.clear();
 187         InMemorySnapshotStore.clear();
 188
 189         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 190         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 191
 192         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
 193         Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
 194
 195         follower2System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member3"));
 196         Cluster.get(follower2System).join(MEMBER_1_ADDRESS);
 197     }
 198
 199     @After
 200     public void tearDown() {
 201         if (followerDistributedDataStore != null) {
 202             followerDistributedDataStore.close();
 203         }
 204         if (leaderDistributedDataStore != null) {
 205             leaderDistributedDataStore.close();
 206         }
 207
 208         TestKit.shutdownActorSystem(leaderSystem, true);
 209         TestKit.shutdownActorSystem(followerSystem, true);
 210         TestKit.shutdownActorSystem(follower2System,true);
 211
 212         InMemoryJournal.clear();
 213         InMemorySnapshotStore.clear();
 214     }
 215
 216     private void initDatastoresWithCars(final String type) throws Exception {
 217         initDatastores(type, MODULE_SHARDS_CARS_ONLY_1_2, CARS);
 218     }
 219
 220     private void initDatastoresWithCarsAndPeople(final String type) throws Exception {
 221         initDatastores(type, MODULE_SHARDS_CARS_PEOPLE_1_2, CARS_AND_PEOPLE);
 222     }
 223
 224     private void initDatastores(final String type, final String moduleShardsConfig, final String[] shards)
 225             throws Exception {
 226         initDatastores(type, moduleShardsConfig, shards, leaderDatastoreContextBuilder,
 227                 followerDatastoreContextBuilder);
 228     }
 229
 230     private void initDatastores(final String type, final String moduleShardsConfig, final String[] shards,
 231             final DatastoreContext.Builder leaderBuilder, final DatastoreContext.Builder followerBuilder)
 232                     throws Exception {
 233         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderBuilder, commitTimeout);
 234
 235         leaderDistributedDataStore = leaderTestKit.setupDataStore(testParameter, type, moduleShardsConfig, false,
 236             shards);
 237
 238         followerTestKit = new IntegrationTestKit(followerSystem, followerBuilder, commitTimeout);
 239         followerDistributedDataStore = followerTestKit.setupDataStore(
 240                 testParameter, type, moduleShardsConfig, false, shards);
 241
 242         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(), shards);
 243
 244         leaderTestKit.waitForMembersUp("member-2");
 245         followerTestKit.waitForMembersUp("member-1");
 246     }
 247
 248     private static void verifyCars(final DOMStoreReadTransaction readTx, final MapEntryNode... entries)
 249             throws Exception {
 250         assertEquals("Car list node",
 251             Optional.of(ImmutableNodes.mapNodeBuilder(CarsModel.CAR_QNAME).withValue(Arrays.asList(entries)).build()),
 252             readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS));
 253     }
 254
 255     private static void verifyNode(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path,
 256             final NormalizedNode expNode) throws Exception {
 257         assertEquals(Optional.of(expNode), readTx.read(path).get(5, TimeUnit.SECONDS));
 258     }
 259
 260     private static void verifyExists(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path)
 261             throws Exception {
 262         assertEquals("exists", Boolean.TRUE, readTx.exists(path).get(5, TimeUnit.SECONDS));
 263     }
 264
 265     @Test
 266     public void testWriteTransactionWithSingleShard() throws Exception {
 267         final String testName = "testWriteTransactionWithSingleShard";
 268         initDatastoresWithCars(testName);
 269
 270         final String followerCarShardName = "member-2-shard-cars-" + testName;
 271
 272         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 273         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 274
 275         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 276         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 277
 278         final MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 279         final YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 280         writeTx.merge(car1Path, car1);
 281
 282         final MapEntryNode car2 = CarsModel.newCarEntry("sportage", Uint64.valueOf(25000));
 283         final YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
 284         writeTx.merge(car2Path, car2);
 285
 286         followerTestKit.doCommit(writeTx.ready());
 287
 288         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 289
 290         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 291
 292         // Test delete
 293
 294         writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 295
 296         writeTx.delete(car1Path);
 297
 298         followerTestKit.doCommit(writeTx.ready());
 299
 300         verifyExists(followerDistributedDataStore.newReadOnlyTransaction(), car2Path);
 301
 302         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
 303
 304         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car2);
 305
 306         // Re-instate the follower member 2 as a single-node to verify replication and recovery.
 307
 308         // The following is a bit tricky. Before we reinstate the follower we need to ensure it has persisted and
 309         // applied and all the log entries from the leader. Since we've verified the car data above we know that
 310         // all the transactions have been applied on the leader so we first read and capture its lastAppliedIndex.
 311         final AtomicLong leaderLastAppliedIndex = new AtomicLong();
 312         IntegrationTestKit.verifyShardState(leaderDistributedDataStore, CARS[0],
 313             state -> leaderLastAppliedIndex.set(state.getLastApplied()));
 314
 315         // Now we need to make sure the follower has persisted the leader's lastAppliedIndex via ApplyJournalEntries.
 316         // However we don't know exactly how many ApplyJournalEntries messages there will be as it can differ between
 317         // the tell-based and ask-based front-ends. For ask-based there will be exactly 2 ApplyJournalEntries but
 318         // tell-based persists additional payloads which could be replicated and applied in a batch resulting in
 319         // either 2 or 3 ApplyJournalEntries. To handle this we read the follower's persisted ApplyJournalEntries
 320         // until we find the one that encompasses the leader's lastAppliedIndex.
 321         Stopwatch sw = Stopwatch.createStarted();
 322         boolean done = false;
 323         while (!done) {
 324             final List<ApplyJournalEntries> entries = InMemoryJournal.get(followerCarShardName,
 325                     ApplyJournalEntries.class);
 326             for (ApplyJournalEntries aje: entries) {
 327                 if (aje.getToIndex() >= leaderLastAppliedIndex.get()) {
 328                     done = true;
 329                     break;
 330                 }
 331             }
 332
 333             assertTrue("Follower did not persist ApplyJournalEntries containing leader's lastAppliedIndex "
 334                     + leaderLastAppliedIndex + ". Entries persisted: " + entries, sw.elapsed(TimeUnit.SECONDS) <= 5);
 335
 336             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 337         }
 338
 339         TestKit.shutdownActorSystem(leaderSystem, true);
 340         TestKit.shutdownActorSystem(followerSystem, true);
 341
 342         final ActorSystem newSystem = newActorSystem("reinstated-member2", "Member2");
 343
 344         try (var member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder, commitTimeout)
 345                 .setupDataStore(testParameter, testName, "module-shards-member2", true, CARS)) {
 346             verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
 347         }
 348     }
 349
 350     @Test
 351     public void testSingleTransactionsWritesInQuickSuccession() throws Exception {
 352         initDatastoresWithCars("testSingleTransactionsWritesInQuickSuccession");
 353
 354         final DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
 355
 356         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 357         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 358         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 359         followerTestKit.doCommit(writeTx.ready());
 360
 361         int numCars = 5;
 362         for (int i = 0; i < numCars; i++) {
 363             writeTx = txChain.newWriteOnlyTransaction();
 364             writeTx.write(CarsModel.newCarPath("car" + i), CarsModel.newCarEntry("car" + i, Uint64.valueOf(20000)));
 365             followerTestKit.doCommit(writeTx.ready());
 366
 367             try (var tx = txChain.newReadOnlyTransaction()) {
 368                 tx.read(CarsModel.BASE_PATH).get();
 369             }
 370         }
 371
 372         // wait to let the shard catch up with purged
 373         await("Range set leak test").atMost(5, TimeUnit.SECONDS)
 374             .pollInterval(500, TimeUnit.MILLISECONDS)
 375             .untilAsserted(() -> {
 376                 final var localShard = leaderDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
 377                 final var frontendMetadata =
 378                     (FrontendShardDataTreeSnapshotMetadata) leaderDistributedDataStore.getActorUtils()
 379                     .executeOperation(localShard, new RequestFrontendMetadata());
 380
 381                 assertClientMetadata(frontendMetadata.getClients().get(0), numCars * 2);
 382             });
 383
 384         try (var tx = txChain.newReadOnlyTransaction()) {
 385             final var body = assertInstanceOf(Collection.class,
 386                 tx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS).orElseThrow().body());
 387             assertEquals(numCars, ((Collection<?>) body).size());
 388         }
 389     }
 390
 391     private static void assertClientMetadata(final FrontendClientMetadata clientMeta, final long lastPurged) {
 392         final var iterator = clientMeta.getCurrentHistories().iterator();
 393         var metadata = iterator.next();
 394         while (iterator.hasNext() && metadata.getHistoryId() != 1) {
 395             metadata = iterator.next();
 396         }
 397
 398         assertEquals(UnsignedLongBitmap.of(), metadata.getClosedTransactions());
 399         assertEquals("[[0.." + lastPurged + "]]", metadata.getPurgedTransactions().ranges().toString());
 400     }
 401
 402     @Test
 403     public void testCloseTransactionMetadataLeak() throws Exception {
 404         initDatastoresWithCars("testCloseTransactionMetadataLeak");
 405
 406         final var txChain = followerDistributedDataStore.createTransactionChain();
 407
 408         var writeTx = txChain.newWriteOnlyTransaction();
 409         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 410         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 411         followerTestKit.doCommit(writeTx.ready());
 412
 413         int numCars = 5;
 414         for (int i = 0; i < numCars; i++) {
 415             try (var tx = txChain.newWriteOnlyTransaction()) {
 416                 // Empty on purpose
 417             }
 418
 419             try (var tx = txChain.newReadOnlyTransaction()) {
 420                 tx.read(CarsModel.BASE_PATH).get();
 421             }
 422         }
 423
 424         // wait to let the shard catch up with purged
 425         await("wait for purges to settle").atMost(5, TimeUnit.SECONDS)
 426             .pollInterval(500, TimeUnit.MILLISECONDS)
 427             .untilAsserted(() -> {
 428                 final var localShard = leaderDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
 429                 final var frontendMetadata =
 430                     (FrontendShardDataTreeSnapshotMetadata) leaderDistributedDataStore.getActorUtils()
 431                     .executeOperation(localShard, new RequestFrontendMetadata());
 432
 433                 assertClientMetadata(frontendMetadata.getClients().get(0), numCars * 2);
 434             });
 435     }
 436
 437     @Test
 438     public void testReadWriteTransactionWithSingleShard() throws Exception {
 439         initDatastoresWithCars("testReadWriteTransactionWithSingleShard");
 440
 441         final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
 442         assertNotNull("newReadWriteTransaction returned null", rwTx);
 443
 444         rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 445         rwTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 446
 447         final MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 448         rwTx.merge(CarsModel.newCarPath("optima"), car1);
 449
 450         verifyCars(rwTx, car1);
 451
 452         final MapEntryNode car2 = CarsModel.newCarEntry("sportage", Uint64.valueOf(25000));
 453         final YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
 454         rwTx.merge(car2Path, car2);
 455
 456         verifyExists(rwTx, car2Path);
 457
 458         followerTestKit.doCommit(rwTx.ready());
 459
 460         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 461     }
 462
 463     @Test
 464     public void testWriteTransactionWithMultipleShards() throws Exception {
 465         initDatastoresWithCarsAndPeople("testWriteTransactionWithMultipleShards");
 466
 467         final DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 468         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 469
 470         final YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
 471         final NormalizedNode carsNode = CarsModel.emptyContainer();
 472         writeTx.write(carsPath, carsNode);
 473
 474         final YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
 475         final NormalizedNode peopleNode = PeopleModel.emptyContainer();
 476         writeTx.write(peoplePath, peopleNode);
 477
 478         followerTestKit.doCommit(writeTx.ready());
 479
 480         final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 481
 482         verifyNode(readTx, carsPath, carsNode);
 483         verifyNode(readTx, peoplePath, peopleNode);
 484     }
 485
 486     @Test
 487     public void testReadWriteTransactionWithMultipleShards() throws Exception {
 488         initDatastoresWithCarsAndPeople("testReadWriteTransactionWithMultipleShards");
 489
 490         final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
 491         assertNotNull("newReadWriteTransaction returned null", rwTx);
 492
 493         final YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
 494         final NormalizedNode carsNode = CarsModel.emptyContainer();
 495         rwTx.write(carsPath, carsNode);
 496
 497         final YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
 498         final NormalizedNode peopleNode = PeopleModel.emptyContainer();
 499         rwTx.write(peoplePath, peopleNode);
 500
 501         followerTestKit.doCommit(rwTx.ready());
 502
 503         final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 504
 505         verifyNode(readTx, carsPath, carsNode);
 506         verifyNode(readTx, peoplePath, peopleNode);
 507     }
 508
 509     @Test
 510     public void testTransactionChainWithSingleShard() throws Exception {
 511         initDatastoresWithCars("testTransactionChainWithSingleShard");
 512
 513         final DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
 514
 515         // Add the top-level cars container with write-only.
 516
 517         final DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 518         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 519
 520         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 521
 522         final DOMStoreThreePhaseCommitCohort writeTxReady = writeTx.ready();
 523
 524         // Verify the top-level cars container with read-only.
 525
 526         verifyNode(txChain.newReadOnlyTransaction(), CarsModel.BASE_PATH, CarsModel.emptyContainer());
 527
 528         // Perform car operations with read-write.
 529
 530         final DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
 531
 532         verifyNode(rwTx, CarsModel.BASE_PATH, CarsModel.emptyContainer());
 533
 534         rwTx.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 535
 536         final MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 537         final YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 538         rwTx.write(car1Path, car1);
 539
 540         verifyExists(rwTx, car1Path);
 541
 542         verifyCars(rwTx, car1);
 543
 544         final MapEntryNode car2 = CarsModel.newCarEntry("sportage", Uint64.valueOf(25000));
 545         rwTx.merge(CarsModel.newCarPath("sportage"), car2);
 546
 547         rwTx.delete(car1Path);
 548
 549         followerTestKit.doCommit(writeTxReady);
 550
 551         followerTestKit.doCommit(rwTx.ready());
 552
 553         txChain.close();
 554
 555         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
 556     }
 557
 558     @Test
 559     public void testTransactionChainWithMultipleShards() throws Exception {
 560         initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards");
 561
 562         final DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
 563
 564         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 565         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 566
 567         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 568         writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 569
 570         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 571         writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
 572
 573         followerTestKit.doCommit(writeTx.ready());
 574
 575         final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
 576
 577         final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 578         final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
 579         readWriteTx.write(carPath, car);
 580
 581         final MapEntryNode person = PeopleModel.newPersonEntry("jack");
 582         final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
 583         readWriteTx.merge(personPath, person);
 584
 585         assertEquals(Optional.of(car), readWriteTx.read(carPath).get(5, TimeUnit.SECONDS));
 586         assertEquals(Optional.of(person), readWriteTx.read(personPath).get(5, TimeUnit.SECONDS));
 587
 588         final DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
 589
 590         writeTx = txChain.newWriteOnlyTransaction();
 591
 592         writeTx.delete(personPath);
 593
 594         final DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
 595
 596         followerTestKit.doCommit(cohort2);
 597         followerTestKit.doCommit(cohort3);
 598
 599         txChain.close();
 600
 601         final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
 602         verifyCars(readTx, car);
 603
 604         assertEquals(Optional.empty(), readTx.read(personPath).get(5, TimeUnit.SECONDS));
 605     }
 606
 607     @Test
 608     public void testChainedTransactionFailureWithSingleShard() throws Exception {
 609         initDatastoresWithCars("testChainedTransactionFailureWithSingleShard");
 610
 611         final ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
 612                 ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
 613                         LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(),
 614                         MoreExecutors.directExecutor());
 615
 616         final var listener = mock(FutureCallback.class);
 617         final DOMTransactionChain txChain = broker.createTransactionChain();
 618         txChain.addCallback(listener);
 619
 620         final DOMDataTreeWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 621
 622         writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, Builders.containerBuilder()
 623             .withNodeIdentifier(new NodeIdentifier(CarsModel.BASE_QNAME))
 624             .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk"))
 625             .build());
 626
 627         final var ex = assertThrows(ExecutionException.class, () -> writeTx.commit().get(5, TimeUnit.SECONDS));
 628         assertInstanceOf(TransactionCommitFailedException.class, ex.getCause());
 629
 630         verify(listener, timeout(5000)).onFailure(any());
 631
 632         txChain.close();
 633         broker.close();
 634     }
 635
 636     @Test
 637     public void testChainedTransactionFailureWithMultipleShards() throws Exception {
 638         initDatastoresWithCarsAndPeople("testChainedTransactionFailureWithMultipleShards");
 639
 640         try (var broker = new ConcurrentDOMDataBroker(
 641             Map.of(LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore), MoreExecutors.directExecutor())) {
 642
 643             final var listener = mock(FutureCallback.class);
 644             final DOMTransactionChain txChain = broker.createTransactionChain();
 645             txChain.addCallback(listener);
 646
 647             final DOMDataTreeWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 648
 649             writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 650
 651             // Note that merge will validate the data and fail but put succeeds b/c deep validation is not
 652             // done for put for performance reasons.
 653             writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, Builders.containerBuilder()
 654                 .withNodeIdentifier(new NodeIdentifier(CarsModel.BASE_QNAME))
 655                 .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk"))
 656                 .build());
 657
 658             final var ex = assertThrows(ExecutionException.class, () -> writeTx.commit().get(5, TimeUnit.SECONDS))
 659                 .getCause();
 660             assertThat(ex, instanceOf(TransactionCommitFailedException.class));
 661
 662             verify(listener, timeout(5000)).onFailure(any());
 663
 664             txChain.close();
 665         }
 666     }
 667
 668     @Test
 669     public void testSingleShardTransactionsWithLeaderChanges() throws Exception {
 670         followerDatastoreContextBuilder.backendAlivenessTimerIntervalInSeconds(2);
 671         final String testName = "testSingleShardTransactionsWithLeaderChanges";
 672         initDatastoresWithCars(testName);
 673
 674         final String followerCarShardName = "member-2-shard-cars-" + testName;
 675         InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class);
 676
 677         // Write top-level car container from the follower so it uses a remote Tx.
 678
 679         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 680
 681         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 682         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 683
 684         followerTestKit.doCommit(writeTx.ready());
 685
 686         InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
 687
 688         // Switch the leader to the follower
 689
 690         sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
 691                 .shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
 692
 693         TestKit.shutdownActorSystem(leaderSystem, true);
 694         Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
 695
 696         followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorUtils(), CARS);
 697
 698         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 699         Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
 700
 701         final DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder()
 702                 .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
 703         IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder, commitTimeout);
 704
 705         try (var ds = newMember1TestKit.setupDataStore(testParameter, testName, MODULE_SHARDS_CARS_ONLY_1_2, false,
 706             CARS)) {
 707
 708             followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), CARS);
 709
 710             // Write a car entry to the new leader - should switch to local Tx
 711
 712             writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 713
 714             MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 715             YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
 716             writeTx.merge(car1Path, car1);
 717
 718             followerTestKit.doCommit(writeTx.ready());
 719
 720             verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1);
 721         }
 722     }
 723
 724     @Test
 725     public void testReadyLocalTransactionForwardedToLeader() throws Exception {
 726         initDatastoresWithCars("testReadyLocalTransactionForwardedToLeader");
 727         followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), "cars");
 728
 729         final Optional<ActorRef> carsFollowerShard =
 730                 followerDistributedDataStore.getActorUtils().findLocalShard("cars");
 731         assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
 732
 733         final DataTree dataTree = new InMemoryDataTreeFactory().create(
 734             DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full());
 735
 736         // Send a tx with immediate commit.
 737
 738         DataTreeModification modification = dataTree.takeSnapshot().newModification();
 739         new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification);
 740         new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification);
 741
 742         final MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 743         new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification);
 744         modification.ready();
 745
 746         ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(tx1 , modification, true, Optional.empty());
 747
 748         carsFollowerShard.orElseThrow().tell(readyLocal, followerTestKit.getRef());
 749         Object resp = followerTestKit.expectMsgClass(Object.class);
 750         if (resp instanceof akka.actor.Status.Failure) {
 751             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
 752         }
 753
 754         assertEquals("Response type", CommitTransactionReply.class, resp.getClass());
 755
 756         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1);
 757
 758         // Send another tx without immediate commit.
 759
 760         modification = dataTree.takeSnapshot().newModification();
 761         MapEntryNode car2 = CarsModel.newCarEntry("sportage", Uint64.valueOf(30000));
 762         new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification);
 763         modification.ready();
 764
 765         readyLocal = new ReadyLocalTransaction(tx2 , modification, false, Optional.empty());
 766
 767         carsFollowerShard.orElseThrow().tell(readyLocal, followerTestKit.getRef());
 768         resp = followerTestKit.expectMsgClass(Object.class);
 769         if (resp instanceof akka.actor.Status.Failure) {
 770             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
 771         }
 772
 773         assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
 774
 775         final ActorSelection txActor = leaderDistributedDataStore.getActorUtils().actorSelection(
 776                 ((ReadyTransactionReply)resp).getCohortPath());
 777
 778         ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(leaderDistributedDataStore.getActorUtils(),
 779             List.of(new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor),
 780                 () -> DataStoreVersions.CURRENT_VERSION)), tx2);
 781         cohort.canCommit().get(5, TimeUnit.SECONDS);
 782         cohort.preCommit().get(5, TimeUnit.SECONDS);
 783         cohort.commit().get(5, TimeUnit.SECONDS);
 784
 785         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 786     }
 787
 788     @Test
 789     public void testForwardedReadyTransactionForwardedToLeader() throws Exception {
 790         initDatastoresWithCars("testForwardedReadyTransactionForwardedToLeader");
 791         followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), "cars");
 792
 793         final Optional<ActorRef> carsFollowerShard =
 794                 followerDistributedDataStore.getActorUtils().findLocalShard("cars");
 795         assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
 796
 797         carsFollowerShard.orElseThrow().tell(GetShardDataTree.INSTANCE, followerTestKit.getRef());
 798         final DataTree dataTree = followerTestKit.expectMsgClass(DataTree.class);
 799
 800         // Send a tx with immediate commit.
 801
 802         DataTreeModification modification = dataTree.takeSnapshot().newModification();
 803         new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification);
 804         new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification);
 805
 806         final MapEntryNode car1 = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
 807         new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification);
 808
 809         ForwardedReadyTransaction forwardedReady = new ForwardedReadyTransaction(tx1, DataStoreVersions.CURRENT_VERSION,
 810             new ReadWriteShardDataTreeTransaction(mock(ShardDataTreeTransactionParent.class), tx1, modification),
 811             true, Optional.empty());
 812
 813         carsFollowerShard.orElseThrow().tell(forwardedReady, followerTestKit.getRef());
 814         Object resp = followerTestKit.expectMsgClass(Object.class);
 815         if (resp instanceof akka.actor.Status.Failure) {
 816             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
 817         }
 818
 819         assertEquals("Response type", CommitTransactionReply.class, resp.getClass());
 820
 821         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1);
 822
 823         // Send another tx without immediate commit.
 824
 825         modification = dataTree.takeSnapshot().newModification();
 826         MapEntryNode car2 = CarsModel.newCarEntry("sportage", Uint64.valueOf(30000));
 827         new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification);
 828
 829         forwardedReady = new ForwardedReadyTransaction(tx2, DataStoreVersions.CURRENT_VERSION,
 830             new ReadWriteShardDataTreeTransaction(mock(ShardDataTreeTransactionParent.class), tx2, modification),
 831             false, Optional.empty());
 832
 833         carsFollowerShard.orElseThrow().tell(forwardedReady, followerTestKit.getRef());
 834         resp = followerTestKit.expectMsgClass(Object.class);
 835         if (resp instanceof akka.actor.Status.Failure) {
 836             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
 837         }
 838
 839         assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
 840
 841         ActorSelection txActor = leaderDistributedDataStore.getActorUtils().actorSelection(
 842                 ((ReadyTransactionReply)resp).getCohortPath());
 843
 844         final ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
 845             leaderDistributedDataStore.getActorUtils(), List.of(
 846                 new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor),
 847                     () -> DataStoreVersions.CURRENT_VERSION)), tx2);
 848         cohort.canCommit().get(5, TimeUnit.SECONDS);
 849         cohort.preCommit().get(5, TimeUnit.SECONDS);
 850         cohort.commit().get(5, TimeUnit.SECONDS);
 851
 852         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2);
 853     }
 854
 855     @Test
 856     public void testTransactionForwardedToLeaderAfterRetry() throws Exception {
 857         followerDatastoreContextBuilder.shardBatchedModificationCount(2);
 858         leaderDatastoreContextBuilder.shardBatchedModificationCount(2);
 859         initDatastoresWithCarsAndPeople("testTransactionForwardedToLeaderAfterRetry");
 860
 861         // Verify backend statistics on start
 862         verifyCarsReadWriteTransactions(leaderDistributedDataStore, 0);
 863         verifyCarsReadWriteTransactions(followerDistributedDataStore, 0);
 864
 865         // Do an initial write to get the primary shard info cached.
 866
 867         final DOMStoreWriteTransaction initialWriteTx = followerDistributedDataStore.newWriteOnlyTransaction();
 868         initialWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 869         initialWriteTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 870         followerTestKit.doCommit(initialWriteTx.ready());
 871
 872         // Wait for the commit to be replicated to the follower.
 873
 874         MemberNode.verifyRaftState(followerDistributedDataStore, "cars",
 875             raftState -> assertEquals("getLastApplied", 1, raftState.getLastApplied()));
 876
 877         MemberNode.verifyRaftState(followerDistributedDataStore, "people",
 878             raftState -> assertEquals("getLastApplied", 1, raftState.getLastApplied()));
 879
 880         // Prepare, ready and canCommit a WO tx that writes to 2 shards. This will become the current tx in
 881         // the leader shard.
 882
 883         final DOMStoreWriteTransaction writeTx1 = followerDistributedDataStore.newWriteOnlyTransaction();
 884         writeTx1.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
 885         writeTx1.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 886         final DOMStoreThreePhaseCommitCohort writeTx1Cohort = writeTx1.ready();
 887         final ListenableFuture<Boolean> writeTx1CanCommit = writeTx1Cohort.canCommit();
 888         writeTx1CanCommit.get(5, TimeUnit.SECONDS);
 889
 890         // Prepare and ready another WO tx that writes to 2 shards but don't canCommit yet. This will be queued
 891         // in the leader shard.
 892
 893         final DOMStoreWriteTransaction writeTx2 = followerDistributedDataStore.newWriteOnlyTransaction();
 894         final LinkedList<MapEntryNode> cars = new LinkedList<>();
 895         int carIndex = 1;
 896         cars.add(CarsModel.newCarEntry("car" + carIndex, Uint64.valueOf(carIndex)));
 897         writeTx2.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
 898         carIndex++;
 899         NormalizedNode people = ImmutableNodes.mapNodeBuilder(PeopleModel.PERSON_QNAME)
 900                 .withChild(PeopleModel.newPersonEntry("Dude")).build();
 901         writeTx2.write(PeopleModel.PERSON_LIST_PATH, people);
 902         final DOMStoreThreePhaseCommitCohort writeTx2Cohort = writeTx2.ready();
 903
 904         // At this point only leader should see the transactions
 905         verifyCarsReadWriteTransactions(leaderDistributedDataStore, 2);
 906         verifyCarsReadWriteTransactions(followerDistributedDataStore, 0);
 907
 908         // Prepare another WO that writes to a single shard and thus will be directly committed on ready. This
 909         // tx writes 5 cars so 2 BatchedModifications messages will be sent initially and cached in the leader shard
 910         // (with shardBatchedModificationCount set to 2). The 3rd BatchedModifications will be sent on ready.
 911
 912         final DOMStoreWriteTransaction writeTx3 = followerDistributedDataStore.newWriteOnlyTransaction();
 913         for (int i = 1; i <= 5; i++, carIndex++) {
 914             cars.add(CarsModel.newCarEntry("car" + carIndex, Uint64.valueOf(carIndex)));
 915             writeTx3.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
 916         }
 917
 918         // Prepare another WO that writes to a single shard. This will send a single BatchedModifications message
 919         // on ready.
 920
 921         final DOMStoreWriteTransaction writeTx4 = followerDistributedDataStore.newWriteOnlyTransaction();
 922         cars.add(CarsModel.newCarEntry("car" + carIndex, Uint64.valueOf(carIndex)));
 923         writeTx4.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
 924         carIndex++;
 925
 926         // Prepare a RW tx that will create a tx actor and send a ForwardedReadyTransaction message to the leader shard
 927         // on ready.
 928
 929         final DOMStoreReadWriteTransaction readWriteTx = followerDistributedDataStore.newReadWriteTransaction();
 930         cars.add(CarsModel.newCarEntry("car" + carIndex, Uint64.valueOf(carIndex)));
 931         final YangInstanceIdentifier carPath = CarsModel.newCarPath("car" + carIndex);
 932         readWriteTx.write(carPath, cars.getLast());
 933
 934         // There is a difference here between implementations: tell-based protocol enforces batching on per-transaction
 935         // level whereas ask-based protocol has a global limit towards a shard -- and hence flushes out last two
 936         // transactions eagerly.
 937         verifyCarsReadWriteTransactions(leaderDistributedDataStore, 3);
 938         verifyCarsReadWriteTransactions(followerDistributedDataStore, 0);
 939
 940         // Disable elections on the leader so it switches to follower.
 941
 942         sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
 943                 .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName())
 944                 .shardElectionTimeoutFactor(10));
 945
 946         leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorUtils(), "cars");
 947
 948         // Submit all tx's - the messages should get queued for retry.
 949
 950         final ListenableFuture<Boolean> writeTx2CanCommit = writeTx2Cohort.canCommit();
 951         final DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready();
 952         final DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready();
 953         final DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready();
 954
 955         // Enable elections on the other follower so it becomes the leader, at which point the
 956         // tx's should get forwarded from the previous leader to the new leader to complete the commits.
 957
 958         sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
 959                 .customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1));
 960         IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "cars")
 961                 .tell(TimeoutNow.INSTANCE, ActorRef.noSender());
 962         IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "people")
 963                 .tell(TimeoutNow.INSTANCE, ActorRef.noSender());
 964
 965         followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort);
 966         followerTestKit.doCommit(writeTx2CanCommit, writeTx2Cohort);
 967         followerTestKit.doCommit(writeTx3Cohort);
 968         followerTestKit.doCommit(writeTx4Cohort);
 969         followerTestKit.doCommit(rwTxCohort);
 970
 971         // At this point everything is committed and the follower datastore should see 5 transactions, but leader should
 972         // only see the initial transactions
 973         verifyCarsReadWriteTransactions(leaderDistributedDataStore, 3);
 974         verifyCarsReadWriteTransactions(followerDistributedDataStore, 5);
 975
 976         DOMStoreReadTransaction readTx = leaderDistributedDataStore.newReadOnlyTransaction();
 977         verifyCars(readTx, cars.toArray(new MapEntryNode[cars.size()]));
 978         verifyNode(readTx, PeopleModel.PERSON_LIST_PATH, people);
 979     }
 980
 981     private static void verifyCarsReadWriteTransactions(final ClientBackedDataStore datastore, final int expected)
 982             throws Exception {
 983         IntegrationTestKit.verifyShardStats(datastore, "cars",
 984             stats -> assertEquals("getReadWriteTransactionCount", expected, stats.getReadWriteTransactionCount()));
 985     }
 986
 987     @Test
 988     public void testLeadershipTransferOnShutdown() throws Exception {
 989         leaderDatastoreContextBuilder.shardBatchedModificationCount(1);
 990         followerDatastoreContextBuilder.shardElectionTimeoutFactor(10).customRaftPolicyImplementation(null);
 991         final String testName = "testLeadershipTransferOnShutdown";
 992         initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS_AND_PEOPLE);
 993
 994         final IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
 995                 DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(500),
 996                 commitTimeout);
 997         try (var follower2DistributedDataStore = follower2TestKit.setupDataStore(testParameter, testName,
 998             MODULE_SHARDS_CARS_PEOPLE_1_2_3, false)) {
 999
1000             followerTestKit.waitForMembersUp("member-3");
1001             follower2TestKit.waitForMembersUp("member-1", "member-2");
1002
1003             // Create and submit a couple tx's so they're pending.
1004
1005             DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
1006             writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1007             writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
1008             writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
1009             final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
1010
1011             // FIXME: this assertion should be made in an explicit Shard test
1012             //            IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
1013             //                stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize()));
1014
1015             writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
1016             final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000));
1017             writeTx.write(CarsModel.newCarPath("optima"), car);
1018             final DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready();
1019
1020             // FIXME: this assertion should be made in an explicit Shard test
1021             //            IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
1022             //                stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()));
1023
1024             // Gracefully stop the leader via a Shutdown message.
1025
1026             sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
1027                 .shardElectionTimeoutFactor(100));
1028
1029             final FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
1030             final Future<ActorRef> future = leaderDistributedDataStore.getActorUtils().findLocalShardAsync("cars");
1031             final ActorRef leaderActor = Await.result(future, duration);
1032
1033             final Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
1034
1035             // Commit the 2 transactions. They should finish and succeed.
1036
1037             followerTestKit.doCommit(cohort1);
1038             followerTestKit.doCommit(cohort2);
1039
1040             // Wait for the leader actor stopped.
1041
1042             final Boolean stopped = Await.result(stopFuture, duration);
1043             assertEquals("Stopped", Boolean.TRUE, stopped);
1044
1045             // Verify leadership was transferred by reading the committed data from the other nodes.
1046
1047             verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car);
1048             verifyCars(follower2DistributedDataStore.newReadOnlyTransaction(), car);
1049         }
1050     }
1051
1052     @Test
1053     public void testTransactionWithIsolatedLeader() throws Exception {
1054         // Set the isolated leader check interval high so we can control the switch to IsolatedLeader.
1055         leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(10000000);
1056         final String testName = "testTransactionWithIsolatedLeader";
1057         initDatastoresWithCars(testName);
1058
1059         // Tx that is submitted after the follower is stopped but before the leader transitions to IsolatedLeader.
1060         final DOMStoreWriteTransaction preIsolatedLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
1061         preIsolatedLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1062
1063         // Tx that is submitted after the leader transitions to IsolatedLeader.
1064         final DOMStoreWriteTransaction noShardLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
1065         noShardLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1066
1067         // Tx that is submitted after the follower is reinstated.
1068         final DOMStoreWriteTransaction successWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
1069         successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1070
1071         // Stop the follower
1072         followerTestKit.watch(followerDistributedDataStore.getActorUtils().getShardManager());
1073         followerDistributedDataStore.close();
1074         followerTestKit.expectTerminated(followerDistributedDataStore.getActorUtils().getShardManager());
1075
1076         // Submit the preIsolatedLeaderWriteTx so it's pending
1077         final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
1078
1079         // Change the isolated leader check interval low so it changes to IsolatedLeader.
1080         sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
1081                 .shardIsolatedLeaderCheckIntervalInMillis(200));
1082
1083         MemberNode.verifyRaftState(leaderDistributedDataStore, "cars",
1084             raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState()));
1085
1086         final var noShardLeaderCohort = noShardLeaderWriteTx.ready();
1087         // tell-based canCommit() does not have a real timeout and hence continues
1088         final var canCommit = noShardLeaderCohort.canCommit();
1089         Uninterruptibles.sleepUninterruptibly(commitTimeout, TimeUnit.SECONDS);
1090         assertFalse(canCommit.isDone());
1091
1092         sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
1093                 .shardElectionTimeoutFactor(100));
1094
1095         final DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready();
1096
1097         followerDistributedDataStore = followerTestKit.setupDataStore(testParameter, testName,
1098             MODULE_SHARDS_CARS_ONLY_1_2, false, CARS);
1099
1100         leaderTestKit.doCommit(preIsolatedLeaderTxCohort);
1101         leaderTestKit.doCommit(successTxCohort);
1102
1103         // continuation of canCommit(): readied transaction will complete commit, but will report an OLFE
1104         final var ex = assertThrows(ExecutionException.class,
1105             () -> canCommit.get(commitTimeout, TimeUnit.SECONDS)).getCause();
1106         assertThat(ex, instanceOf(OptimisticLockFailedException.class));
1107         assertEquals("Optimistic lock failed for path " + CarsModel.BASE_PATH, ex.getMessage());
1108         final var cause = ex.getCause();
1109         assertThat(cause, instanceOf(ConflictingModificationAppliedException.class));
1110         final var cmae = (ConflictingModificationAppliedException) cause;
1111         assertEquals("Node was created by other transaction.", cmae.getMessage());
1112         assertEquals(CarsModel.BASE_PATH, cmae.getPath());
1113     }
1114
1115     @Test
1116     public void testTransactionWithShardLeaderNotResponding() throws Exception {
1117         followerDatastoreContextBuilder.frontendRequestTimeoutInSeconds(2);
1118         followerDatastoreContextBuilder.shardElectionTimeoutFactor(50);
1119         initDatastoresWithCars("testTransactionWithShardLeaderNotResponding");
1120
1121         // Do an initial read to get the primary shard info cached.
1122
1123         final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
1124         readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1125
1126         // Shutdown the leader and try to create a new tx.
1127
1128         TestKit.shutdownActorSystem(leaderSystem, true);
1129
1130         followerDatastoreContextBuilder.operationTimeoutInMillis(50).shardElectionTimeoutFactor(1);
1131         sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder);
1132
1133         final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
1134
1135         rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1136
1137         final var ex = assertThrows(ExecutionException.class, () -> followerTestKit.doCommit(rwTx.ready()));
1138         assertThat("Unexpected exception: " + Throwables.getStackTraceAsString(ex.getCause()),
1139             Throwables.getRootCause(ex), instanceOf(RequestTimeoutException.class));
1140     }
1141
1142     @Test
1143     public void testTransactionWithCreateTxFailureDueToNoLeader() throws Exception {
1144         followerDatastoreContextBuilder.frontendRequestTimeoutInSeconds(2);
1145         initDatastoresWithCars("testTransactionWithCreateTxFailureDueToNoLeader");
1146
1147         // Do an initial read to get the primary shard info cached.
1148
1149         final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
1150         readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1151
1152         // Shutdown the leader and try to create a new tx.
1153
1154         TestKit.shutdownActorSystem(leaderSystem, true);
1155
1156         Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
1157
1158         Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
1159
1160         sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
1161                 .operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
1162
1163         final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
1164
1165         rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1166
1167         final var ex = assertThrows(ExecutionException.class, () -> followerTestKit.doCommit(rwTx.ready()));
1168         assertThat("Unexpected exception: " + Throwables.getStackTraceAsString(ex.getCause()),
1169             Throwables.getRootCause(ex), instanceOf(RequestTimeoutException.class));
1170     }
1171
1172     @Test
1173     public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception {
1174         followerDatastoreContextBuilder.backendAlivenessTimerIntervalInSeconds(2);
1175         String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
1176         initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);
1177
1178         final DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder()
1179                 .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10);
1180         final IntegrationTestKit follower2TestKit = new IntegrationTestKit(
1181                 follower2System, follower2DatastoreContextBuilder, commitTimeout);
1182
1183         try (var ds = follower2TestKit.setupDataStore(testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS)) {
1184
1185             followerTestKit.waitForMembersUp("member-1", "member-3");
1186             follower2TestKit.waitForMembersUp("member-1", "member-2");
1187
1188             // Do an initial read to get the primary shard info cached.
1189
1190             final DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
1191             readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
1192
1193             // Shutdown the leader and try to create a new tx.
1194
1195             TestKit.shutdownActorSystem(leaderSystem, true);
1196
1197             Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
1198
1199             sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
1200                 .operationTimeoutInMillis(500).shardElectionTimeoutFactor(5).customRaftPolicyImplementation(null));
1201
1202             final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
1203
1204             rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1205
1206             followerTestKit.doCommit(rwTx.ready());
1207         }
1208     }
1209
1210     @Test
1211     public void testSemiReachableCandidateNotDroppingLeader() throws Exception {
1212         final String testName = "testSemiReachableCandidateNotDroppingLeader";
1213         initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);
1214
1215         final DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder()
1216                 .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10);
1217         final IntegrationTestKit follower2TestKit = new IntegrationTestKit(
1218                 follower2System, follower2DatastoreContextBuilder, commitTimeout);
1219
1220         final var ds2 = follower2TestKit.setupDataStore(testParameter, testName, MODULE_SHARDS_CARS_1_2_3, false, CARS);
1221
1222         followerTestKit.waitForMembersUp("member-1", "member-3");
1223         follower2TestKit.waitForMembersUp("member-1", "member-2");
1224
1225         // behavior is controlled by akka.coordinated-shutdown.run-by-actor-system-terminate configuration option
1226         TestKit.shutdownActorSystem(follower2System, true);
1227
1228         ActorRef cars = leaderDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
1229         final OnDemandRaftState initialState = (OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
1230                 .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
1231
1232         Cluster leaderCluster = Cluster.get(leaderSystem);
1233         Cluster followerCluster = Cluster.get(followerSystem);
1234         Cluster follower2Cluster = Cluster.get(follower2System);
1235
1236         Member follower2Member = follower2Cluster.readView().self();
1237
1238         await().atMost(10, TimeUnit.SECONDS)
1239                 .until(() -> containsUnreachable(leaderCluster, follower2Member));
1240         await().atMost(10, TimeUnit.SECONDS)
1241                 .until(() -> containsUnreachable(followerCluster, follower2Member));
1242
1243         ActorRef followerCars = followerDistributedDataStore.getActorUtils().findLocalShard("cars").orElseThrow();
1244
1245         // to simulate a follower not being able to receive messages, but still being able to send messages and becoming
1246         // candidate, we can just send a couple of RequestVotes to both leader and follower.
1247         cars.tell(new RequestVote(initialState.getCurrentTerm() + 1, "member-3-shard-cars", -1, -1), null);
1248         followerCars.tell(new RequestVote(initialState.getCurrentTerm() + 1, "member-3-shard-cars", -1, -1), null);
1249         cars.tell(new RequestVote(initialState.getCurrentTerm() + 3, "member-3-shard-cars", -1, -1), null);
1250         followerCars.tell(new RequestVote(initialState.getCurrentTerm() + 3, "member-3-shard-cars", -1, -1), null);
1251
1252         OnDemandRaftState stateAfter = (OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
1253                 .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
1254         OnDemandRaftState followerState = (OnDemandRaftState) followerDistributedDataStore.getActorUtils()
1255                 .executeOperation(cars, GetOnDemandRaftState.INSTANCE);
1256
1257         assertEquals(initialState.getCurrentTerm(), stateAfter.getCurrentTerm());
1258         assertEquals(initialState.getCurrentTerm(), followerState.getCurrentTerm());
1259
1260         ds2.close();
1261     }
1262
1263     private static Boolean containsUnreachable(final Cluster cluster, final Member member) {
1264         // unreachableMembers() returns scala.collection.immutable.Set, but we are using scala.collection.Set to fix JDT
1265         // see https://bugs.eclipse.org/bugs/show_bug.cgi?id=468276#c32
1266         final Set<Member> members = cluster.readView().unreachableMembers();
1267         return members.contains(member);
1268     }
1269
1270     @Test
1271     public void testInstallSnapshot() throws Exception {
1272         final String testName = "testInstallSnapshot";
1273         final String leaderCarShardName = "member-1-shard-cars-" + testName;
1274         final String followerCarShardName = "member-2-shard-cars-" + testName;
1275
1276         // Setup a saved snapshot on the leader. The follower will startup with no data and the leader should
1277         // install a snapshot to sync the follower.
1278
1279         DataTree tree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_CONFIGURATION,
1280             SchemaContextHelper.full());
1281
1282         final ContainerNode carsNode = CarsModel.newCarsNode(
1283                 CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", Uint64.valueOf(20000))));
1284         AbstractShardTest.writeToStore(tree, CarsModel.BASE_PATH, carsNode);
1285
1286         final NormalizedNode snapshotRoot = AbstractShardTest.readStore(tree, YangInstanceIdentifier.of());
1287         final Snapshot initialSnapshot = Snapshot.create(
1288                 new ShardSnapshotState(new MetadataShardDataTreeSnapshot(snapshotRoot)),
1289                 Collections.emptyList(), 5, 1, 5, 1, 1, null, null);
1290         InMemorySnapshotStore.addSnapshot(leaderCarShardName, initialSnapshot);
1291
1292         InMemorySnapshotStore.addSnapshotSavedLatch(leaderCarShardName);
1293         InMemorySnapshotStore.addSnapshotSavedLatch(followerCarShardName);
1294
1295         initDatastoresWithCars(testName);
1296
1297         assertEquals(Optional.of(carsNode), leaderDistributedDataStore.newReadOnlyTransaction().read(
1298             CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS));
1299
1300         verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(leaderCarShardName, Snapshot.class),
1301                 initialSnapshot, snapshotRoot);
1302
1303         verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(followerCarShardName, Snapshot.class),
1304                 initialSnapshot, snapshotRoot);
1305     }
1306
1307     @Test
1308     public void testReadWriteMessageSlicing() throws Exception {
1309         leaderDatastoreContextBuilder.maximumMessageSliceSize(100);
1310         followerDatastoreContextBuilder.maximumMessageSliceSize(100);
1311         initDatastoresWithCars("testLargeReadReplySlicing");
1312
1313         final DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
1314
1315         final NormalizedNode carsNode = CarsModel.create();
1316         rwTx.write(CarsModel.BASE_PATH, carsNode);
1317
1318         verifyNode(rwTx, CarsModel.BASE_PATH, carsNode);
1319     }
1320
1321     @SuppressWarnings("IllegalCatch")
1322     @Test
1323     public void testRaftCallbackDuringLeadershipDrop() throws Exception {
1324         final String testName = "testRaftCallbackDuringLeadershipDrop";
1325         initDatastores(testName, MODULE_SHARDS_CARS_1_2_3, CARS);
1326
1327         final ExecutorService executor = Executors.newSingleThreadExecutor();
1328
1329         final IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
1330                 DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(500)
1331                         .shardLeaderElectionTimeoutInSeconds(3600),
1332                 commitTimeout);
1333
1334         final DOMStoreWriteTransaction initialWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
1335         initialWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
1336         leaderTestKit.doCommit(initialWriteTx.ready());
1337
1338         try (var follower2DistributedDataStore = follower2TestKit.setupDataStore(testParameter, testName,
1339             MODULE_SHARDS_CARS_1_2_3, false)) {
1340
1341             final ActorRef member3Cars = ((LocalShardStore) follower2DistributedDataStore).getLocalShards()
1342                     .getLocalShards().get("cars").getActor();
1343             final ActorRef member2Cars = ((LocalShardStore)followerDistributedDataStore).getLocalShards()
1344                     .getLocalShards().get("cars").getActor();
1345             member2Cars.tell(new StartDropMessages<>(AppendEntries.class), null);
1346             member3Cars.tell(new StartDropMessages<>(AppendEntries.class), null);
1347
1348             final DOMStoreWriteTransaction newTx = leaderDistributedDataStore.newWriteOnlyTransaction();
1349             newTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
1350             final AtomicBoolean submitDone = new AtomicBoolean(false);
1351             executor.submit(() -> {
1352                 try {
1353                     leaderTestKit.doCommit(newTx.ready());
1354                     submitDone.set(true);
1355                 } catch (Exception e) {
1356                     throw new RuntimeException(e);
1357                 }
1358             });
1359             final ActorRef leaderCars = ((LocalShardStore) leaderDistributedDataStore).getLocalShards()
1360                     .getLocalShards().get("cars").getActor();
1361             await().atMost(10, TimeUnit.SECONDS)
1362                     .until(() -> ((OnDemandRaftState) leaderDistributedDataStore.getActorUtils()
1363                             .executeOperation(leaderCars, GetOnDemandRaftState.INSTANCE)).getLastIndex() >= 1);
1364
1365             final OnDemandRaftState raftState = (OnDemandRaftState)leaderDistributedDataStore.getActorUtils()
1366                     .executeOperation(leaderCars, GetOnDemandRaftState.INSTANCE);
1367
1368             // Simulate a follower not receiving heartbeats but still being able to send messages ie RequestVote with
1369             // new term(switching to candidate after election timeout)
1370             leaderCars.tell(new RequestVote(raftState.getCurrentTerm() + 1,
1371                     "member-3-shard-cars-testRaftCallbackDuringLeadershipDrop", -1,
1372                             -1), member3Cars);
1373
1374             member2Cars.tell(new StopDropMessages<>(AppendEntries.class), null);
1375             member3Cars.tell(new StopDropMessages<>(AppendEntries.class), null);
1376
1377             await("Is tx stuck in COMMIT_PENDING")
1378                     .atMost(10, TimeUnit.SECONDS).untilAtomic(submitDone, equalTo(true));
1379
1380         }
1381
1382         executor.shutdownNow();
1383     }
1384
1385     @Test
1386     public void testSnapshotOnRootOverwrite() throws Exception {
1387         initDatastores("testSnapshotOnRootOverwrite", "module-shards-default-cars-member1-and-2.conf",
1388             new String[] {"cars", "default"},
1389             leaderDatastoreContextBuilder.snapshotOnRootOverwrite(true),
1390             followerDatastoreContextBuilder.snapshotOnRootOverwrite(true));
1391
1392         leaderTestKit.waitForMembersUp("member-2");
1393         final ContainerNode rootNode = Builders.containerBuilder()
1394                 .withNodeIdentifier(NodeIdentifier.create(SchemaContext.NAME))
1395                 .withChild(CarsModel.create())
1396                 .build();
1397
1398         leaderTestKit.testWriteTransaction(leaderDistributedDataStore, YangInstanceIdentifier.of(), rootNode);
1399
1400         // FIXME: CONTROLLER-2020: ClientBackedDatastore does not have stable indexes/term,
1401         //                         the snapshot index seems to fluctuate
1402         assumeTrue(false);
1403         IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
1404             state -> assertEquals(1, state.getSnapshotIndex()));
1405
1406         IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
1407             state -> assertEquals(1, state.getSnapshotIndex()));
1408
1409         verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 1);
1410         verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 1);
1411
1412         for (int i = 0; i < 10; i++) {
1413             leaderTestKit.testWriteTransaction(leaderDistributedDataStore, CarsModel.newCarPath("car " + i),
1414                     CarsModel.newCarEntry("car " + i, Uint64.ONE));
1415         }
1416
1417         // fake snapshot causes the snapshotIndex to move
1418         IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
1419             state -> assertEquals(10, state.getSnapshotIndex()));
1420         IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
1421             state -> assertEquals(10, state.getSnapshotIndex()));
1422
1423         // however the real snapshot still has not changed and was taken at index 1
1424         verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 1);
1425         verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 1);
1426
1427         // root overwrite so expect a snapshot
1428         leaderTestKit.testWriteTransaction(leaderDistributedDataStore, YangInstanceIdentifier.of(), rootNode);
1429
1430         // this was a real snapshot so everything should be in it(1(DisableTrackingPayload) + 1 + 10 + 1)
1431         IntegrationTestKit.verifyShardState(leaderDistributedDataStore, "cars",
1432             state -> assertEquals(12, state.getSnapshotIndex()));
1433         IntegrationTestKit.verifyShardState(followerDistributedDataStore, "cars",
1434             state -> assertEquals(12, state.getSnapshotIndex()));
1435
1436         verifySnapshot("member-1-shard-cars-testSnapshotOnRootOverwrite", 12);
1437         verifySnapshot("member-2-shard-cars-testSnapshotOnRootOverwrite", 12);
1438     }
1439
1440     private static void verifySnapshot(final String persistenceId, final long lastAppliedIndex) {
1441         await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
1442                 List<Snapshot> snap = InMemorySnapshotStore.getSnapshots(persistenceId, Snapshot.class);
1443                 assertEquals(1, snap.size());
1444                 assertEquals(lastAppliedIndex, snap.get(0).getLastAppliedIndex());
1445             }
1446         );
1447     }
1448
1449     private static void verifySnapshot(final Snapshot actual, final Snapshot expected,
1450                                        final NormalizedNode expRoot) {
1451         assertEquals("Snapshot getLastAppliedTerm", expected.getLastAppliedTerm(), actual.getLastAppliedTerm());
1452         assertEquals("Snapshot getLastAppliedIndex", expected.getLastAppliedIndex(), actual.getLastAppliedIndex());
1453         assertEquals("Snapshot getLastTerm", expected.getLastTerm(), actual.getLastTerm());
1454         assertEquals("Snapshot getLastIndex", expected.getLastIndex(), actual.getLastIndex());
1455         assertEquals("Snapshot state type", ShardSnapshotState.class, actual.getState().getClass());
1456         MetadataShardDataTreeSnapshot shardSnapshot =
1457                 (MetadataShardDataTreeSnapshot) ((ShardSnapshotState)actual.getState()).getSnapshot();
1458         assertEquals("Snapshot root node", expRoot, shardSnapshot.getRootNode().orElseThrow());
1459     }
1460
1461     private static void sendDatastoreContextUpdate(final ClientBackedDataStore dataStore, final Builder builder) {
1462         final Builder newBuilder = DatastoreContext.newBuilderFrom(builder.build());
1463         final DatastoreContextFactory mockContextFactory = mock(DatastoreContextFactory.class);
1464         final Answer<DatastoreContext> answer = invocation -> newBuilder.build();
1465         doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext();
1466         doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(anyString());
1467         dataStore.onDatastoreContextUpdated(mockContextFactory);
1468     }
1469 }