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%2FIntegrationTestKit.java;h=2cb63ce79c49187d0b4a06a49bea9323c843ef20;hb=refs%2Fchanges%2F91%2F84091%2F17;hp=01678ad74e90141ef238845ce6202fd922f35836;hpb=898b12df2bde612b152742677bc9739b277ab047;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java index 01678ad74e..2cb63ce79c 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java @@ -9,6 +9,7 @@ package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import akka.actor.ActorRef; @@ -17,17 +18,18 @@ import akka.cluster.Cluster; import akka.cluster.ClusterEvent.CurrentClusterState; import akka.cluster.Member; import akka.cluster.MemberStatus; -import com.google.common.base.Optional; import com.google.common.base.Stopwatch; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.Uninterruptibles; import java.lang.reflect.Constructor; +import java.util.Optional; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import org.mockito.Mockito; +import org.opendaylight.controller.cluster.databroker.ClientBackedDataStore; import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl; @@ -35,14 +37,14 @@ import org.opendaylight.controller.cluster.datastore.config.EmptyModuleShardConf import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState; import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; -import org.opendaylight.controller.cluster.datastore.utils.ActorContext; +import org.opendaylight.controller.cluster.datastore.utils.ActorUtils; import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; -import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; -import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; +import org.opendaylight.mdsal.common.api.LogicalDatastoreType; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain; +import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.model.api.SchemaContext; @@ -50,7 +52,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.Await; import scala.concurrent.Future; -import scala.concurrent.duration.Duration; +import scala.concurrent.duration.FiniteDuration; public class IntegrationTestKit extends ShardTestKit { @@ -64,7 +66,8 @@ public class IntegrationTestKit extends ShardTestKit { this(actorSystem, datastoreContextBuilder, 7); } - public IntegrationTestKit(final ActorSystem actorSystem, final Builder datastoreContextBuilder, int commitTimeout) { + public IntegrationTestKit(final ActorSystem actorSystem, final Builder datastoreContextBuilder, + final int commitTimeout) { super(actorSystem); this.datastoreContextBuilder = datastoreContextBuilder; this.commitTimeout = commitTimeout; @@ -128,7 +131,10 @@ public class IntegrationTestKit extends ShardTestKit { final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); final Configuration config = new ConfigurationImpl(moduleShardsConfig, modulesConfig); - datastoreContextBuilder.dataStoreName(typeName); + setDataStoreName(typeName); + + // Make sure we set up datastore context correctly + datastoreContextBuilder.useTellBasedProtocol(ClientBackedDataStore.class.isAssignableFrom(implementation)); final DatastoreContext datastoreContext = datastoreContextBuilder.build(); final DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); @@ -139,25 +145,35 @@ public class IntegrationTestKit extends ShardTestKit { ActorSystem.class, ClusterWrapper.class, Configuration.class, DatastoreContextFactory.class, DatastoreSnapshot.class); - final AbstractDataStore dataStore = constructor.newInstance( - getSystem(), cluster, config, mockContextFactory, restoreFromSnapshot); + final AbstractDataStore dataStore = constructor.newInstance(getSystem(), cluster, config, mockContextFactory, + restoreFromSnapshot); dataStore.onGlobalContextUpdated(schemaContext); if (waitUntilLeader) { - waitUntilLeader(dataStore.getActorContext(), shardNames); + waitUntilLeader(dataStore.getActorUtils(), shardNames); } datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext); return dataStore; } + private void setDataStoreName(final String typeName) { + if ("config".equals(typeName)) { + datastoreContextBuilder.logicalStoreType(LogicalDatastoreType.CONFIGURATION); + } else if ("operational".equals(typeName)) { + datastoreContextBuilder.logicalStoreType(LogicalDatastoreType.OPERATIONAL); + } else { + datastoreContextBuilder.dataStoreName(typeName); + } + } + public DistributedDataStore setupDistributedDataStoreWithoutConfig(final String typeName, final SchemaContext schemaContext) { final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); final ConfigurationImpl configuration = new ConfigurationImpl(new EmptyModuleShardConfigProvider()); - getDatastoreContextBuilder().dataStoreName(typeName); + setDataStoreName(typeName); final DatastoreContext datastoreContext = getDatastoreContextBuilder().build(); @@ -180,7 +196,7 @@ public class IntegrationTestKit extends ShardTestKit { final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); final ConfigurationImpl configuration = new ConfigurationImpl(new EmptyModuleShardConfigProvider()); - getDatastoreContextBuilder().dataStoreName(typeName); + setDataStoreName(typeName); final DatastoreContext datastoreContext = getDatastoreContextBuilder().logicalStoreType(storeType).build(); @@ -198,9 +214,9 @@ public class IntegrationTestKit extends ShardTestKit { return dataStore; } - public void waitUntilLeader(final ActorContext actorContext, final String... shardNames) { + public void waitUntilLeader(final ActorUtils actorUtils, final String... shardNames) { for (String shardName: shardNames) { - ActorRef shard = findLocalShard(actorContext, shardName); + ActorRef shard = findLocalShard(actorUtils, shardName); assertNotNull("Shard was not created for " + shardName, shard); @@ -208,9 +224,9 @@ public class IntegrationTestKit extends ShardTestKit { } } - public void waitUntilNoLeader(final ActorContext actorContext, final String... shardNames) { + public void waitUntilNoLeader(final ActorUtils actorUtils, final String... shardNames) { for (String shardName: shardNames) { - ActorRef shard = findLocalShard(actorContext, shardName); + ActorRef shard = findLocalShard(actorUtils, shardName); assertNotNull("No local shard found for " + shardName, shard); waitUntilNoLeader(shard); @@ -235,11 +251,11 @@ public class IntegrationTestKit extends ShardTestKit { fail("Member(s) " + otherMembersSet + " are not Up"); } - public static ActorRef findLocalShard(final ActorContext actorContext, final String shardName) { + public static ActorRef findLocalShard(final ActorUtils actorUtils, final String shardName) { ActorRef shard = null; for (int i = 0; i < 20 * 5 && shard == null; i++) { Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); - Optional shardReply = actorContext.findLocalShard(shardName); + Optional shardReply = actorUtils.findLocalShard(shardName); if (shardReply.isPresent()) { shard = shardReply.get(); } @@ -247,11 +263,11 @@ public class IntegrationTestKit extends ShardTestKit { return shard; } - public static void waitUntilShardIsDown(final ActorContext actorContext, final String shardName) { + public static void waitUntilShardIsDown(final ActorUtils actorUtils, final String shardName) { for (int i = 0; i < 20 * 5 ; i++) { LOG.debug("Waiting for shard down {}", shardName); Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); - Optional shardReply = actorContext.findLocalShard(shardName); + Optional shardReply = actorUtils.findLocalShard(shardName); if (!shardReply.isPresent()) { return; } @@ -262,15 +278,15 @@ public class IntegrationTestKit extends ShardTestKit { public static void verifyShardStats(final AbstractDataStore datastore, final String shardName, final ShardStatsVerifier verifier) throws Exception { - ActorContext actorContext = datastore.getActorContext(); + ActorUtils actorUtils = datastore.getActorUtils(); - Future future = actorContext.findLocalShardAsync(shardName); - ActorRef shardActor = Await.result(future, Duration.create(10, TimeUnit.SECONDS)); + Future future = actorUtils.findLocalShardAsync(shardName); + ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS)); AssertionError lastError = null; Stopwatch sw = Stopwatch.createStarted(); while (sw.elapsed(TimeUnit.SECONDS) <= 5) { - ShardStats shardStats = (ShardStats)actorContext + ShardStats shardStats = (ShardStats)actorUtils .executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE); try { @@ -287,15 +303,15 @@ public class IntegrationTestKit extends ShardTestKit { public static void verifyShardState(final AbstractDataStore datastore, final String shardName, final Consumer verifier) throws Exception { - ActorContext actorContext = datastore.getActorContext(); + ActorUtils actorUtils = datastore.getActorUtils(); - Future future = actorContext.findLocalShardAsync(shardName); - ActorRef shardActor = Await.result(future, Duration.create(10, TimeUnit.SECONDS)); + Future future = actorUtils.findLocalShardAsync(shardName); + ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS)); AssertionError lastError = null; Stopwatch sw = Stopwatch.createStarted(); while (sw.elapsed(TimeUnit.SECONDS) <= 5) { - OnDemandShardState shardState = (OnDemandShardState)actorContext + OnDemandShardState shardState = (OnDemandShardState)actorUtils .executeOperation(shardActor, GetOnDemandRaftState.INSTANCE); try { @@ -335,13 +351,13 @@ public class IntegrationTestKit extends ShardTestKit { DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction(); Optional> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS); - assertEquals("isPresent", true, optional.isPresent()); + assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", nodeToWrite, optional.get()); } public void doCommit(final DOMStoreThreePhaseCommitCohort cohort) throws Exception { Boolean canCommit = cohort.canCommit().get(commitTimeout, TimeUnit.SECONDS); - assertEquals("canCommit", true, canCommit); + assertEquals("canCommit", Boolean.TRUE, canCommit); cohort.preCommit().get(5, TimeUnit.SECONDS); cohort.commit().get(5, TimeUnit.SECONDS); } @@ -349,14 +365,13 @@ public class IntegrationTestKit extends ShardTestKit { void doCommit(final ListenableFuture canCommitFuture, final DOMStoreThreePhaseCommitCohort cohort) throws Exception { Boolean canCommit = canCommitFuture.get(commitTimeout, TimeUnit.SECONDS); - assertEquals("canCommit", true, canCommit); + assertEquals("canCommit", Boolean.TRUE, canCommit); cohort.preCommit().get(5, TimeUnit.SECONDS); cohort.commit().get(5, TimeUnit.SECONDS); } @SuppressWarnings("checkstyle:IllegalCatch") - void assertExceptionOnCall(final Callable callable, final Class expType) - throws Exception { + void assertExceptionOnCall(final Callable callable, final Class expType) { try { callable.call(); fail("Expected " + expType.getSimpleName()); @@ -366,7 +381,7 @@ public class IntegrationTestKit extends ShardTestKit { } void assertExceptionOnTxChainCreates(final DOMStoreTransactionChain txChain, - final Class expType) throws Exception { + final Class expType) { assertExceptionOnCall(() -> { txChain.newWriteOnlyTransaction(); return null;