BUG-2138: Create DistributedShardFrontend
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / DistributedDataStoreRemotingIntegrationTest.java
index 037a1dec5b076dc9a073ce2d796d4ebf078a09e5..3b14705613fe00c49604d308b76c9bde983b6552 100644 (file)
@@ -15,6 +15,7 @@ import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
+
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.ActorSystem;
@@ -27,12 +28,15 @@ import akka.pattern.Patterns;
 import akka.testkit.JavaTestKit;
 import com.google.common.base.Optional;
 import com.google.common.base.Supplier;
+import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.typesafe.config.ConfigFactory;
 import java.math.BigInteger;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -40,14 +44,12 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
+import org.opendaylight.controller.cluster.databroker.ConcurrentDOMDataBroker;
 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
-import org.opendaylight.controller.cluster.datastore.IntegrationTestKit.ShardStatsVerifier;
-import org.opendaylight.controller.cluster.datastore.MemberNode.RaftStateVerifier;
 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
 import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException;
-import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
@@ -55,11 +57,14 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransact
 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
-import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
-import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
+import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
+import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
+import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
@@ -97,13 +102,15 @@ import scala.concurrent.duration.FiniteDuration;
  *
  * @author Thomas Pantelis
  */
-public class DistributedDataStoreRemotingIntegrationTest {
+public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest {
 
     private static final String[] CARS_AND_PEOPLE = {"cars", "people"};
     private static final String[] CARS = {"cars"};
 
-    private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
-    private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2559");
+    private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse(
+            "akka://cluster-test@127.0.0.1:2558");
+    private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse(
+            "akka://cluster-test@127.0.0.1:2559");
 
     private static final String MODULE_SHARDS_CARS_ONLY_1_2 = "module-shards-cars-member-1-and-2.conf";
     private static final String MODULE_SHARDS_CARS_PEOPLE_1_2 = "module-shards-member1-and-2.conf";
@@ -117,16 +124,21 @@ public class DistributedDataStoreRemotingIntegrationTest {
             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
 
     private final DatastoreContext.Builder followerDatastoreContextBuilder =
-            DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5).
-                customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
+            DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5)
+                .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
+    private final TransactionIdentifier tx1 = nextTransactionId();
+    private final TransactionIdentifier tx2 = nextTransactionId();
 
-    private DistributedDataStore followerDistributedDataStore;
-    private DistributedDataStore leaderDistributedDataStore;
+    private AbstractDataStore followerDistributedDataStore;
+    private AbstractDataStore leaderDistributedDataStore;
     private IntegrationTestKit followerTestKit;
     private IntegrationTestKit leaderTestKit;
 
     @Before
     public void setUp() {
+        InMemoryJournal.clear();
+        InMemorySnapshotStore.clear();
+
         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 
@@ -139,50 +151,63 @@ public class DistributedDataStoreRemotingIntegrationTest {
 
     @After
     public void tearDown() {
+        if (followerDistributedDataStore != null) {
+            leaderDistributedDataStore.close();
+        }
+        if (leaderDistributedDataStore != null) {
+            leaderDistributedDataStore.close();
+        }
+
         JavaTestKit.shutdownActorSystem(leaderSystem);
         JavaTestKit.shutdownActorSystem(followerSystem);
         JavaTestKit.shutdownActorSystem(follower2System);
+
+        InMemoryJournal.clear();
+        InMemorySnapshotStore.clear();
     }
 
-    private void initDatastoresWithCars(String type) {
+    private void initDatastoresWithCars(final String type) {
         initDatastores(type, MODULE_SHARDS_CARS_ONLY_1_2, CARS);
     }
 
-    private void initDatastoresWithCarsAndPeople(String type) {
+    private void initDatastoresWithCarsAndPeople(final String type) {
         initDatastores(type, MODULE_SHARDS_CARS_PEOPLE_1_2, CARS_AND_PEOPLE);
     }
 
-    private void initDatastores(String type, String moduleShardsConfig, String[] shards) {
+    private void initDatastores(final String type, final String moduleShardsConfig, final String[] shards) {
         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 
         leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, shards);
 
         followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
-        followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, shards);
+        followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false,
+                shards);
 
         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), shards);
     }
 
-    private static void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception {
+    private static void verifyCars(final DOMStoreReadTransaction readTx, final MapEntryNode... entries)
+            throws Exception {
         Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
         assertEquals("isPresent", true, optional.isPresent());
 
         CollectionNodeBuilder<MapEntryNode, MapNode> listBuilder = ImmutableNodes.mapNodeBuilder(CarsModel.CAR_QNAME);
-        for(NormalizedNode<?, ?> entry: entries) {
+        for (NormalizedNode<?, ?> entry: entries) {
             listBuilder.withChild((MapEntryNode) entry);
         }
 
         assertEquals("Car list node", listBuilder.build(), optional.get());
     }
 
-    private static void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode<?, ?> expNode)
-            throws Exception {
+    private static void verifyNode(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path,
+            final NormalizedNode<?, ?> expNode) throws Exception {
         Optional<NormalizedNode<?, ?>> optional = readTx.read(path).get(5, TimeUnit.SECONDS);
         assertEquals("isPresent", true, optional.isPresent());
         assertEquals("Data node", expNode, optional.get());
     }
 
-    private static void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception {
+    private static void verifyExists(final DOMStoreReadTransaction readTx, final YangInstanceIdentifier path)
+            throws Exception {
         Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS);
         assertEquals("exists", true, exists);
     }
@@ -193,7 +218,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
         initDatastoresWithCars(testName);
 
         String followerCarShardName = "member-2-shard-cars-" + testName;
-        InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 2, ApplyJournalEntries.class );
+        InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 2, ApplyJournalEntries.class);
 
         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
@@ -238,10 +263,10 @@ public class DistributedDataStoreRemotingIntegrationTest {
 
         ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load().getConfig("Member2"));
 
-        DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder).
-                setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE);
-
-        verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
+        try (AbstractDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder)
+                .setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) {
+            verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
+        }
 
         JavaTestKit.shutdownActorSystem(newSystem);
     }
