1 package org.opendaylight.controller.cluster.datastore;
3 import static org.junit.Assert.assertEquals;
4 import static org.junit.Assert.assertNotNull;
5 import static org.junit.Assert.fail;
6 import akka.actor.ActorRef;
7 import akka.actor.ActorSystem;
8 import akka.actor.PoisonPill;
9 import com.google.common.base.Optional;
10 import com.google.common.util.concurrent.CheckedFuture;
11 import com.google.common.util.concurrent.Uninterruptibles;
12 import java.util.concurrent.Callable;
13 import java.util.concurrent.CountDownLatch;
14 import java.util.concurrent.ExecutionException;
15 import java.util.concurrent.TimeUnit;
16 import java.util.concurrent.atomic.AtomicReference;
17 import org.junit.Test;
18 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
19 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
20 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
21 import org.opendaylight.controller.cluster.datastore.utils.InMemoryJournal;
22 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
23 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
24 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
25 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
26 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
27 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
28 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
29 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
30 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
31 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
33 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
35 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
36 import org.opendaylight.yangtools.concepts.ListenerRegistration;
37 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
38 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
40 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
41 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
43 public class DistributedDataStoreIntegrationTest extends AbstractActorTest {
45 private final DatastoreContext.Builder datastoreContextBuilder =
46 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100);
49 public void testWriteTransactionWithSingleShard() throws Exception{
50 new IntegrationTestKit(getSystem()) {{
51 DistributedDataStore dataStore =
52 setupDistributedDataStore("transactionIntegrationTest", "test-1");
54 testWriteTransaction(dataStore, TestModel.TEST_PATH,
55 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
57 testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
58 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
65 public void testWriteTransactionWithMultipleShards() throws Exception{
66 new IntegrationTestKit(getSystem()) {{
67 DistributedDataStore dataStore =
68 setupDistributedDataStore("testWriteTransactionWithMultipleShards", "cars-1", "people-1");
70 DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
71 assertNotNull("newWriteOnlyTransaction returned null", writeTx);
73 YangInstanceIdentifier nodePath1 = CarsModel.BASE_PATH;
74 NormalizedNode<?, ?> nodeToWrite1 = CarsModel.emptyContainer();
75 writeTx.write(nodePath1, nodeToWrite1);
77 YangInstanceIdentifier nodePath2 = PeopleModel.BASE_PATH;
78 NormalizedNode<?, ?> nodeToWrite2 = PeopleModel.emptyContainer();
79 writeTx.write(nodePath2, nodeToWrite2);
81 DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
85 // Verify the data in the store
87 DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
89 Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath1).get();
90 assertEquals("isPresent", true, optional.isPresent());
91 assertEquals("Data node", nodeToWrite1, optional.get());
93 optional = readTx.read(nodePath2).get();
94 assertEquals("isPresent", true, optional.isPresent());
95 assertEquals("Data node", nodeToWrite2, optional.get());
102 public void testReadWriteTransaction() throws Exception{
103 System.setProperty("shard.persistent", "true");
104 new IntegrationTestKit(getSystem()) {{
105 DistributedDataStore dataStore =
106 setupDistributedDataStore("testReadWriteTransaction", "test-1");
108 // 1. Create a read-write Tx
110 DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
111 assertNotNull("newReadWriteTransaction returned null", readWriteTx);
113 // 2. Write some data
115 YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
116 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
117 readWriteTx.write(nodePath, nodeToWrite );
119 // 3. Read the data from Tx
121 Boolean exists = readWriteTx.exists(nodePath).checkedGet(5, TimeUnit.SECONDS);
122 assertEquals("exists", true, exists);
124 Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
125 assertEquals("isPresent", true, optional.isPresent());
126 assertEquals("Data node", nodeToWrite, optional.get());
128 // 4. Ready the Tx for commit
130 DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
136 // 6. Verify the data in the store
138 DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
140 optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
141 assertEquals("isPresent", true, optional.isPresent());
142 assertEquals("Data node", nodeToWrite, optional.get());
149 public void testTransactionWritesWithShardNotInitiallyReady() throws Exception{
150 new IntegrationTestKit(getSystem()) {{
151 String testName = "testTransactionWritesWithShardNotInitiallyReady";
152 String shardName = "test-1";
154 // Setup the InMemoryJournal to block shard recovery to ensure the shard isn't
155 // initialized until we create and submit the write the Tx.
156 String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
157 CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
158 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
160 DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
162 // Create the write Tx
164 final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
165 assertNotNull("newReadWriteTransaction returned null", writeTx);
167 // Do some modification operations and ready the Tx on a separate thread.
169 final YangInstanceIdentifier listEntryPath = YangInstanceIdentifier.builder(
170 TestModel.OUTER_LIST_PATH).nodeWithKey(TestModel.OUTER_LIST_QNAME,
171 TestModel.ID_QNAME, 1).build();
173 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
174 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
175 final CountDownLatch txReady = new CountDownLatch(1);
176 Thread txThread = new Thread() {
180 writeTx.write(TestModel.TEST_PATH,
181 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
183 writeTx.merge(TestModel.OUTER_LIST_PATH, ImmutableNodes.mapNodeBuilder(
184 TestModel.OUTER_LIST_QNAME).build());
186 writeTx.write(listEntryPath, ImmutableNodes.mapEntry(
187 TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
189 writeTx.delete(listEntryPath);
191 txCohort.set(writeTx.ready());
192 } catch(Exception e) {
203 // Wait for the Tx operations to complete.
205 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
206 if(caughtEx.get() != null) {
207 throw caughtEx.get();
210 assertEquals("Tx ready", true, done);
212 // At this point the Tx operations should be waiting for the shard to initialize so
213 // trigger the latch to let the shard recovery to continue.
215 blockRecoveryLatch.countDown();
217 // Wait for the Tx commit to complete.
219 doCommit(txCohort.get());
221 // Verify the data in the store
223 DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
225 Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).
226 get(5, TimeUnit.SECONDS);
227 assertEquals("isPresent", true, optional.isPresent());
229 optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
230 assertEquals("isPresent", true, optional.isPresent());
232 optional = readTx.read(listEntryPath).get(5, TimeUnit.SECONDS);
233 assertEquals("isPresent", false, optional.isPresent());
240 public void testTransactionReadsWithShardNotInitiallyReady() throws Exception{
241 new IntegrationTestKit(getSystem()) {{
242 String testName = "testTransactionReadsWithShardNotInitiallyReady";
243 String shardName = "test-1";
245 // Setup the InMemoryJournal to block shard recovery to ensure the shard isn't
246 // initialized until we create the Tx.
247 String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
248 CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
249 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
251 DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
253 // Create the read-write Tx
255 final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
256 assertNotNull("newReadWriteTransaction returned null", readWriteTx);
258 // Do some reads on the Tx on a separate thread.
260 final AtomicReference<CheckedFuture<Boolean, ReadFailedException>> txExistsFuture =
261 new AtomicReference<>();
262 final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
263 txReadFuture = new AtomicReference<>();
264 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
265 final CountDownLatch txReadsDone = new CountDownLatch(1);
266 Thread txThread = new Thread() {
270 readWriteTx.write(TestModel.TEST_PATH,
271 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
273 txExistsFuture.set(readWriteTx.exists(TestModel.TEST_PATH));
275 txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
276 } catch(Exception e) {
280 txReadsDone.countDown();
287 // Wait for the Tx operations to complete.
289 boolean done = Uninterruptibles.awaitUninterruptibly(txReadsDone, 5, TimeUnit.SECONDS);
290 if(caughtEx.get() != null) {
291 throw caughtEx.get();
294 assertEquals("Tx reads done", true, done);
296 // At this point the Tx operations should be waiting for the shard to initialize so
297 // trigger the latch to let the shard recovery to continue.
299 blockRecoveryLatch.countDown();
301 // Wait for the reads to complete and verify.
303 assertEquals("exists", true, txExistsFuture.get().checkedGet(5, TimeUnit.SECONDS));
304 assertEquals("read", true, txReadFuture.get().checkedGet(5, TimeUnit.SECONDS).isPresent());
312 @Test(expected=NotInitializedException.class)
313 public void testTransactionCommitFailureWithShardNotInitialized() throws Throwable{
314 new IntegrationTestKit(getSystem()) {{
315 String testName = "testTransactionCommitFailureWithShardNotInitialized";
316 String shardName = "test-1";
318 // Set the shard initialization timeout low for the test.
320 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
322 // Setup the InMemoryJournal to block shard recovery indefinitely.
324 String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
325 CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
326 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
328 DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
330 // Create the write Tx
332 final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
333 assertNotNull("newReadWriteTransaction returned null", writeTx);
335 // Do some modifications and ready the Tx on a separate thread.
337 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
338 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
339 final CountDownLatch txReady = new CountDownLatch(1);
340 Thread txThread = new Thread() {
344 writeTx.write(TestModel.TEST_PATH,
345 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
347 txCohort.set(writeTx.ready());
348 } catch(Exception e) {
359 // Wait for the Tx operations to complete.
361 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
362 if(caughtEx.get() != null) {
363 throw caughtEx.get();
366 assertEquals("Tx ready", true, done);
368 // Wait for the commit to complete. Since the shard never initialized, the Tx should
369 // have timed out and throw an appropriate exception cause.
372 txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
373 } catch(ExecutionException e) {
376 blockRecoveryLatch.countDown();
382 @Test(expected=NotInitializedException.class)
383 public void testTransactionReadFailureWithShardNotInitialized() throws Throwable{
384 new IntegrationTestKit(getSystem()) {{
385 String testName = "testTransactionReadFailureWithShardNotInitialized";
386 String shardName = "test-1";
388 // Set the shard initialization timeout low for the test.
390 datastoreContextBuilder.shardInitializationTimeout(300, TimeUnit.MILLISECONDS);
392 // Setup the InMemoryJournal to block shard recovery indefinitely.
394 String persistentID = String.format("member-1-shard-%s-%s", shardName, testName);
395 CountDownLatch blockRecoveryLatch = new CountDownLatch(1);
396 InMemoryJournal.addBlockReadMessagesLatch(persistentID, blockRecoveryLatch);
398 DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
400 // Create the read-write Tx
402 final DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
403 assertNotNull("newReadWriteTransaction returned null", readWriteTx);
405 // Do a read on the Tx on a separate thread.
407 final AtomicReference<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>>
408 txReadFuture = new AtomicReference<>();
409 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
410 final CountDownLatch txReadDone = new CountDownLatch(1);
411 Thread txThread = new Thread() {
415 readWriteTx.write(TestModel.TEST_PATH,
416 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
418 txReadFuture.set(readWriteTx.read(TestModel.TEST_PATH));
421 } catch(Exception e) {
425 txReadDone.countDown();
432 // Wait for the Tx operations to complete.
434 boolean done = Uninterruptibles.awaitUninterruptibly(txReadDone, 5, TimeUnit.SECONDS);
435 if(caughtEx.get() != null) {
436 throw caughtEx.get();
439 assertEquals("Tx read done", true, done);
441 // Wait for the read to complete. Since the shard never initialized, the Tx should
442 // have timed out and throw an appropriate exception cause.
445 txReadFuture.get().checkedGet(5, TimeUnit.SECONDS);
446 } catch(ReadFailedException e) {
449 blockRecoveryLatch.countDown();
455 @Test(expected=NoShardLeaderException.class)
456 public void testTransactionCommitFailureWithNoShardLeader() throws Throwable{
457 new IntegrationTestKit(getSystem()) {{
458 String testName = "testTransactionCommitFailureWithNoShardLeader";
459 String shardName = "test-1";
461 // We don't want the shard to become the leader so prevent shard election from completing
462 // by setting the election timeout, which is based on the heartbeat interval, really high.
464 datastoreContextBuilder.shardHeartbeatIntervalInMillis(30000);
466 // Set the leader election timeout low for the test.
468 datastoreContextBuilder.shardLeaderElectionTimeout(1, TimeUnit.MILLISECONDS);
470 DistributedDataStore dataStore = setupDistributedDataStore(testName, false, shardName);
472 // Create the write Tx.
474 final DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
475 assertNotNull("newReadWriteTransaction returned null", writeTx);
477 // Do some modifications and ready the Tx on a separate thread.
479 final AtomicReference<DOMStoreThreePhaseCommitCohort> txCohort = new AtomicReference<>();
480 final AtomicReference<Exception> caughtEx = new AtomicReference<>();
481 final CountDownLatch txReady = new CountDownLatch(1);
482 Thread txThread = new Thread() {
486 writeTx.write(TestModel.TEST_PATH,
487 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
489 txCohort.set(writeTx.ready());
490 } catch(Exception e) {
501 // Wait for the Tx operations to complete.
503 boolean done = Uninterruptibles.awaitUninterruptibly(txReady, 5, TimeUnit.SECONDS);
504 if(caughtEx.get() != null) {
505 throw caughtEx.get();
508 assertEquals("Tx ready", true, done);
510 // Wait for the commit to complete. Since no shard leader was elected in time, the Tx
511 // should have timed out and throw an appropriate exception cause.
514 txCohort.get().canCommit().get(5, TimeUnit.SECONDS);
515 } catch(ExecutionException e) {
524 public void testTransactionAbort() throws Exception{
525 System.setProperty("shard.persistent", "true");
526 new IntegrationTestKit(getSystem()) {{
527 DistributedDataStore dataStore =
528 setupDistributedDataStore("transactionAbortIntegrationTest", "test-1");
530 DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
531 assertNotNull("newWriteOnlyTransaction returned null", writeTx);
533 writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
535 DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
537 cohort.canCommit().get(5, TimeUnit.SECONDS);
539 cohort.abort().get(5, TimeUnit.SECONDS);
541 testWriteTransaction(dataStore, TestModel.TEST_PATH,
542 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
549 public void testTransactionChain() throws Exception{
550 new IntegrationTestKit(getSystem()) {{
551 DistributedDataStore dataStore = setupDistributedDataStore("testTransactionChain", "test-1");
553 // 1. Create a Tx chain and write-only Tx
555 DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
557 DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
558 assertNotNull("newWriteOnlyTransaction returned null", writeTx);
560 // 2. Write some data
562 NormalizedNode<?, ?> testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
563 writeTx.write(TestModel.TEST_PATH, testNode);
565 // 3. Ready the Tx for commit
567 final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
569 // 4. Commit the Tx on another thread that first waits for the second read Tx.
571 final CountDownLatch continueCommit1 = new CountDownLatch(1);
572 final CountDownLatch commit1Done = new CountDownLatch(1);
573 final AtomicReference<Exception> commit1Error = new AtomicReference<>();
578 continueCommit1.await();
580 } catch (Exception e) {
583 commit1Done.countDown();
588 // 5. Create a new read Tx from the chain to read and verify the data from the first
589 // Tx is visible after being readied.
591 DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
592 Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
593 assertEquals("isPresent", true, optional.isPresent());
594 assertEquals("Data node", testNode, optional.get());
596 // 6. Create a new RW Tx from the chain, write more data, and ready it
598 DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
599 MapNode outerNode = ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build();
600 rwTx.write(TestModel.OUTER_LIST_PATH, outerNode);
602 DOMStoreThreePhaseCommitCohort cohort2 = rwTx.ready();
604 // 7. Create a new read Tx from the chain to read the data from the last RW Tx to
605 // verify it is visible.
607 readTx = txChain.newReadWriteTransaction();
608 optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
609 assertEquals("isPresent", true, optional.isPresent());
610 assertEquals("Data node", outerNode, optional.get());
612 // 8. Wait for the 2 commits to complete and close the chain.
614 continueCommit1.countDown();
615 Uninterruptibles.awaitUninterruptibly(commit1Done, 5, TimeUnit.SECONDS);
617 if(commit1Error.get() != null) {
618 throw commit1Error.get();
625 // 9. Create a new read Tx from the data store and verify committed data.
627 readTx = dataStore.newReadOnlyTransaction();
628 optional = readTx.read(TestModel.OUTER_LIST_PATH).get(5, TimeUnit.SECONDS);
629 assertEquals("isPresent", true, optional.isPresent());
630 assertEquals("Data node", outerNode, optional.get());
637 public void testCreateChainedTransactionWhenPreviousNotReady() throws Throwable {
638 new IntegrationTestKit(getSystem()) {{
639 DistributedDataStore dataStore = setupDistributedDataStore(
640 "testCreateChainedTransactionWhenPreviousNotReady", "test-1");
642 final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
644 DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
645 assertNotNull("newWriteOnlyTransaction returned null", writeTx);
647 writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
649 // Try to create another Tx of each type - each should fail b/c the previous Tx wasn't
652 assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
657 public void testCreateChainedTransactionAfterClose() throws Throwable {
658 new IntegrationTestKit(getSystem()) {{
659 DistributedDataStore dataStore = setupDistributedDataStore(
660 "testCreateChainedTransactionAfterClose", "test-1");
662 DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
666 // Try to create another Tx of each type - should fail b/c the previous Tx was closed.
668 assertExceptionOnTxChainCreates(txChain, TransactionChainClosedException.class);
673 public void testChangeListenerRegistration() throws Exception{
674 new IntegrationTestKit(getSystem()) {{
675 DistributedDataStore dataStore =
676 setupDistributedDataStore("testChangeListenerRegistration", "test-1");
678 testWriteTransaction(dataStore, TestModel.TEST_PATH,
679 ImmutableNodes.containerNode(TestModel.TEST_QNAME));
681 MockDataChangeListener listener = new MockDataChangeListener(1);
683 ListenerRegistration<MockDataChangeListener>
684 listenerReg = dataStore.registerChangeListener(TestModel.TEST_PATH, listener,
685 DataChangeScope.SUBTREE);
687 assertNotNull("registerChangeListener returned null", listenerReg);
689 // Wait for the initial notification
691 listener.waitForChangeEvents(TestModel.TEST_PATH);
697 testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
698 ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
700 YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
701 nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
702 testWriteTransaction(dataStore, listPath,
703 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
705 // Wait for the 2 updates.
707 listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
711 testWriteTransaction(dataStore, YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
712 nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
713 ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
715 listener.expectNoMoreChanges("Received unexpected change after close");
721 class IntegrationTestKit extends ShardTestKit {
723 IntegrationTestKit(ActorSystem actorSystem) {
727 DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) {
728 return setupDistributedDataStore(typeName, true, shardNames);
731 DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader,
732 String... shardNames) {
733 MockClusterWrapper cluster = new MockClusterWrapper();
734 Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
735 ShardStrategyFactory.setConfiguration(config);
737 DatastoreContext datastoreContext = datastoreContextBuilder.build();
738 DistributedDataStore dataStore = new DistributedDataStore(getSystem(), typeName, cluster,
739 config, datastoreContext);
741 SchemaContext schemaContext = SchemaContextHelper.full();
742 dataStore.onGlobalContextUpdated(schemaContext);
744 if(waitUntilLeader) {
745 for(String shardName: shardNames) {
746 ActorRef shard = null;
747 for(int i = 0; i < 20 * 5 && shard == null; i++) {
748 Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
749 Optional<ActorRef> shardReply = dataStore.getActorContext().findLocalShard(shardName);
750 if(shardReply.isPresent()) {
751 shard = shardReply.get();
755 assertNotNull("Shard was not created", shard);
757 waitUntilLeader(shard);
764 void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath,
765 NormalizedNode<?, ?> nodeToWrite) throws Exception {
767 // 1. Create a write-only Tx
769 DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
770 assertNotNull("newWriteOnlyTransaction returned null", writeTx);
772 // 2. Write some data
774 writeTx.write(nodePath, nodeToWrite);
776 // 3. Ready the Tx for commit
778 DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
784 // 5. Verify the data in the store
786 DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
788 Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
789 assertEquals("isPresent", true, optional.isPresent());
790 assertEquals("Data node", nodeToWrite, optional.get());
793 void doCommit(final DOMStoreThreePhaseCommitCohort cohort) throws Exception {
794 Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
795 assertEquals("canCommit", true, canCommit);
796 cohort.preCommit().get(5, TimeUnit.SECONDS);
797 cohort.commit().get(5, TimeUnit.SECONDS);
800 void cleanup(DistributedDataStore dataStore) {
801 dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), null);
804 void assertExceptionOnCall(Callable<Void> callable, Class<? extends Exception> expType)
808 fail("Expected " + expType.getSimpleName());
809 } catch(Exception e) {
810 assertEquals("Exception type", expType, e.getClass());
814 void assertExceptionOnTxChainCreates(final DOMStoreTransactionChain txChain,
815 Class<? extends Exception> expType) throws Exception {
816 assertExceptionOnCall(new Callable<Void>() {
818 public Void call() throws Exception {
819 txChain.newWriteOnlyTransaction();
824 assertExceptionOnCall(new Callable<Void>() {
826 public Void call() throws Exception {
827 txChain.newReadWriteTransaction();
832 assertExceptionOnCall(new Callable<Void>() {
834 public Void call() throws Exception {
835 txChain.newReadOnlyTransaction();