X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDistributedDataStoreRemotingIntegrationTest.java;h=3b14705613fe00c49604d308b76c9bde983b6552;hb=c1336f9b497bc6867536a24f629c3f0b002ccb2f;hp=4c0aac48346a59e2db49db247cdaa37bbbd2abf0;hpb=a0b8be5ce48c0d1e0b573d1952211913c58d4935;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java index 4c0aac4834..3b14705613 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java @@ -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,6 +28,7 @@ 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; @@ -34,6 +36,7 @@ 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; @@ -54,10 +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.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; @@ -100,8 +107,10 @@ 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"; @@ -115,18 +124,21 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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); @@ -149,47 +161,53 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); CollectionNodeBuilder 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> 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); } @@ -200,7 +218,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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); @@ -245,8 +263,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load().getConfig("Member2")); - try (DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder). - setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) { + try (AbstractDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder) + .setupDistributedDataStore(testName, "module-shards-member2", true, CARS_AND_PEOPLE)) { verifyCars(member2Datastore.newReadOnlyTransaction(), car2); } @@ -373,7 +391,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } @Test - public void testTransactionChainWithMultipleShards() throws Exception{ + public void testTransactionChainWithMultipleShards() throws Exception { initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards"); DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain(); @@ -442,8 +460,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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); @@ -477,8 +495,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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. @@ -503,7 +521,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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. @@ -518,8 +536,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // 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); @@ -529,11 +547,11 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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); - try (DistributedDataStore ds = + try (AbstractDataStore ds = newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) { followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS); @@ -578,7 +596,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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()); } @@ -597,7 +615,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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()); } @@ -645,7 +663,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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()); } @@ -665,7 +683,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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()); } @@ -702,10 +720,10 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // Wait for the commit to be replicated to the follower. MemberNode.verifyRaftState(followerDistributedDataStore, "cars", - raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); + raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); MemberNode.verifyRaftState(followerDistributedDataStore, "people", - raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); + raftState -> assertEquals("getLastApplied", 0, raftState.getLastApplied())); // Prepare, ready and canCommit a WO tx that writes to 2 shards. This will become the current tx in // the leader shard. @@ -728,7 +746,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { carIndex++; NormalizedNode people = PeopleModel.newPersonMapNode(); writeTx2.write(PeopleModel.PERSON_LIST_PATH, people); - DOMStoreThreePhaseCommitCohort writeTx2Cohort = writeTx2.ready(); + 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 @@ -736,7 +754,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // sent on ready. DOMStoreWriteTransaction writeTx3 = followerDistributedDataStore.newWriteOnlyTransaction(); - for(int i = 1; i <= 5; i++, carIndex++) { + for (int i = 1; i <= 5; i++, carIndex++) { cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex))); writeTx3.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); } @@ -757,13 +775,13 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { readWriteTx.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", - stats -> assertEquals("getReadWriteTransactionCount", 1, stats.getReadWriteTransactionCount())); + 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"); @@ -771,15 +789,15 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { // Submit all tx's - the messages should get queued for retry. ListenableFuture writeTx2CanCommit = writeTx2Cohort.canCommit(); - DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready(); - DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready(); - DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready(); + final DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready(); + final DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready(); + final DOMStoreThreePhaseCommitCohort rwTxCohort = readWriteTx.ready(); // 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. - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1)); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1)); followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort); followerTestKit.doCommit(writeTx2CanCommit, writeTx2Cohort); @@ -799,8 +817,9 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { String testName = "testLeadershipTransferOnShutdown"; initDatastores(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, CARS_AND_PEOPLE); - IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, followerDatastoreContextBuilder); - try (DistributedDataStore follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(testName, + 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. @@ -809,23 +828,23 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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(); + final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready(); IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", - stats -> assertEquals("getTxCohortCacheSize", 1, stats.getTxCohortCacheSize())); + 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(); + final DOMStoreThreePhaseCommitCohort cohort2 = writeTx.ready(); IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", - stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize())); + stats -> assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize())); // 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 future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars"); @@ -851,7 +870,7 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } @Test - public void testTransactionWithIsolatedLeader() throws Throwable { + 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"; @@ -875,14 +894,14 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { followerTestKit.expectTerminated(followerDistributedDataStore.getActorContext().getShardManager()); // Submit the preIsolatedLeaderWriteTx so it's pending - DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready(); + final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready(); // Change the isolated leader check interval low so it changes to IsolatedLeader. - sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. - shardIsolatedLeaderCheckIntervalInMillis(200)); + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .shardIsolatedLeaderCheckIntervalInMillis(200)); MemberNode.verifyRaftState(leaderDistributedDataStore, "cars", - raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState())); + raftState -> assertEquals("getRaftState", "IsolatedLeader", raftState.getRaftState())); try { leaderTestKit.doCommit(noShardLeaderWriteTx.ready()); @@ -891,8 +910,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { assertEquals("getCause", NoShardLeaderException.class, e.getCause().getClass()); } - sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. - shardElectionTimeoutFactor(100)); + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder + .shardElectionTimeoutFactor(100)); DOMStoreThreePhaseCommitCohort successTxCohort = successWriteTx.ready(); @@ -903,8 +922,9 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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. @@ -929,12 +949,13 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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. @@ -950,8 +971,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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(); @@ -960,7 +981,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { try { followerTestKit.doCommit(rwTx.ready()); } catch (ExecutionException e) { - throw e.getCause(); + Throwables.propagateIfInstanceOf(e.getCause(), Exception.class); + Throwables.propagate(e.getCause()); } } @@ -969,11 +991,11 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { 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); - try (DistributedDataStore ds = + try (AbstractDataStore ds = follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS)) { followerTestKit.waitForMembersUp("member-1", "member-3"); @@ -990,8 +1012,8 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS); - sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder. - operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder + .operationTimeoutInMillis(500).shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null)); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -1001,7 +1023,57 @@ public class DistributedDataStoreRemotingIntegrationTest extends AbstractTest { } } - private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) { + @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> 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(final AbstractDataStore dataStore, final Builder builder) { final Builder newBuilder = DatastoreContext.newBuilderFrom(builder.build()); DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); Answer answer = invocation -> newBuilder.build();