X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDistributedDataStoreRemotingIntegrationTest.java;h=132ac47e7da77de739c6fb1b50fb7d201ce81259;hb=a1f54a11f6c8675404efc1afcca3e0bd5f9a7e36;hp=f7a5630336a7f69e5520e52fd2bd9cea094b8fa2;hpb=671ee25af46a1f852548de03cd0d3b8f74d7b7ba;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 f7a5630336..132ac47e7d 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 @@ -23,14 +23,18 @@ import akka.actor.AddressFromURIString; import akka.cluster.Cluster; import akka.dispatch.Futures; import akka.pattern.AskTimeoutException; +import akka.pattern.Patterns; import akka.testkit.JavaTestKit; import com.google.common.base.Optional; +import com.google.common.base.Supplier; 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.LinkedList; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import org.junit.After; @@ -39,6 +43,7 @@ import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +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; @@ -54,6 +59,7 @@ import org.opendaylight.controller.cluster.datastore.modification.MergeModificat 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.raft.client.messages.Shutdown; import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; @@ -84,6 +90,9 @@ import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder; import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder; import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; +import scala.concurrent.Await; +import scala.concurrent.Future; +import scala.concurrent.duration.FiniteDuration; /** * End-to-end distributed data store tests that exercise remote shards and transactions. @@ -534,6 +543,7 @@ public class DistributedDataStoreRemotingIntegrationTest { verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1); } + @SuppressWarnings("unchecked") @Test public void testReadyLocalTransactionForwardedToLeader() throws Exception { initDatastoresWithCars("testReadyLocalTransactionForwardedToLeader"); @@ -563,7 +573,7 @@ public class DistributedDataStoreRemotingIntegrationTest { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } - assertEquals("Response type", CommitTransactionReply.SERIALIZABLE_CLASS, resp.getClass()); + assertEquals("Response type", CommitTransactionReply.class, resp.getClass()); verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1); @@ -587,8 +597,11 @@ public class DistributedDataStoreRemotingIntegrationTest { ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection( ((ReadyTransactionReply)resp).getCohortPath()); + Supplier versionSupplier = Mockito.mock(Supplier.class); + Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get(); ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy( - leaderDistributedDataStore.getActorContext(), Arrays.asList(Futures.successful(txActor)), "tx-2"); + leaderDistributedDataStore.getActorContext(), Arrays.asList( + new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), "tx-2"); cohort.canCommit().get(5, TimeUnit.SECONDS); cohort.preCommit().get(5, TimeUnit.SECONDS); cohort.commit().get(5, TimeUnit.SECONDS); @@ -596,6 +609,7 @@ public class DistributedDataStoreRemotingIntegrationTest { verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2); } + @SuppressWarnings("unchecked") @Test public void testForwardedReadyTransactionForwardedToLeader() throws Exception { initDatastoresWithCars("testForwardedReadyTransactionForwardedToLeader"); @@ -618,7 +632,7 @@ public class DistributedDataStoreRemotingIntegrationTest { ForwardedReadyTransaction forwardedReady = new ForwardedReadyTransaction("tx-1", DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction( - Mockito.mock(ShardDataTreeTransactionParent.class), "tx-1", modification), true, true); + Mockito.mock(ShardDataTreeTransactionParent.class), "tx-1", modification), true); carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); Object resp = followerTestKit.expectMsgClass(Object.class); @@ -626,7 +640,7 @@ public class DistributedDataStoreRemotingIntegrationTest { throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); } - assertEquals("Response type", CommitTransactionReply.SERIALIZABLE_CLASS, resp.getClass()); + assertEquals("Response type", CommitTransactionReply.class, resp.getClass()); verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1); @@ -638,7 +652,7 @@ public class DistributedDataStoreRemotingIntegrationTest { forwardedReady = new ForwardedReadyTransaction("tx-2", DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction( - Mockito.mock(ShardDataTreeTransactionParent.class), "tx-2", modification), true, false); + Mockito.mock(ShardDataTreeTransactionParent.class), "tx-2", modification), false); carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); resp = followerTestKit.expectMsgClass(Object.class); @@ -651,8 +665,11 @@ public class DistributedDataStoreRemotingIntegrationTest { ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection( ((ReadyTransactionReply)resp).getCohortPath()); + Supplier versionSupplier = Mockito.mock(Supplier.class); + Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get(); ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy( - leaderDistributedDataStore.getActorContext(), Arrays.asList(Futures.successful(txActor)), "tx-2"); + leaderDistributedDataStore.getActorContext(), Arrays.asList( + new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), "tx-2"); cohort.canCommit().get(5, TimeUnit.SECONDS); cohort.preCommit().get(5, TimeUnit.SECONDS); cohort.commit().get(5, TimeUnit.SECONDS); @@ -662,14 +679,16 @@ public class DistributedDataStoreRemotingIntegrationTest { @Test public void testTransactionForwardedToLeaderAfterRetry() throws Exception { - initDatastoresWithCars("testTransactionForwardedToLeaderAfterRetry"); + followerDatastoreContextBuilder.shardBatchedModificationCount(2); + leaderDatastoreContextBuilder.shardBatchedModificationCount(2); + initDatastoresWithCarsAndPeople("testTransactionForwardedToLeaderAfterRetry"); // Do an initial write to get the primary shard info cached. - DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction(); - writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); - writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()); - followerTestKit.doCommit(writeTx.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. @@ -680,15 +699,54 @@ public class DistributedDataStoreRemotingIntegrationTest { } }); - // Create and prepare wo and rw tx's. + // 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 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 cars = new LinkedList<>(); + int carIndex = 1; + 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); + 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))); + writeTx3.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); + } - writeTx = followerDistributedDataStore.newWriteOnlyTransaction(); - MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); - writeTx.write(CarsModel.newCarPath("optima"), car1); + // 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))); + 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(); - MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000)); - readWriteTx.write(CarsModel.newCarPath("sportage"), car2); + cars.add(CarsModel.newCarEntry("car" + carIndex, BigInteger.valueOf(carIndex))); + readWriteTx.write(CarsModel.newCarPath("car" + carIndex), cars.getLast()); IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { @Override @@ -705,19 +763,93 @@ public class DistributedDataStoreRemotingIntegrationTest { 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 = writeTx.ready(); - DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready(); + ListenableFuture writeTx2CanCommit = writeTx2Cohort.canCommit(); + DOMStoreThreePhaseCommitCohort writeTx3Cohort = writeTx3.ready(); + DOMStoreThreePhaseCommitCohort writeTx4Cohort = writeTx4.ready(); + 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)); + 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 + public void testLeadershipTransferOnShutdown() throws Exception { + leaderDatastoreContextBuilder.shardBatchedModificationCount(1); + followerDatastoreContextBuilder.shardElectionTimeoutFactor(10).customRaftPolicyImplementation(null); + 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); + + // 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(); + + IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { + @Override + public void verify(ShardStats 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(); + + IntegrationTestKit.verifyShardStats(leaderDistributedDataStore, "cars", new ShardStatsVerifier() { + @Override + public void verify(ShardStats stats) { + assertEquals("getTxCohortCacheSize", 2, stats.getTxCohortCacheSize()); + } + }); + + // Gracefully stop the leader via a Shutdown message. + + sendDatastoreContextUpdate(leaderDistributedDataStore, leaderDatastoreContextBuilder. + shardElectionTimeoutFactor(100)); + + FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS); + Future future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars"); + ActorRef leaderActor = Await.result(future, duration); + + Future stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE); + + // Commit the 2 transactions. They should finish and succeed. + followerTestKit.doCommit(cohort1); followerTestKit.doCommit(cohort2); - verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2); + // Wait for the leader actor 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. + + verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car); + verifyCars(follower2DistributedDataStore.newReadOnlyTransaction(), car); } @Test @@ -830,6 +962,9 @@ public class DistributedDataStoreRemotingIntegrationTest { IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, follower2DatastoreContextBuilder); follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_PEOPLE_1_2_3, false, CARS); + followerTestKit.waitForMembersUp("member-1", "member-3"); + follower2TestKit.waitForMembersUp("member-1", "member-2"); + // Do an initial read to get the primary shard info cached. DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();