X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxyTest.java;h=844feb2f47e988b89498a80946d97aecbff8c049;hb=e8746b7ae6620d9e0dc159f2a13d3385d6197c56;hp=b95eaf64d75d2ab83bd5cc2ae1c94f7e2f50e2c4;hpb=fd7d9b244744df4c2c1460c8c24431532a9a45b3;p=controller.git 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 b95eaf64d7..844feb2f47 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 @@ -41,6 +41,7 @@ import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundE 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.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; @@ -129,7 +130,7 @@ 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()); } @@ -208,7 +209,7 @@ 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( @@ -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( @@ -336,8 +337,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - expectBatchedModifications(actorRef, 1); - expectReadyTransaction(actorRef); + expectBatchedModificationsReady(actorRef); final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -376,7 +376,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { // This sends the batched modification. transactionProxy.ready(); - verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false); + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), true); } @Test(expected=IllegalStateException.class) @@ -425,7 +425,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { } @Test - public void testReadyWithReadWrite() throws Exception { + public void testReadWrite() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -434,7 +434,34 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { eq(actorSelection(actorRef)), eqSerializedReadData()); expectBatchedModifications(actorRef, 1); - expectReadyTransaction(actorRef); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); + + verifyBatchedModifications(batchedModifications.get(0), false, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); + } + + @Test + public void testReadyWithReadWrite() throws Exception { + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); + + expectBatchedModificationsReady(actorRef, true); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -444,17 +471,64 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + assertTrue(ready instanceof SingleCommitCohortProxy); + + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + verifyBatchedModifications(batchedModifications.get(0), true, true, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent()); + } - verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), - isA(BatchedModifications.class)); + @Test + public void testReadyWithNoModifications() throws Exception { + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); - verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), - isA(ReadyTransaction.SERIALIZABLE_CLASS)); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); + + expectBatchedModificationsReady(actorRef, true); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + transactionProxy.read(TestModel.TEST_PATH); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof SingleCommitCohortProxy); + + verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable()); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); + + 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(mockActorContext, 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 @@ -465,7 +539,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModificationsReady(actorRef, 1); + expectBatchedModificationsReady(actorRef, true); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -473,16 +547,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)); verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), @@ -496,7 +568,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModificationsReady(actorRef, 1); + expectBatchedModificationsReady(actorRef, true); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -504,11 +576,9 @@ 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", 2, batchedModifications.size()); @@ -516,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)); @@ -538,11 +608,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + assertTrue(ready instanceof SingleCommitCohortProxy); - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - - verifyCohortFutures(proxy, TestException.class); + verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class); } private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception { @@ -560,11 +628,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + assertTrue(ready instanceof SingleCommitCohortProxy); - verifyCohortFutures(proxy, toThrow.getClass()); + verifyCohortFutures((SingleCommitCohortProxy)ready, toThrow.getClass()); } @Test @@ -585,27 +651,26 @@ 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)); + + expectBatchedModificationsReady(actorRef2); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + 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 @@ -691,25 +756,18 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); doReturn(true).when(mockActorContext).isPathLocal(anyString()); - doReturn(batchedModificationsReply(1)).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(BatchedModifications.class)); + expectBatchedModificationsReady(actorRef, true); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - // testing ready - doReturn(readyTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.class)); - 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 { @@ -776,7 +834,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { when(mockActorContext).actorSelection(shardActorRef.path().toString()); if(shardFound) { - doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + doReturn(primaryShardInfoReply(actorSystem, shardActorRef)). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } else { doReturn(Futures.failed(new PrimaryNotFoundException("test"))) @@ -1130,8 +1188,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, shardBatchedModificationCount); - expectReadyTransaction(actorRef); - YangInstanceIdentifier writePath1 = TestModel.TEST_PATH; NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -1176,9 +1232,10 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1), new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3)); - boolean optimizedWriteOnly = type == WRITE_ONLY && dataStoreContextBuilder.build().isWriteOnlyTransactionOptimizationsEnabled(); - verifyBatchedModifications(batchedModifications.get(2), optimizedWriteOnly, 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()); } @Test @@ -1342,7 +1399,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());