@@ -366,7 +391,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
     }
 
     @Test
-    public void testTransactionChainWithMultipleShards() throws Exception{
+    public void testTransactionChainWithMultipleShards() throws Exception {
         initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards");
 
         DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
@@ -435,8 +460,8 @@ public class DistributedDataStoreRemotingIntegrationTest {
         DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 
         ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
-                new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
-                    withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
+                new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
+                    .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 
         writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
 
@@ -470,8 +495,8 @@ public class DistributedDataStoreRemotingIntegrationTest {
         writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
 
         ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
-                new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
-                    withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
+                new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME))
+                    .withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
 
         // Note that merge will validate the data and fail but put succeeds b/c deep validation is not
         // done for put for performance reasons.
@@ -496,7 +521,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
         initDatastoresWithCars(testName);
 
         String followerCarShardName = "member-2-shard-cars-" + testName;
-        InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class );
+        InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class);
 
         // Write top-level car container from the follower so it uses a remote Tx.
 
@@ -511,34 +536,38 @@ public class DistributedDataStoreRemotingIntegrationTest {
 
         // Switch the leader to the follower
 
-        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
-                shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+                .shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
 
         JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+        Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
 
         followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), CARS);
 
         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
         Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
 
-        DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder().
-                shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
+        DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder()
+                .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
         IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder);
-        newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS);
 
-        followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS);
+        try (AbstractDataStore ds =
+                newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) {
 
-        // Write a car entry to the new leader - should switch to local Tx
+            followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS);
 
-        writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+            // Write a car entry to the new leader - should switch to local Tx
 
-        MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
-        YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
-        writeTx.merge(car1Path, car1);
+            writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
 
-        followerTestKit.doCommit(writeTx.ready());
+            MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
+            YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
+            writeTx.merge(car1Path, car1);
+
+            followerTestKit.doCommit(writeTx.ready());
 
-        verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1);
+            verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1);
+        }
     }
 
     @SuppressWarnings("unchecked")
@@ -563,11 +592,11 @@ public class DistributedDataStoreRemotingIntegrationTest {
         new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification);
         modification.ready();
 
-        ReadyLocalTransaction readyLocal = new ReadyLocalTransaction("tx-1" , modification, true);
+        ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(tx1 , modification, true);
 
         carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef());
         Object resp = followerTestKit.expectMsgClass(Object.class);
-        if(resp instanceof akka.actor.Status.Failure) {
+        if (resp instanceof akka.actor.Status.Failure) {
             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
         }
 
@@ -582,11 +611,11 @@ public class DistributedDataStoreRemotingIntegrationTest {
         new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification);
         modification.ready();
 
-        readyLocal = new ReadyLocalTransaction("tx-2" , modification, false);
+        readyLocal = new ReadyLocalTransaction(tx2 , modification, false);
 
         carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef());
         resp = followerTestKit.expectMsgClass(Object.class);
