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%2FDistributedDataStoreIntegrationTest.java;h=ec8aee2b09d4dfaf4ed4b5b732a9168783beb73c;hb=739927d16b298c15accbff118e38293416a3dfa7;hp=b5e3d24ef6d536d6afe2c9a6abc111bbb4b8688b;hpb=287dd97fc2375eb32186515ecfdac32ee1a36d83;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java index b5e3d24ef6..ec8aee2b09 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java @@ -1,9 +1,15 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorSystem; +import akka.event.Logging; import akka.testkit.JavaTestKit; + import com.google.common.base.Optional; import com.google.common.util.concurrent.ListenableFuture; + +import junit.framework.Assert; + +import org.apache.commons.io.FileUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -15,20 +21,33 @@ import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelpe import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; +import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; +import java.io.File; +import java.io.IOException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static junit.framework.Assert.assertEquals; import static junit.framework.Assert.assertTrue; +import static junit.framework.Assert.fail; -public class DistributedDataStoreIntegrationTest{ +public class DistributedDataStoreIntegrationTest { private static ActorSystem system; @Before - public void setUp() { + public void setUp() throws IOException { + File journal = new File("journal"); + + if(journal.exists()) { + FileUtils.deleteDirectory(journal); + } + + System.setProperty("shard.persistent", "false"); system = ActorSystem.create("test"); } @@ -45,79 +64,246 @@ public class DistributedDataStoreIntegrationTest{ @Test public void integrationTest() throws Exception { - Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf"); + final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf"); ShardStrategyFactory.setConfiguration(configuration); - DistributedDataStore distributedDataStore = - new DistributedDataStore(getSystem(), "config", new MockClusterWrapper(), configuration); - distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext()); - Thread.sleep(1000); - DOMStoreReadWriteTransaction transaction = - distributedDataStore.newReadWriteTransaction(); + new JavaTestKit(getSystem()) { + { - transaction.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + new Within(duration("10 seconds")) { + @Override + protected void run() { + try { + final DistributedDataStore distributedDataStore = + new DistributedDataStore(getSystem(), "config", + new MockClusterWrapper(), configuration, + new DatastoreContext()); - ListenableFuture>> future = - transaction.read(TestModel.TEST_PATH); + distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext()); - Optional> optional = future.get(); + // Wait for a specific log message to show up + final boolean result = + new JavaTestKit.EventFilter(Logging.Info.class + ) { + @Override + protected Boolean run() { + return true; + } + }.from("akka://test/user/shardmanager-config/member-1-shard-test-1-config") + .message("Switching from state Candidate to Leader") + .occurrences(1).exec(); - NormalizedNode normalizedNode = optional.get(); + assertEquals(true, result); - assertEquals(TestModel.TEST_QNAME, normalizedNode.getNodeType()); + DOMStoreReadWriteTransaction transaction = + distributedDataStore.newReadWriteTransaction(); - DOMStoreThreePhaseCommitCohort ready = transaction.ready(); + transaction + .write(TestModel.TEST_PATH, ImmutableNodes + .containerNode(TestModel.TEST_QNAME)); - ListenableFuture canCommit = ready.canCommit(); + ListenableFuture>> + future = + transaction.read(TestModel.TEST_PATH); - assertTrue(canCommit.get()); + Optional> optional = + future.get(); - ListenableFuture preCommit = ready.preCommit(); + Assert.assertTrue("Node not found", optional.isPresent()); - preCommit.get(); + NormalizedNode normalizedNode = + optional.get(); - ListenableFuture commit = ready.commit(); + assertEquals(TestModel.TEST_QNAME, + normalizedNode.getNodeType()); - commit.get(); + DOMStoreThreePhaseCommitCohort ready = + transaction.ready(); - } + ListenableFuture canCommit = + ready.canCommit(); + + assertTrue(canCommit.get(5, TimeUnit.SECONDS)); + + ListenableFuture preCommit = + ready.preCommit(); + preCommit.get(5, TimeUnit.SECONDS); + + ListenableFuture commit = ready.commit(); + + commit.get(5, TimeUnit.SECONDS); + } catch (ExecutionException | TimeoutException | InterruptedException e){ + fail(e.getMessage()); + } + } + }; + } + }; + + } @Test + public void transactionChainIntegrationTest() throws Exception { + final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf"); + ShardStrategyFactory.setConfiguration(configuration); + + + + new JavaTestKit(getSystem()) { + { + + new Within(duration("10 seconds")) { + @Override + protected void run() { + try { + final DistributedDataStore distributedDataStore = + new DistributedDataStore(getSystem(), "config", + new MockClusterWrapper(), configuration, + new DatastoreContext()); + + distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext()); + + // Wait for a specific log message to show up + final boolean result = + new JavaTestKit.EventFilter(Logging.Info.class + ) { + @Override + protected Boolean run() { + return true; + } + }.from("akka://test/user/shardmanager-config/member-1-shard-test-1-config") + .message("Switching from state Candidate to Leader") + .occurrences(1).exec(); + + assertEquals(true, result); + + DOMStoreTransactionChain transactionChain = + distributedDataStore.createTransactionChain(); + + DOMStoreReadWriteTransaction transaction = + transactionChain.newReadWriteTransaction(); + + transaction + .write(TestModel.TEST_PATH, ImmutableNodes + .containerNode(TestModel.TEST_QNAME)); + + ListenableFuture>> + future = + transaction.read(TestModel.TEST_PATH); + + Optional> optional = + future.get(); + + Assert.assertTrue("Node not found", optional.isPresent()); + + NormalizedNode normalizedNode = + optional.get(); + + assertEquals(TestModel.TEST_QNAME, + normalizedNode.getNodeType()); + + DOMStoreThreePhaseCommitCohort ready = + transaction.ready(); + + ListenableFuture canCommit = + ready.canCommit(); + + assertTrue(canCommit.get(5, TimeUnit.SECONDS)); + + ListenableFuture preCommit = + ready.preCommit(); + + preCommit.get(5, TimeUnit.SECONDS); + + ListenableFuture commit = ready.commit(); + + commit.get(5, TimeUnit.SECONDS); + + transactionChain.close(); + } catch (ExecutionException | TimeoutException | InterruptedException e){ + fail(e.getMessage()); + } + } + }; + } + }; + + } + + + //FIXME : Disabling test because it's flaky + //@Test public void integrationTestWithMultiShardConfiguration() - throws ExecutionException, InterruptedException { - Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf"); + throws ExecutionException, InterruptedException, TimeoutException { + final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf"); ShardStrategyFactory.setConfiguration(configuration); - DistributedDataStore distributedDataStore = - new DistributedDataStore(getSystem(), "config", new MockClusterWrapper(), configuration); + + new JavaTestKit(getSystem()) { + { + + new Within(duration("10 seconds")) { + @Override + protected void run() { + try { + final DistributedDataStore distributedDataStore = + new DistributedDataStore(getSystem(), "config", + new MockClusterWrapper(), configuration, null); + + distributedDataStore.onGlobalContextUpdated( + SchemaContextHelper.full()); + + // Wait for a specific log message to show up + final boolean result = + new JavaTestKit.EventFilter( + Logging.Info.class + ) { + @Override + protected Boolean run() { + return true; + } + }.from( + "akka://test/user/shardmanager-config/member-1-shard-cars-1-config") + .message( + "Switching from state Candidate to Leader") + .occurrences(1) + .exec(); + + Thread.sleep(1000); - distributedDataStore.onGlobalContextUpdated(SchemaContextHelper.full()); + DOMStoreReadWriteTransaction transaction = + distributedDataStore.newReadWriteTransaction(); - Thread.sleep(1000); + transaction.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + transaction.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer()); - DOMStoreReadWriteTransaction transaction = - distributedDataStore.newReadWriteTransaction(); + DOMStoreThreePhaseCommitCohort ready = transaction.ready(); - transaction.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); - transaction.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer()); + ListenableFuture canCommit = ready.canCommit(); - DOMStoreThreePhaseCommitCohort ready = transaction.ready(); + assertTrue(canCommit.get(5, TimeUnit.SECONDS)); - ListenableFuture canCommit = ready.canCommit(); + ListenableFuture preCommit = ready.preCommit(); - assertTrue(canCommit.get()); + preCommit.get(5, TimeUnit.SECONDS); - ListenableFuture preCommit = ready.preCommit(); + ListenableFuture commit = ready.commit(); - preCommit.get(); + commit.get(5, TimeUnit.SECONDS); - ListenableFuture commit = ready.commit(); + assertEquals(true, result); + } catch(ExecutionException | TimeoutException | InterruptedException e){ + fail(e.getMessage()); + } + } + }; + } + }; - commit.get(); }