X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDistributedDataStoreRemotingIntegrationTest.java;h=005e904e4c9401938fd334738b3fdfa97a478399;hp=90600f952926de70e615377d7a97c61b9a9ec2eb;hb=08528168cbcf67ce2b0a897e5ac22888b1da53dd;hpb=5c5c980e564d2b5f6cd26821ffd26997f59af260 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 90600f9529..005e904e4c 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 @@ -16,10 +16,12 @@ 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; import akka.actor.Address; import akka.actor.AddressFromURIString; import akka.cluster.Cluster; +import akka.dispatch.Futures; import akka.pattern.AskTimeoutException; import akka.testkit.JavaTestKit; import com.google.common.base.Optional; @@ -28,16 +30,23 @@ 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.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; 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.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction; +import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree; import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; +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; @@ -62,8 +71,10 @@ import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode; import org.opendaylight.yangtools.yang.data.api.schema.MapNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType; 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; @@ -133,7 +144,7 @@ public class DistributedDataStoreRemotingIntegrationTest { leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), SHARD_NAMES); } - private void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception { + private static void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception { Optional> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); @@ -145,14 +156,14 @@ public class DistributedDataStoreRemotingIntegrationTest { assertEquals("Car list node", listBuilder.build(), optional.get()); } - private void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode expNode) + private static void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, 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 void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception { + private static void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception { Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS); assertEquals("exists", true, exists); } @@ -482,7 +493,7 @@ public class DistributedDataStoreRemotingIntegrationTest { // Switch the leader to the follower followerDatastoreContextBuilder.shardElectionTimeoutFactor(1); - followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build()); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); JavaTestKit.shutdownActorSystem(leaderSystem, null, true); @@ -514,27 +525,127 @@ public class DistributedDataStoreRemotingIntegrationTest { @Test public void testReadyLocalTransactionForwardedToLeader() throws Exception { initDatastores("testReadyLocalTransactionForwardedToLeader"); + followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), "cars"); Optional carsFollowerShard = followerDistributedDataStore.getActorContext().findLocalShard("cars"); assertEquals("Cars follower shard found", true, carsFollowerShard.isPresent()); - TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create(); + TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL); dataTree.setSchemaContext(SchemaContextHelper.full()); - DataTreeModification modification = dataTree.takeSnapshot().newModification(); + // Send a tx with immediate commit. + + DataTreeModification modification = dataTree.takeSnapshot().newModification(); new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification); new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification); - MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); - new WriteModification(CarsModel.newCarPath("optima"), car).apply(modification); + MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification); + modification.ready(); - String transactionID = "tx-1"; - ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(transactionID , modification, true); + ReadyLocalTransaction readyLocal = new ReadyLocalTransaction("tx-1" , modification, true); carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef()); - followerTestKit.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS); + Object resp = followerTestKit.expectMsgClass(Object.class); + if(resp instanceof akka.actor.Status.Failure) { + throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); + } + + assertEquals("Response type", CommitTransactionReply.SERIALIZABLE_CLASS, resp.getClass()); - verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car); + verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1); + + // Send another tx without immediate commit. + + modification = dataTree.takeSnapshot().newModification(); + MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000)); + new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification); + modification.ready(); + + readyLocal = new ReadyLocalTransaction("tx-2" , modification, false); + + carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef()); + resp = followerTestKit.expectMsgClass(Object.class); + if(resp instanceof akka.actor.Status.Failure) { + throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); + } + + assertEquals("Response type", ReadyTransactionReply.class, resp.getClass()); + + ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection( + ((ReadyTransactionReply)resp).getCohortPath()); + + ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy( + leaderDistributedDataStore.getActorContext(), Arrays.asList(Futures.successful(txActor)), "tx-2"); + cohort.canCommit().get(5, TimeUnit.SECONDS); + cohort.preCommit().get(5, TimeUnit.SECONDS); + cohort.commit().get(5, TimeUnit.SECONDS); + + verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2); + } + + @Test + public void testForwardedReadyTransactionForwardedToLeader() throws Exception { + initDatastores("testForwardedReadyTransactionForwardedToLeader"); + followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), "cars"); + + Optional carsFollowerShard = followerDistributedDataStore.getActorContext().findLocalShard("cars"); + assertEquals("Cars follower shard found", true, carsFollowerShard.isPresent()); + + carsFollowerShard.get().tell(GetShardDataTree.INSTANCE, followerTestKit.getRef()); + DataTree dataTree = followerTestKit.expectMsgClass(DataTree.class); + + // Send a tx with immediate commit. + + DataTreeModification modification = dataTree.takeSnapshot().newModification(); + new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification); + new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification); + + MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + new WriteModification(CarsModel.newCarPath("optima"), car1).apply(modification); + + ForwardedReadyTransaction forwardedReady = new ForwardedReadyTransaction("tx-1", + DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction( + Mockito.mock(ShardDataTreeTransactionParent.class), "tx-1", modification), true, true); + + carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); + Object resp = followerTestKit.expectMsgClass(Object.class); + if(resp instanceof akka.actor.Status.Failure) { + throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); + } + + assertEquals("Response type", CommitTransactionReply.SERIALIZABLE_CLASS, resp.getClass()); + + verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1); + + // Send another tx without immediate commit. + + modification = dataTree.takeSnapshot().newModification(); + MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000)); + new WriteModification(CarsModel.newCarPath("sportage"), car2).apply(modification); + + forwardedReady = new ForwardedReadyTransaction("tx-2", + DataStoreVersions.CURRENT_VERSION, new ReadWriteShardDataTreeTransaction( + Mockito.mock(ShardDataTreeTransactionParent.class), "tx-2", modification), true, false); + + carsFollowerShard.get().tell(forwardedReady, followerTestKit.getRef()); + resp = followerTestKit.expectMsgClass(Object.class); + if(resp instanceof akka.actor.Status.Failure) { + throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); + } + + assertEquals("Response type", ReadyTransactionReply.class, resp.getClass()); + + ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection( + ((ReadyTransactionReply)resp).getCohortPath()); + + ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy( + leaderDistributedDataStore.getActorContext(), Arrays.asList(Futures.successful(txActor)), "tx-2"); + cohort.canCommit().get(5, TimeUnit.SECONDS); + cohort.preCommit().get(5, TimeUnit.SECONDS); + cohort.commit().get(5, TimeUnit.SECONDS); + + verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2); } @Test(expected=NoShardLeaderException.class) @@ -573,7 +684,7 @@ public class DistributedDataStoreRemotingIntegrationTest { JavaTestKit.shutdownActorSystem(leaderSystem, null, true); followerDatastoreContextBuilder.operationTimeoutInMillis(50).shardElectionTimeoutFactor(1); - followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build()); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -605,7 +716,7 @@ public class DistributedDataStoreRemotingIntegrationTest { Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); followerDatastoreContextBuilder.operationTimeoutInMillis(10).shardElectionTimeoutFactor(1); - followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build()); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -639,7 +750,7 @@ public class DistributedDataStoreRemotingIntegrationTest { JavaTestKit.shutdownActorSystem(leaderSystem, null, true); followerDatastoreContextBuilder.operationTimeoutInMillis(500); - followerDistributedDataStore.onDatastoreContextUpdated(followerDatastoreContextBuilder.build()); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); @@ -647,4 +758,17 @@ public class DistributedDataStoreRemotingIntegrationTest { followerTestKit.doCommit(rwTx.ready()); } + + private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) { + DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + Answer answer = new Answer() { + @Override + public DatastoreContext answer(InvocationOnMock invocation) { + return builder.build(); + } + }; + Mockito.doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext(); + Mockito.doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); + dataStore.onDatastoreContextUpdated(mockContextFactory); + } }