-        if(resp instanceof akka.actor.Status.Failure) {
+        if (resp instanceof akka.actor.Status.Failure) {
             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
         }
 
@@ -599,7 +628,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
         Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
         ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
                 leaderDistributedDataStore.getActorContext(), Arrays.asList(
-                        new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), "tx-2");
+                        new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
         cohort.canCommit().get(5, TimeUnit.SECONDS);
         cohort.preCommit().get(5, TimeUnit.SECONDS);
         cohort.commit().get(5, TimeUnit.SECONDS);
@@ -628,13 +657,13 @@ public class DistributedDataStoreRemotingIntegrationTest {
         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
         new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification);
 
-        ForwardedReadyTransaction forwardedReady = new ForwardedReadyTransaction("tx-1",
+        ForwardedReadyTransaction forwardedReady = new ForwardedReadyTransaction(tx1,
                 DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction(
-                        Mockito.mock(ShardDataTreeTransactionParent.class), "tx-1", modification), true);
+                        Mockito.mock(ShardDataTreeTransactionParent.class), tx1, modification), true);
 
         carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef());
         Object resp = followerTestKit.expectMsgClass(Object.class);
-        if(resp instanceof akka.actor.Status.Failure) {
+        if (resp instanceof akka.actor.Status.Failure) {
             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
         }
 
@@ -648,13 +677,13 @@ public class DistributedDataStoreRemotingIntegrationTest {
         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000));
         new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification);
 
-        forwardedReady = new ForwardedReadyTransaction("tx-2",
+        forwardedReady = new ForwardedReadyTransaction(tx2,
                 DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction(
-                        Mockito.mock(ShardDataTreeTransactionParent.class), "tx-2", modification), false);
+                        Mockito.mock(ShardDataTreeTransactionParent.class), tx2, modification), false);
 
         carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef());
         resp = followerTestKit.expectMsgClass(Object.class);
-        if(resp instanceof akka.actor.Status.Failure) {
+        if (resp instanceof akka.actor.Status.Failure) {
             throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause());
         }
 
@@ -667,7 +696,7 @@ public class DistributedDataStoreRemotingIntegrationTest {
         Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
         ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
                 leaderDistributedDataStore.getActorContext(), Arrays.asList(
-                        new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), "tx-2");
+                        new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
         cohort.canCommit().get(5, TimeUnit.SECONDS);
         cohort.preCommit().get(5, TimeUnit.SECONDS);
         cohort.commit().get(5, TimeUnit.SECONDS);
@@ -679,73 +708,106 @@ public class DistributedDataStoreRemotingIntegrationTest {
     public void testTransactionForwardedToLeaderAfterRetry() throws Exception {
         followerDatastoreContextBuilder.shardBatchedModificationCount(2);
         leaderDatastoreContextBuilder.shardBatchedModificationCount(2);
-        initDatastoresWithCars("testTransactionForwardedToLeaderAfterRetry");
+        initDatastoresWithCarsAndPeople("testTransactionForwardedToLeaderAfterRetry");
 
         // Do an initial write to get the primary shard info cached.
 
-        DOMStoreWriteTransaction writeTx1 = followerDistributedDataStore.newWriteOnlyTransaction();
-        writeTx1.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
-        writeTx1.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
-        followerTestKit.doCommit(writeTx1.ready());
+        DOMStoreWriteTransaction initialWriteTx = followerDistributedDataStore.newWriteOnlyTransaction();
+        initialWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+        initialWriteTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
+        followerTestKit.doCommit(initialWriteTx.ready());
 
         // Wait for the commit to be replicated to the follower.
 
-        MemberNode.verifyRaftState(followerDistributedDataStore, "cars", new RaftStateVerifier() {
-            @Override
-            public void verify(OnDemandRaftState raftState) {
-                assertEquals("getLastApplied", 0, raftState.getLastApplied());
-            }
-        });
+        MemberNode.verifyRaftState(followerDistributedDataStore, "cars",
+            raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied()));
 
-        // Create and prepare wo and rw tx's.
+        MemberNode.verifyRaftState(followerDistributedDataStore, "people",
+            raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied()));
 
