- @Test
- public void testTransactionChainWithMultipleShards() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore("testTransactionChainWithMultipleShards",
- "cars-1", "people-1");
-
- DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
-
- DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
- assertNotNull("newWriteOnlyTransaction returned null", writeTx);
-
- writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
- writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
-
- writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
- writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
-
- DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
-
- DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
-
- MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
- YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
- readWriteTx.write(carPath, car);
-
- MapEntryNode person = PeopleModel.newPersonEntry("jack");
- YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
- readWriteTx.merge(personPath, person);
-
- Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("Data node", car, optional.get());
-
- optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("Data node", person, optional.get());
-
- DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
-
- writeTx = txChain.newWriteOnlyTransaction();
-
- writeTx.delete(carPath);
-
- DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
-
- ListenableFuture<Boolean> canCommit1 = cohort1.canCommit();
- ListenableFuture<Boolean> canCommit2 = cohort2.canCommit();
-
- doCommit(canCommit1, cohort1);
- doCommit(canCommit2, cohort2);
- doCommit(cohort3);
-
- txChain.close();
-
- DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
-
- optional = readTx.read(carPath).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", false, optional.isPresent());
-
- optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("Data node", person, optional.get());
-
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testCreateChainedTransactionsInQuickSuccession() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testCreateChainedTransactionsInQuickSuccession", "cars-1");
-
- ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
- ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
- LogicalDatastoreType.CONFIGURATION, dataStore).build(), MoreExecutors.directExecutor());
-
- TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
- DOMTransactionChain txChain = broker.createTransactionChain(listener);
-
- List<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
-
- DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
- writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, CarsModel.emptyContainer());
- writeTx.put(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
- futures.add(writeTx.submit());
-
- int nCars = 100;
- for(int i = 0; i < nCars; i++) {
- DOMDataReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
-
- rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i),
- CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000)));
-
- futures.add(rwTx.submit());
- }
-
- for(CheckedFuture<Void, TransactionCommitFailedException> f: futures) {
- f.checkedGet();
- }
-
- Optional<NormalizedNode<?, ?>> optional = txChain.newReadOnlyTransaction().read(
- LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("# cars", nCars, ((Collection<?>)optional.get().getValue()).size());
-
- txChain.close();
-
- broker.close();
-
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testCreateChainedTransactionAfterEmptyTxReadied() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testCreateChainedTransactionAfterEmptyTxReadied", "test-1");
-
- DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
-
- DOMStoreReadWriteTransaction rwTx1 = txChain.newReadWriteTransaction();
-
- rwTx1.ready();
-
- DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction();
-
- Optional<NormalizedNode<?, ?>> optional = rwTx2.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", false, optional.isPresent());
-
- txChain.close();
-
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testCreateChainedTransactionWhenPreviousNotReady() throws Throwable {
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testCreateChainedTransactionWhenPreviousNotReady", "test-1");
-
- final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
-
- DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
- assertNotNull("newWriteOnlyTransaction returned null", writeTx);
-
- writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
-
- // Try to create another Tx of each type - each should fail b/c the previous Tx wasn't
- // readied.
-
- assertExceptionOnTxChainCreates(txChain, IllegalStateException.class);
- }};
- }
-
- @Test
- public void testCreateChainedTransactionAfterClose() throws Throwable {
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testCreateChainedTransactionAfterClose", "test-1");
-
- DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
-
- txChain.close();
-
- // Try to create another Tx of each type - should fail b/c the previous Tx was closed.
-
- assertExceptionOnTxChainCreates(txChain, TransactionChainClosedException.class);
- }};
- }
-
- @Test
- public void testChainWithReadOnlyTxAfterPreviousReady() throws Throwable {
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testChainWithReadOnlyTxAfterPreviousReady", "test-1");
-
- final DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
-
- // Create a write tx and submit.
-
- DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
- writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
- DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready();
-
- // Create read-only tx's and issue a read.
-
- CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture1 =
- txChain.newReadOnlyTransaction().read(TestModel.TEST_PATH);
-
- CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readFuture2 =
- txChain.newReadOnlyTransaction().read(TestModel.TEST_PATH);
-
- // Create another write tx and issue the write.
-
- DOMStoreWriteTransaction writeTx2 = txChain.newWriteOnlyTransaction();
- writeTx2.write(TestModel.OUTER_LIST_PATH,
- ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
-
- // Ensure the reads succeed.
-
- assertEquals("isPresent", true, readFuture1.checkedGet(5, TimeUnit.SECONDS).isPresent());
- assertEquals("isPresent", true, readFuture2.checkedGet(5, TimeUnit.SECONDS).isPresent());
-
- // Ensure the writes succeed.
-
- DOMStoreThreePhaseCommitCohort cohort2 = writeTx2.ready();
-
- doCommit(cohort1);
- doCommit(cohort2);
-
- assertEquals("isPresent", true, txChain.newReadOnlyTransaction().read(TestModel.OUTER_LIST_PATH).
- checkedGet(5, TimeUnit.SECONDS).isPresent());
- }};
- }
-
- @Test
- public void testChainedTransactionFailureWithSingleShard() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testChainedTransactionFailureWithSingleShard", "cars-1");
-
- ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
- ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
- LogicalDatastoreType.CONFIGURATION, dataStore).build(), MoreExecutors.directExecutor());
-
- TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
- DOMTransactionChain txChain = broker.createTransactionChain(listener);
-
- DOMDataReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
-
- ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
- new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
- withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
-
- rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
-
- try {
- rwTx.submit().checkedGet(5, TimeUnit.SECONDS);
- fail("Expected TransactionCommitFailedException");
- } catch (TransactionCommitFailedException e) {
- // Expected
- }
-
- verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(rwTx), any(Throwable.class));
-
- txChain.close();
- broker.close();
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testChainedTransactionFailureWithMultipleShards() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore = setupDistributedDataStore(
- "testChainedTransactionFailureWithMultipleShards", "cars-1", "people-1");
-
- ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
- ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
- LogicalDatastoreType.CONFIGURATION, dataStore).build(), MoreExecutors.directExecutor());
-
- TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
- DOMTransactionChain txChain = broker.createTransactionChain(listener);
-
- DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
-
- writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
-
- ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
- new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
- withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
-
- // Note that merge will validate the data and fail but put succeeds b/c deep validation is not
- // done for put for performance reasons.
- writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData);
-
- try {
- writeTx.submit().checkedGet(5, TimeUnit.SECONDS);
- fail("Expected TransactionCommitFailedException");
- } catch (TransactionCommitFailedException e) {
- // Expected
- }
-
- verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx), any(Throwable.class));
-
- txChain.close();
- broker.close();
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testChangeListenerRegistration() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- DistributedDataStore dataStore =
- setupDistributedDataStore("testChangeListenerRegistration", "test-1");
-
- testWriteTransaction(dataStore, TestModel.TEST_PATH,
- ImmutableNodes.containerNode(TestModel.TEST_QNAME));
-
- MockDataChangeListener listener = new MockDataChangeListener(1);
-
- ListenerRegistration<MockDataChangeListener>
- listenerReg = dataStore.registerChangeListener(TestModel.TEST_PATH, listener,
- DataChangeScope.SUBTREE);
-
- assertNotNull("registerChangeListener returned null", listenerReg);
-
- // Wait for the initial notification
-
- listener.waitForChangeEvents(TestModel.TEST_PATH);
-
- listener.reset(2);
-
- // Write 2 updates.
-
- testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
- ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
-
- 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));
-
- // Wait for the 2 updates.
-
- listener.waitForChangeEvents(TestModel.OUTER_LIST_PATH, listPath);
-
- 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");
-
- cleanup(dataStore);
- }};
- }
-
- @Test
- public void testRestoreFromDatastoreSnapshot() throws Exception{
- new IntegrationTestKit(getSystem(), datastoreContextBuilder) {{
- String name = "transactionIntegrationTest";
-
- ContainerNode carsNode = CarsModel.newCarsNode(CarsModel.newCarsMapNode(
- CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)),
- CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L))));
-
- ShardDataTree dataTree = new ShardDataTree(SchemaContextHelper.full());
- AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode);
- NormalizedNode<?, ?> root = AbstractShardTest.readStore(dataTree.getDataTree(),
- YangInstanceIdentifier.builder().build());
-
- Snapshot carsSnapshot = Snapshot.create(SerializationUtils.serializeNormalizedNode(root),
- Collections.<ReplicatedLogEntry>emptyList(), 2, 1, 2, 1, 1, "member-1");
-
- NormalizedNode<?, ?> peopleNode = PeopleModel.create();
- dataTree = new ShardDataTree(SchemaContextHelper.full());
- AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode);
- root = AbstractShardTest.readStore(dataTree.getDataTree(), YangInstanceIdentifier.builder().build());
-
- Snapshot peopleSnapshot = Snapshot.create(SerializationUtils.serializeNormalizedNode(root),
- Collections.<ReplicatedLogEntry>emptyList(), 2, 1, 2, 1, 1, "member-1");
-
- restoreFromSnapshot = new DatastoreSnapshot(name, null, Arrays.asList(
- new DatastoreSnapshot.ShardSnapshot("cars",
- org.apache.commons.lang3.SerializationUtils.serialize(carsSnapshot)),
- new DatastoreSnapshot.ShardSnapshot("people",
- org.apache.commons.lang3.SerializationUtils.serialize(peopleSnapshot))));
-
- DistributedDataStore dataStore = setupDistributedDataStore(name, "module-shards-member1.conf",
- true, "cars", "people");
-
- DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
-
- Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("Data node", carsNode, optional.get());
-
- optional = readTx.read(PeopleModel.BASE_PATH).get(5, TimeUnit.SECONDS);
- assertEquals("isPresent", true, optional.isPresent());
- assertEquals("Data node", peopleNode, optional.get());
-
- cleanup(dataStore);
- }};
- }