X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxyTest.java;h=6cf63157e16ff7f37033ee3142f4ca1c06e84df8;hp=cc9692bfd91b72f3245f2bb6bd160f12551720b5;hb=2f7c93174d7834a4c4aedacc9b88aa53a5a0422c;hpb=fe45ad923c8cf83d730cf4d576c310967afabdf3 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java index cc9692bfd9..6cf63157e1 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java @@ -11,9 +11,9 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY; -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_WRITE; -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.WRITE_ONLY; +import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_ONLY; +import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE; +import static org.opendaylight.controller.cluster.datastore.TransactionType.WRITE_ONLY; import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; @@ -34,13 +34,14 @@ import org.junit.Assert; import org.junit.Test; import org.mockito.InOrder; import org.mockito.Mockito; -import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; +import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; @@ -57,6 +58,7 @@ import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCoh import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Promise; @@ -76,7 +78,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { public void testRead() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); @@ -105,7 +107,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS); } @@ -117,7 +119,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } @@ -129,14 +131,14 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { if (exToThrow instanceof PrimaryNotFoundException) { doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); } else { - doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), actorRef)). when(mockActorContext).findPrimaryShardAsync(anyString()); } doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync( any(ActorSelection.class), any()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(invoker.invoke(transactionProxy)); } @@ -177,7 +179,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, expectedNode); @@ -197,7 +199,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testReadPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.read(TestModel.TEST_PATH); } @@ -208,13 +210,13 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext). actorSelection(actorRef.path().toString()); - doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), actorRef)). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync( eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } @@ -223,7 +225,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { public void testExists() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); @@ -257,8 +259,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS); } @@ -270,7 +271,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); } @@ -286,7 +287,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -304,7 +305,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testExistsPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.exists(TestModel.TEST_PATH); } @@ -317,7 +318,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -326,7 +327,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testWriteAfterAsyncRead() throws Throwable { - ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem()); + ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem(), DefaultShardStrategy.DEFAULT_SHARD); Promise createTxPromise = akka.dispatch.Futures.promise(); doReturn(createTxPromise).when(mockActorContext).executeOperationAsync( @@ -340,7 +341,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - final TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + final TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); final CountDownLatch readComplete = new CountDownLatch(1); final AtomicReference caughtEx = new AtomicReference<>(); @@ -380,13 +381,13 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testWritePreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); } @Test(expected=IllegalStateException.class) public void testWriteAfterReadyPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.ready(); @@ -402,7 +403,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); @@ -416,7 +417,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.delete(TestModel.TEST_PATH); @@ -434,7 +435,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -460,9 +461,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - expectBatchedModificationsReady(actorRef); + expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -470,16 +471,14 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), true, + verifyBatchedModifications(batchedModifications.get(0), true, true, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent()); @@ -492,24 +491,44 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - expectBatchedModificationsReady(actorRef); + expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), true); + verifyBatchedModifications(batchedModifications.get(0), true, true); + } + + @Test + public void testReadyWithMultipleShardWrites() throws Exception { + ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); + + ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk"); + + expectBatchedModificationsReady(actorRef1); + expectBatchedModificationsReady(actorRef2); + + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); + + transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME)); + transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + + verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef1), + actorSelection(actorRef2)); } @Test @@ -520,24 +539,22 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModificationsReady(actorRef); + expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), true, + verifyBatchedModifications(batchedModifications.get(0), true, true, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), @@ -551,19 +568,17 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModificationsReady(actorRef); + expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 2, batchedModifications.size()); @@ -571,7 +586,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - verifyBatchedModifications(batchedModifications.get(1), true); + verifyBatchedModifications(batchedModifications.get(1), true, true); verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); @@ -587,23 +602,21 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectFailedBatchedModifications(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, TestException.class); + verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class); } private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception { doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -615,11 +628,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + assertTrue(ready instanceof SingleCommitCohortProxy); - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - - verifyCohortFutures(proxy, toThrow.getClass()); + verifyCohortFutures((SingleCommitCohortProxy)ready, toThrow.getClass()); } @Test @@ -640,34 +651,32 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testReadyWithInvalidReplyMessageType() throws Exception { dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - //expectBatchedModifications(actorRef, 1); + ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk"); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), - isA(BatchedModifications.class)); + executeOperationAsync(eq(actorSelection(actorRef1)), isA(BatchedModifications.class)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + expectBatchedModificationsReady(actorRef2); - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); + + transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME)); + transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - - verifyCohortFutures(proxy, IllegalArgumentException.class); + verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef2), + IllegalArgumentException.class); } @Test public void testGetIdentifier() { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - TransactionProxy.TransactionType.READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); Object id = transactionProxy.getIdentifier(); assertNotNull("getIdentifier returned null", id); @@ -681,7 +690,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -709,7 +718,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(true).when(mockActorContext).isPathLocal(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); // negative test case with null as the reply doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync( @@ -746,20 +755,18 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); doReturn(true).when(mockActorContext).isPathLocal(anyString()); - expectBatchedModificationsReady(actorRef); + expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); } private static interface TransactionProxyOperation { @@ -770,6 +777,10 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { throttleOperation(operation, 1, true); } + private PrimaryShardInfo newPrimaryShardInfo(ActorRef actorRef){ + return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), Optional.absent()); + } + private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){ ActorSystem actorSystem = getSystem(); ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); @@ -780,7 +791,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { when(mockActorContext).actorSelection(shardActorRef.path().toString()); if(shardFound) { - doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } else { doReturn(Futures.failed(new Exception("not found"))) @@ -798,7 +809,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(true).when(mockActorContext).isPathLocal(actorPath); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); long start = System.nanoTime(); @@ -826,7 +837,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { when(mockActorContext).actorSelection(shardActorRef.path().toString()); if(shardFound) { - doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } else { doReturn(Futures.failed(new PrimaryNotFoundException("test"))) @@ -847,7 +858,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(true).when(mockActorContext).isPathLocal(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); long start = System.nanoTime(); @@ -860,22 +871,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expected, (end-start)), (end - start) <= expected); } - public void testWriteThrottling(boolean shardFound){ - - throttleOperation(new TransactionProxyOperation() { - @Override - public void run(TransactionProxy transactionProxy) { - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - expectBatchedModifications(2); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - } - }, 1, shardFound); - } - @Test public void testWriteThrottlingWhenShardFound(){ dataStoreContextBuilder.shardBatchedModificationCount(1); @@ -1201,7 +1196,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH; YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH; - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, type); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, type); transactionProxy.write(writePath1, writeNode1); transactionProxy.write(writePath2, writeNode2); @@ -1224,8 +1219,8 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1), new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3)); - verifyBatchedModifications(batchedModifications.get(2), true, new MergeModification(mergePath3, mergeNode3), - new DeleteModification(deletePath2)); + verifyBatchedModifications(batchedModifications.get(2), true, true, + new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2)); assertEquals("getTotalMessageCount", 3, batchedModifications.get(2).getTotalMessagesSent()); } @@ -1279,7 +1274,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(writePath1, writeNode1); transactionProxy.write(writePath2, writeNode2); @@ -1355,7 +1350,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); Optional> readOptional = transactionProxy.read( YangInstanceIdentifier.builder().build()).get(5, TimeUnit.SECONDS); @@ -1391,7 +1386,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(getSystem().actorSelection(shardActorRef.path())). when(mockActorContext).actorSelection(shardActorRef.path().toString()); - doReturn(Futures.successful(getSystem().actorSelection(shardActorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), shardActorRef)). when(mockActorContext).findPrimaryShardAsync(eq(shardName)); doReturn(true).when(mockActorContext).isPathLocal(shardActorRef.path().toString());