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