-        writeTx1 = followerDistributedDataStore.newWriteOnlyTransaction();
+        // Prepare, ready and canCommit a WO tx that writes to 2 shards. This will become the current tx in
+        // the leader shard.
+
+        DOMStoreWriteTransaction writeTx1 = followerDistributedDataStore.newWriteOnlyTransaction();
+        writeTx1.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
+        writeTx1.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
+        DOMStoreThreePhaseCommitCohort writeTx1Cohort = writeTx1.ready();
+        ListenableFuture<Boolean> writeTx1CanCommit = writeTx1Cohort.canCommit();
+        writeTx1CanCommit.get(5, TimeUnit.SECONDS);
+
+        // Prepare and ready another WO tx that writes to 2 shards but don't canCommit yet. This will be queued
+        // in the leader shard.
+
+        DOMStoreWriteTransaction writeTx2 = followerDistributedDataStore.newWriteOnlyTransaction();
         LinkedList<MapEntryNode> cars = new LinkedList<>();
         int carIndex = 1;
-        for(; carIndex <= 5; carIndex++) {
+        cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex)));
+        writeTx2.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
+        carIndex++;
+        NormalizedNode<?, ?> people = PeopleModel.newPersonMapNode();
+        writeTx2.write(PeopleModel.PERSON_LIST_PATH, people);
+        final DOMStoreThreePhaseCommitCohort writeTx2Cohort = writeTx2.ready();
+
+        // Prepare another WO that writes to a single shard and thus will be directly committed on ready. This
+        // tx writes 5 cars so 2 BatchedModidifications messages will be sent initially and cached in the
+        // leader shard (with shardBatchedModificationCount set to 2). The 3rd BatchedModidifications will be
+        // sent on ready.
+
+        DOMStoreWriteTransaction writeTx3 = followerDistributedDataStore.newWriteOnlyTransaction();
+        for (int i = 1; i <= 5; i++, carIndex++) {
             cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex)));
-            writeTx1.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
+            writeTx3.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
         }
 
-        DOMStoreWriteTransaction writeTx2 = followerDistributedDataStore.newWriteOnlyTransaction();
+        // Prepare another WO that writes to a single shard. This will send a single BatchedModidifications
+        // message on ready.
+
+        DOMStoreWriteTransaction writeTx4 = followerDistributedDataStore.newWriteOnlyTransaction();
         cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex)));
-        writeTx2.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
+        writeTx4.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
         carIndex++;
 
+        // Prepare a RW tx that will create a tx actor and send a ForwardedReadyTransaciton message to the
+        // leader shard on ready.
+
         DOMStoreReadWriteTransaction readWriteTx = followerDistributedDataStore.newReadWriteTransaction();
         cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex)));
         readWriteTx.write(CarsModel.newCarPath("car" + carIndex), cars.getLast());
 
-        IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() {
-            @Override
-            public void verify(ShardStats stats) {
-                assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount());
-            }
-        });
+        IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
+            stats -> assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount()));
 
         // Disable elections on the leader so it switches to follower.
 
-        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder.
-                customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()).
-                shardElectionTimeoutFactor(10));
+        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+                .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName())
+                .shardElectionTimeoutFactor(10));
 
+        Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
         leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorContext(), "cars");
 
-        // Submit tx's and enable elections on the follower so it becomes the leader, at which point the
-        // readied tx's should get forwarded from the previous leader.
+        // Submit all tx's - the messages should get queued for retry.
 
-        DOMStoreThreePhaseCommitCohort cohort1 = writeTx1.ready();
-        DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
-        DOMStoreThreePhaseCommitCohort cohort3 = readWriteTx.ready();
+        ListenableFuture<Boolean> writeTx2CanCommit = writeTx2Cohort.canCommit();
+        final DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready();
+        final DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready();
+        final DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready();
 
-        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
-                customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1));
+        // Enable elections on the other follower so it becomes the leader, at which point the
+        // tx's should get forwarded from the previous leader to the new leader to complete the commits.
 
-        followerTestKit.doCommit(cohort1);
-        followerTestKit.doCommit(cohort2);
-        followerTestKit.doCommit(cohort3);
+        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+                .customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1));
 
