package org.opendaylight.controller.cluster.datastore;
+import akka.actor.ActorRef;
import akka.actor.ActorSystem;
-import akka.event.Logging;
-import akka.testkit.JavaTestKit;
-
+import akka.actor.PoisonPill;
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 com.google.common.util.concurrent.Uninterruptibles;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
+import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
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;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
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.controller.sal.core.spi.data.DOMStoreWriteTransaction;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
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 org.opendaylight.yangtools.yang.model.api.SchemaContext;
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 extends AbstractActorTest {
-public class DistributedDataStoreIntegrationTest {
+ @Test
+ public void testWriteTransactionWithSingleShard() throws Exception{
+ System.setProperty("shard.persistent", "true");
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("transactionIntegrationTest", "test-1");
- private static ActorSystem system;
+ testWriteTransaction(dataStore, TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- @Before
- public void setUp() throws IOException {
- File journal = new File("journal");
+ testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
+ ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
- if(journal.exists()) {
- FileUtils.deleteDirectory(journal);
- }
+ cleanup(dataStore);
+ }};
+ }
+ @Test
+ public void testWriteTransactionWithMultipleShards() throws Exception{
+ System.setProperty("shard.persistent", "true");
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("testWriteTransactionWithMultipleShards", "cars-1", "people-1");
- System.setProperty("shard.persistent", "false");
- system = ActorSystem.create("test");
- }
+ DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
+ assertNotNull("newWriteOnlyTransaction returned null", writeTx);
- @After
- public void tearDown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
+ YangInstanceIdentifier nodePath1 = CarsModel.BASE_PATH;
+ NormalizedNode<?, ?> nodeToWrite1 = CarsModel.emptyContainer();
+ writeTx.write(nodePath1, nodeToWrite1);
+
+ YangInstanceIdentifier nodePath2 = PeopleModel.BASE_PATH;
+ NormalizedNode<?, ?> nodeToWrite2 = PeopleModel.emptyContainer();
+ writeTx.write(nodePath2, nodeToWrite2);
+
+ DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
+
+ Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
+ assertEquals("canCommit", true, canCommit);
+ cohort.preCommit().get(5, TimeUnit.SECONDS);
+ cohort.commit().get(5, TimeUnit.SECONDS);
+
+ // 5. Verify the data in the store
- protected ActorSystem getSystem() {
- return system;
+ DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
+
+ Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath1).get();
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", nodeToWrite1, optional.get());
+
+ optional = readTx.read(nodePath2).get();
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", nodeToWrite2, optional.get());
+
+ cleanup(dataStore);
+ }};
}
@Test
- public void integrationTest() throws Exception {
- final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf");
- ShardStrategyFactory.setConfiguration(configuration);
+ public void testReadWriteTransaction() throws Exception{
+ System.setProperty("shard.persistent", "true");
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("testReadWriteTransaction", "test-1");
+
+ // 1. Create a read-write Tx
+ DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
+ assertNotNull("newReadWriteTransaction returned null", readWriteTx);
+ // 2. Write some data
- new JavaTestKit(getSystem()) {
- {
+ YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
+ NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
+ readWriteTx.write(nodePath, nodeToWrite );
- new Within(duration("10 seconds")) {
- @Override
- protected void run() {
- try {
- final DistributedDataStore distributedDataStore =
- new DistributedDataStore(getSystem(), "config",
- new MockClusterWrapper(), configuration,
- new DatastoreContext());
+ // 3. Read the data from Tx
- distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
+ Boolean exists = readWriteTx.exists(nodePath).checkedGet(5, TimeUnit.SECONDS);
+ assertEquals("exists", true, exists);
- // Wait for a specific log message to show up
- final boolean result =
- new JavaTestKit.EventFilter<Boolean>(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();
+ Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", nodeToWrite, optional.get());
- assertEquals(true, result);
+ // 4. Ready the Tx for commit
- DOMStoreReadWriteTransaction transaction =
- distributedDataStore.newReadWriteTransaction();
+ DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
- transaction
- .write(TestModel.TEST_PATH, ImmutableNodes
- .containerNode(TestModel.TEST_QNAME));
+ // 5. Commit the Tx
- ListenableFuture<Optional<NormalizedNode<?, ?>>>
- future =
- transaction.read(TestModel.TEST_PATH);
+ Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
+ assertEquals("canCommit", true, canCommit);
+ cohort.preCommit().get(5, TimeUnit.SECONDS);
+ cohort.commit().get(5, TimeUnit.SECONDS);
- Optional<NormalizedNode<?, ?>> optional =
- future.get();
+ // 6. Verify the data in the store
- Assert.assertTrue("Node not found", optional.isPresent());
+ DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
- NormalizedNode<?, ?> normalizedNode =
- optional.get();
+ optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", nodeToWrite, optional.get());
- assertEquals(TestModel.TEST_QNAME,
- normalizedNode.getNodeType());
+ cleanup(dataStore);
+ }};
+ }
- DOMStoreThreePhaseCommitCohort ready =
- transaction.ready();
+ @Test
+ public void testTransactionAbort() throws Exception{
+ System.setProperty("shard.persistent", "true");
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("transactionAbortIntegrationTest", "test-1");
- ListenableFuture<Boolean> canCommit =
- ready.canCommit();
+ DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
+ assertNotNull("newWriteOnlyTransaction returned null", writeTx);
- assertTrue(canCommit.get(5, TimeUnit.SECONDS));
+ writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- ListenableFuture<Void> preCommit =
- ready.preCommit();
+ DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
- preCommit.get(5, TimeUnit.SECONDS);
+ cohort.canCommit().get(5, TimeUnit.SECONDS);
- ListenableFuture<Void> commit = ready.commit();
+ cohort.abort().get(5, TimeUnit.SECONDS);
- commit.get(5, TimeUnit.SECONDS);
- } catch (ExecutionException | TimeoutException | InterruptedException e){
- fail(e.getMessage());
- }
- }
- };
- }
- };
+ testWriteTransaction(dataStore, TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ cleanup(dataStore);
+ }};
}
@Test
- public void transactionChainIntegrationTest() throws Exception {
- final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf");
- ShardStrategyFactory.setConfiguration(configuration);
+ public void testTransactionChain() throws Exception{
+ System.setProperty("shard.persistent", "true");
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("transactionChainIntegrationTest", "test-1");
+ // 1. Create a Tx chain and write-only Tx
+ DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
- new JavaTestKit(getSystem()) {
- {
+ DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
+ assertNotNull("newWriteOnlyTransaction returned null", writeTx);
- new Within(duration("10 seconds")) {
- @Override
- protected void run() {
- try {
- final DistributedDataStore distributedDataStore =
- new DistributedDataStore(getSystem(), "config",
- new MockClusterWrapper(), configuration,
- new DatastoreContext());
+ // 2. Write some data
- distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
+ NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
+ writeTx.write(TestModel.TEST_PATH, containerNode);
- // Wait for a specific log message to show up
- final boolean result =
- new JavaTestKit.EventFilter<Boolean>(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();
+ // 3. Ready the Tx for commit
- assertEquals(true, result);
+ DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
- DOMStoreTransactionChain transactionChain =
- distributedDataStore.createTransactionChain();
+ // 4. Commit the Tx
- DOMStoreReadWriteTransaction transaction =
- transactionChain.newReadWriteTransaction();
+ Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
+ assertEquals("canCommit", true, canCommit);
+ cohort.preCommit().get(5, TimeUnit.SECONDS);
+ cohort.commit().get(5, TimeUnit.SECONDS);
- transaction
- .write(TestModel.TEST_PATH, ImmutableNodes
- .containerNode(TestModel.TEST_QNAME));
+ // 5. Verify the data in the store
- ListenableFuture<Optional<NormalizedNode<?, ?>>>
- future =
- transaction.read(TestModel.TEST_PATH);
+ DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
- Optional<NormalizedNode<?, ?>> optional =
- future.get();
+ Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", containerNode, optional.get());
- Assert.assertTrue("Node not found", optional.isPresent());
+ txChain.close();
- NormalizedNode<?, ?> normalizedNode =
- optional.get();
+ cleanup(dataStore);
+ }};
+ }
- assertEquals(TestModel.TEST_QNAME,
- normalizedNode.getNodeType());
+ @Test
+ public void testChangeListenerRegistration() throws Exception{
+ new IntegrationTestKit(getSystem()) {{
+ DistributedDataStore dataStore =
+ setupDistributedDataStore("testChangeListenerRegistration", "test-1");
- DOMStoreThreePhaseCommitCohort ready =
- transaction.ready();
+ MockDataChangeListener listener = new MockDataChangeListener(3);
- ListenableFuture<Boolean> canCommit =
- ready.canCommit();
+ ListenerRegistration<MockDataChangeListener>
+ listenerReg = dataStore.registerChangeListener(TestModel.TEST_PATH, listener,
+ DataChangeScope.SUBTREE);
- assertTrue(canCommit.get(5, TimeUnit.SECONDS));
+ assertNotNull("registerChangeListener returned null", listenerReg);
- ListenableFuture<Void> preCommit =
- ready.preCommit();
+ testWriteTransaction(dataStore, TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- preCommit.get(5, TimeUnit.SECONDS);
+ testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
+ ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
- ListenableFuture<Void> commit = ready.commit();
+ YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
+ nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
+ testWriteTransaction(dataStore, listPath,
+ ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
- commit.get(5, TimeUnit.SECONDS);
+ listener.waitForChangeEvents(TestModel.TEST_PATH, TestModel.OUTER_LIST_PATH, listPath );
- transactionChain.close();
- } catch (ExecutionException | TimeoutException | InterruptedException e){
- fail(e.getMessage());
- }
- }
- };
- }
- };
+ listenerReg.close();
- }
+ testWriteTransaction(dataStore, YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
+ nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
+ ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
+ listener.expectNoMoreChanges("Received unexpected change after close");
- //FIXME : Disabling test because it's flaky
- //@Test
- public void integrationTestWithMultiShardConfiguration()
- throws ExecutionException, InterruptedException, TimeoutException {
- final Configuration configuration = new ConfigurationImpl("module-shards.conf", "modules.conf");
+ cleanup(dataStore);
+ }};
+ }
- ShardStrategyFactory.setConfiguration(configuration);
+ class IntegrationTestKit extends ShardTestKit {
- new JavaTestKit(getSystem()) {
- {
+ IntegrationTestKit(ActorSystem actorSystem) {
+ super(actorSystem);
+ }
- new Within(duration("10 seconds")) {
- @Override
- protected void run() {
- try {
- final DistributedDataStore distributedDataStore =
- new DistributedDataStore(getSystem(), "config",
- new MockClusterWrapper(), configuration, null);
+ DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) {
+ MockClusterWrapper cluster = new MockClusterWrapper();
+ Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
+ ShardStrategyFactory.setConfiguration(config);
+
+ DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
+ DistributedDataStore dataStore = new DistributedDataStore(getSystem(), typeName, cluster,
+ config, datastoreContext);
+
+ SchemaContext schemaContext = SchemaContextHelper.full();
+ dataStore.onGlobalContextUpdated(schemaContext);
+
+ for(String shardName: shardNames) {
+ ActorRef shard = null;
+ for(int i = 0; i < 20 * 5 && shard == null; i++) {
+ Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
+ Optional<ActorRef> shardReply = dataStore.getActorContext().findLocalShard(shardName);
+ if(shardReply.isPresent()) {
+ shard = shardReply.get();
+ }
+ }
- distributedDataStore.onGlobalContextUpdated(
- SchemaContextHelper.full());
+ assertNotNull("Shard was not created", shard);
- // Wait for a specific log message to show up
- final boolean result =
- new JavaTestKit.EventFilter<Boolean>(
- 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();
+ System.out.println("!!!!!!shard: "+shard.path().toString());
+ waitUntilLeader(shard);
+ }
- Thread.sleep(1000);
+ return dataStore;
+ }
+ void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath,
+ NormalizedNode<?, ?> nodeToWrite) throws Exception {
- DOMStoreReadWriteTransaction transaction =
- distributedDataStore.newReadWriteTransaction();
+ // 1. Create a write-only Tx
- transaction.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
- transaction.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
+ DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
+ assertNotNull("newWriteOnlyTransaction returned null", writeTx);
- DOMStoreThreePhaseCommitCohort ready = transaction.ready();
+ // 2. Write some data
- ListenableFuture<Boolean> canCommit = ready.canCommit();
+ writeTx.write(nodePath, nodeToWrite);
- assertTrue(canCommit.get(5, TimeUnit.SECONDS));
+ // 3. Ready the Tx for commit
- ListenableFuture<Void> preCommit = ready.preCommit();
+ DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
- preCommit.get(5, TimeUnit.SECONDS);
+ // 4. Commit the Tx
- ListenableFuture<Void> commit = ready.commit();
+ Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
+ assertEquals("canCommit", true, canCommit);
+ cohort.preCommit().get(5, TimeUnit.SECONDS);
+ cohort.commit().get(5, TimeUnit.SECONDS);
- commit.get(5, TimeUnit.SECONDS);
+ // 5. Verify the data in the store
- assertEquals(true, result);
- } catch(ExecutionException | TimeoutException | InterruptedException e){
- fail(e.getMessage());
- }
- }
- };
- }
- };
+ DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
+ Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
+ assertEquals("isPresent", true, optional.isPresent());
+ assertEquals("Data node", nodeToWrite, optional.get());
+ }
+ void cleanup(DistributedDataStore dataStore) {
+ dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), null);
+ }
}
}
Code Review / controller.git / blobdiff