+ @Test
+ public void testTransactionChainWithMultipleShards() throws Exception{
+ initDatastoresWithCarsAndPeople("testTransactionChainWithMultipleShards");
+
+ DOMStoreTransactionChain txChain = followerDistributedDataStore.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());
+
+ followerTestKit.doCommit(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(personPath);
+
+ DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
+
+ followerTestKit.doCommit(cohort2);
+ followerTestKit.doCommit(cohort3);
+
+ txChain.close();
+
+ DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
+ verifyCars(readTx, car);
+
+ optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
+ assertEquals("isPresent", false, optional.isPresent());
+ }
+
+ @Test
+ public void testChainedTransactionFailureWithSingleShard() throws Exception {
+ initDatastoresWithCars("testChainedTransactionFailureWithSingleShard");
+
+ ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
+ ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
+ LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(),
+ MoreExecutors.directExecutor());
+
+ TransactionChainListener listener = Mockito.mock(TransactionChainListener.class);
+ DOMTransactionChain txChain = broker.createTransactionChain(listener);
+
+ DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
+
+ ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+ new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)).
+ withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build();
+
+ 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();
+ }
+
+ @Test
+ public void testChainedTransactionFailureWithMultipleShards() throws Exception {
+ initDatastoresWithCarsAndPeople("testChainedTransactionFailureWithMultipleShards");
+
+ ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker(
+ ImmutableMap.<LogicalDatastoreType, DOMStore>builder().put(
+ LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).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();
+ }
+
+ @Test
+ public void testSingleShardTransactionsWithLeaderChanges() throws Exception {
+ String testName = "testSingleShardTransactionsWithLeaderChanges";
+ initDatastoresWithCars(testName);
+
+ String followerCarShardName = "member-2-shard-cars-" + testName;
+ InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class );
+
+ // Write top-level car container from the follower so it uses a remote Tx.
+
+ DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+
+ writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
+ writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
+
+ followerTestKit.doCommit(writeTx.ready());
+
+ InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
+
+ // Switch the leader to the follower
+
+ sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder.
+ shardElectionTimeoutFactor(1).customRaftPolicyImplementation(null));
+
+ JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
+
+ followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), CARS);
+
+ leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
+ Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
+
+ DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder().
+ shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
+ IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder);
+ newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS);
+
+ followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS);
+
+ // Write a car entry to the new leader - should switch to local Tx
+
+ writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
+
+ MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
+ YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
+ writeTx.merge(car1Path, car1);
+
+ followerTestKit.doCommit(writeTx.ready());
+
+ verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1);
+ }
+