-        verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), cars.toArray(new MapEntryNode[cars.size()]));
+        followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort);
+        followerTestKit.doCommit(writeTx2CanCommit, writeTx2Cohort);
+        followerTestKit.doCommit(writeTx3Cohort);
+        followerTestKit.doCommit(writeTx4Cohort);
+        followerTestKit.doCommit(rwTxCohort);
+
+        DOMStoreReadTransaction readTx = leaderDistributedDataStore.newReadOnlyTransaction();
+        verifyCars(readTx, cars.toArray(new MapEntryNode[cars.size()]));
+        verifyNode(readTx, PeopleModel.PERSON_LIST_PATH, people);
     }
 
     @Test
@@ -755,107 +817,114 @@ public class DistributedDataStoreRemotingIntegrationTest {
         String testName = "testLeadershipTransferOnShutdown";
         initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS_AND_PEOPLE);
 
-        IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, followerDatastoreContextBuilder);
-        DistributedDataStore follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(testName,
-                MODULE_SHARDS_CARS_PEOPLE_1_2_3, false);
+        IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System,
+                DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build()).operationTimeoutInMillis(100));
+        try (AbstractDataStore follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(testName,
+                MODULE_SHARDS_CARS_PEOPLE_1_2_3, false)) {
 
-        // Create and submit a couple tx's so they're pending.
+            // Create and submit a couple tx's so they're pending.
 
-        DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
-        writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
-        writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
-        writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
-        DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
+            DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+            writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+            writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
+            writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
+            final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
 
-        IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() {
-            @Override
-            public void verify(ShardStats stats) {
-                assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize());
-            }
-        });
+            IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
+                stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize()));
 
-        writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
-        MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
-        writeTx.write(CarsModel.newCarPath("optima"), car);
-        DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready();
+            writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+            MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
+            writeTx.write(CarsModel.newCarPath("optima"), car);
+            final DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready();
 
-        IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() {
-            @Override
-            public void verify(ShardStats stats) {
-                assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize());
-            }
-        });
+            IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars",
+                stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()));
 
-        // Gracefully stop the leader via a Shutdown message.
+            // Gracefully stop the leader via a Shutdown message.
 
-        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder.
-                shardElectionTimeoutFactor(100));
+            sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+                .shardElectionTimeoutFactor(100));
 
-        FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
-        Future<ActorRef> future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars");
-        ActorRef leaderActor = Await.result(future, duration);
+            FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
+            Future<ActorRef> future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars");
+            ActorRef leaderActor = Await.result(future, duration);
 
-        Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
+            Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
 
-        // Commit the 2 transactions. They should finish and succeed.
+            // Commit the 2 transactions. They should finish and succeed.
 
-        followerTestKit.doCommit(cohort1);
-        followerTestKit.doCommit(cohort2);
+            followerTestKit.doCommit(cohort1);
+            followerTestKit.doCommit(cohort2);
 
-        // Wait for the leader actor stopped.
+            // Wait for the leader actor stopped.
 
-        Boolean stopped = Await.result(stopFuture, duration);
-        assertEquals("Stopped", Boolean.TRUE, stopped);
+            Boolean stopped = Await.result(stopFuture, duration);
+            assertEquals("Stopped", Boolean.TRUE, stopped);
 
-        // Verify leadership was transferred by reading the committed data from the other nodes.
+            // Verify leadership was transferred by reading the committed data from the other nodes.
 
-        verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car);
-        verifyCars(follower2DistributedDataStore.newReadOnlyTransaction(), car);
+            verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car);
+            verifyCars(follower2DistributedDataStore.newReadOnlyTransaction(), car);
+        }
     }
 
     @Test
-    public void testTransactionWithIsolatedLeader() throws Throwable {
-        leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(200);
+    public void testTransactionWithIsolatedLeader() throws Exception {
+        // Set the isolated leader check interval high so we can control the switch to IsolatedLeader.
+        leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(10000000);
         String testName = "testTransactionWithIsolatedLeader";
         initDatastoresWithCars(testName);
 
-        DOMStoreWriteTransaction failWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
-        failWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+        // Tx that is submitted after the follower is stopped but before the leader transitions to IsolatedLeader.
+        DOMStoreWriteTransaction preIsolatedLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+        preIsolatedLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+        // Tx that is submitted after the leader transitions to IsolatedLeader.
+        DOMStoreWriteTransaction noShardLeaderWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
+        noShardLeaderWriteTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 
+        // Tx that is submitted after the follower is reinstated.
         DOMStoreWriteTransaction successWriteTx = leaderDistributedDataStore.newWriteOnlyTransaction();
         successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
 
+        // Stop the follower
         followerTestKit.watch(followerDistributedDataStore.getActorContext().getShardManager());
         followerDistributedDataStore.close();
         followerTestKit.expectTerminated(followerDistributedDataStore.getActorContext().getShardManager());
 
-        MemberNode.verifyRaftState(leaderDistributedDataStore, "cars", new RaftStateVerifier() {
-            @Override
-            public void verify(OnDemandRaftState raftState) {
-                assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState());
-            }
-        });
+        // Submit the preIsolatedLeaderWriteTx so it's pending
+        final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
+
+        // Change the isolated leader check interval low so it changes to IsolatedLeader.
+        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+                .shardIsolatedLeaderCheckIntervalInMillis(200));
+
+        MemberNode.verifyRaftState(leaderDistributedDataStore, "cars",
+            raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState()));
 
         try {
-            leaderTestKit.doCommit(failWriteTx.ready());
+            leaderTestKit.doCommit(noShardLeaderWriteTx.ready());
             fail("Expected NoShardLeaderException");
         } catch (ExecutionException e) {
             assertEquals("getCause", NoShardLeaderException.class, e.getCause().getClass());
         }
 
-        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder.
-                shardElectionTimeoutFactor(100));
+        sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder
+                .shardElectionTimeoutFactor(100));
 
-        DOMStoreThreePhaseCommitCohort writeTxCohort = successWriteTx.ready();
+        DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready();
 
         followerDistributedDataStore = followerTestKit.setupDistributedDataStore(testName,
                 MODULE_SHARDS_CARS_ONLY_1_2, false, CARS);
 
-        leaderTestKit.doCommit(writeTxCohort);
+        leaderTestKit.doCommit(preIsolatedLeaderTxCohort);
+        leaderTestKit.doCommit(successTxCohort);
     }
 
-    @Test(expected=AskTimeoutException.class)
-    public void testTransactionWithShardLeaderNotResponding() throws Throwable {
+    @Test(expected = AskTimeoutException.class)
+    public void testTransactionWithShardLeaderNotResponding() throws Exception {
+        followerDatastoreContextBuilder.shardElectionTimeoutFactor(50);
         initDatastoresWithCars("testTransactionWithShardLeaderNotResponding");
 
         // Do an initial read to get the primary shard info cached.
@@ -880,12 +949,13 @@ public class DistributedDataStoreRemotingIntegrationTest {
             assertTrue("Expected ShardLeaderNotRespondingException cause. Actual: " + e.getCause(),
                     e.getCause() instanceof ShardLeaderNotRespondingException);
             assertNotNull("Expected a nested cause", e.getCause().getCause());
-            throw e.getCause().getCause();
+            Throwables.propagateIfInstanceOf(e.getCause().getCause(), Exception.class);
+            Throwables.propagate(e.getCause().getCause());
         }
     }
 
-    @Test(expected=NoShardLeaderException.class)
-    public void testTransactionWithCreateTxFailureDueToNoLeader() throws Throwable {
+    @Test(expected = NoShardLeaderException.class)
+    public void testTransactionWithCreateTxFailureDueToNoLeader() throws Exception {
         initDatastoresWithCars("testTransactionWithCreateTxFailureDueToNoLeader");
 
         // Do an initial read to get the primary shard info cached.
@@ -897,10 +967,12 @@ public class DistributedDataStoreRemotingIntegrationTest {
 
         JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
 
+        Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
+
         Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
 
-        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
-                operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+                .operationTimeoutInMillis(10).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
 
         DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
 
@@ -909,7 +981,8 @@ public class DistributedDataStoreRemotingIntegrationTest {
         try {
             followerTestKit.doCommit(rwTx.ready());
         } catch (ExecutionException e) {
-            throw e.getCause();
+            Throwables.propagateIfInstanceOf(e.getCause(), Exception.class);
+            Throwables.propagate(e.getCause());
         }
     }
 
@@ -918,39 +991,92 @@ public class DistributedDataStoreRemotingIntegrationTest {
         String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx";
         initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS);
 
-        DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder().
-                shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
+        DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder()
+                .shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
         IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, follower2DatastoreContextBuilder);
-        follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS);
 
-        // Do an initial read to get the primary shard info cached.
+        try (AbstractDataStore ds =
+                follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS)) {
 
-        DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
-        readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
+            followerTestKit.waitForMembersUp("member-1", "member-3");
+            follower2TestKit.waitForMembersUp("member-1", "member-2");
 
-        // Shutdown the leader and try to create a new tx.
+            // Do an initial read to get the primary shard info cached.
 
-        JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+            DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+            readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
 
-        sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
-                operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+            // Shutdown the leader and try to create a new tx.
 
-        DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+            JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
 
-        rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+            Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
 
-        followerTestKit.doCommit(rwTx.ready());
+            sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
+                .operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+
+            DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
+
+            rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+
+            followerTestKit.doCommit(rwTx.ready());
+        }
+    }
+
+    @Test
+    public void testInstallSnapshot() throws Exception {
+        final String testName = "testInstallSnapshot";
+        final String leaderCarShardName = "member-1-shard-cars-" + testName;
+        final String followerCarShardName = "member-2-shard-cars-" + testName;
+
+        // Setup a saved snapshot on the leader. The follower will startup with no data and the leader should
+        // install a snapshot to sync the follower.
+
+        TipProducingDataTree tree = InMemoryDataTreeFactory.getInstance().create(TreeType.CONFIGURATION);
+        tree.setSchemaContext(SchemaContextHelper.full());
+
+        ContainerNode carsNode = CarsModel.newCarsNode(
+                CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000))));
+        AbstractShardTest.writeToStore(tree, CarsModel.BASE_PATH, carsNode);
+
+        NormalizedNode<?, ?> snapshotRoot = AbstractShardTest.readStore(tree, YangInstanceIdentifier.EMPTY);
+        Snapshot initialSnapshot = Snapshot.create(
+                new ShardSnapshotState(new MetadataShardDataTreeSnapshot(snapshotRoot)),
+                Collections.emptyList(), 5, 1, 5, 1, 1, null, null);
+        InMemorySnapshotStore.addSnapshot(leaderCarShardName, initialSnapshot);
+
+        InMemorySnapshotStore.addSnapshotSavedLatch(leaderCarShardName);
+        InMemorySnapshotStore.addSnapshotSavedLatch(followerCarShardName);
+
+        initDatastoresWithCars(testName);
+
+        Optional<NormalizedNode<?, ?>> readOptional = leaderDistributedDataStore.newReadOnlyTransaction().read(
+                CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS);
+        assertEquals("isPresent", true, readOptional.isPresent());
+        assertEquals("Node", carsNode, readOptional.get());
+
+        verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(leaderCarShardName, Snapshot.class),
+                initialSnapshot, snapshotRoot);
+
+        verifySnapshot(InMemorySnapshotStore.waitForSavedSnapshot(followerCarShardName, Snapshot.class),
+                initialSnapshot, snapshotRoot);
+    }
+
+    private static void verifySnapshot(Snapshot actual, Snapshot expected, NormalizedNode<?, ?> expRoot) {
+        assertEquals("Snapshot getLastAppliedTerm", expected.getLastAppliedTerm(), actual.getLastAppliedTerm());
+        assertEquals("Snapshot getLastAppliedIndex", expected.getLastAppliedIndex(), actual.getLastAppliedIndex());
+        assertEquals("Snapshot getLastTerm", expected.getLastTerm(), actual.getLastTerm());
+        assertEquals("Snapshot getLastIndex", expected.getLastIndex(), actual.getLastIndex());
+        assertEquals("Snapshot state type", ShardSnapshotState.class, actual.getState().getClass());
+        MetadataShardDataTreeSnapshot shardSnapshot =
+                (MetadataShardDataTreeSnapshot) ((ShardSnapshotState)actual.getState()).getSnapshot();
+        assertEquals("Snapshot root node", expRoot, shardSnapshot.getRootNode().get());
     }
 
-    private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) {
+    private static void sendDatastoreContextUpdate(final AbstractDataStore dataStore, final Builder builder) {
         final Builder newBuilder = DatastoreContext.newBuilderFrom(builder.build());
         DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class);
-        Answer<DatastoreContext> answer = new Answer<DatastoreContext>() {
-            @Override
-            public DatastoreContext answer(InvocationOnMock invocation) {
-                return newBuilder.build();
-            }
-        };
+        Answer<DatastoreContext> answer = invocation -> newBuilder.build();
         Mockito.doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext();
         Mockito.doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString());
         dataStore.onDatastoreContextUpdated(mockContextFactory);