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=cc9692bfd91b72f3245f2bb6bd160f12551720b5;hb=ebaf3d71465066033d5882c61cdd2ec63b29d980;hp=29a5b09c5c8e33fcb8892054564d4264dd2a711d;hpb=4e594bb578f687fb24e242e90b918d91fb59c3ac;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 29a5b09c5c..cc9692bfd9 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 @@ -3,7 +3,6 @@ package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; @@ -41,7 +40,6 @@ import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedEx 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.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; @@ -61,10 +59,7 @@ import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.model.api.SchemaContext; -import scala.concurrent.Await; -import scala.concurrent.Future; import scala.concurrent.Promise; -import scala.concurrent.duration.Duration; @SuppressWarnings("resource") public class TransactionProxyTest extends AbstractTransactionProxyTest { @@ -313,29 +308,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { transactionProxy.exists(TestModel.TEST_PATH); } - private void verifyRecordingOperationFutures(List> futures, - Class... expResultTypes) throws Exception { - assertEquals("getRecordingOperationFutures size", expResultTypes.length, futures.size()); - - int i = 0; - for( Future future: futures) { - assertNotNull("Recording operation Future is null", future); - - Class expResultType = expResultTypes[i++]; - if(Throwable.class.isAssignableFrom(expResultType)) { - try { - Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - fail("Expected exception from recording operation Future"); - } catch(Exception e) { - // Expected - } - } else { - assertEquals(String.format("Recording operation %d Future result type", i +1 ), expResultType, - Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass()); - } - } - } - @Test public void testWrite() throws Exception { dataStoreContextBuilder.shardBatchedModificationCount(1); @@ -364,8 +336,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); @@ -404,10 +375,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { // This sends the batched modification. transactionProxy.ready(); - verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), true); } @Test(expected=IllegalStateException.class) @@ -456,7 +424,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); @@ -465,7 +433,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); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -479,31 +474,29 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); - verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), - isA(BatchedModifications.class)); + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), - isA(ReadyTransaction.SERIALIZABLE_CLASS)); + verifyBatchedModifications(batchedModifications.get(0), true, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); + + assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent()); } @Test - public void testReadyWithWriteOnlyAndLastBatchPending() throws Exception { - dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); - - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); + public void testReadyWithNoModifications() 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, 1); + expectBatchedModificationsReady(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + transactionProxy.read(TestModel.TEST_PATH); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); @@ -511,28 +504,23 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures()); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), true, - new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - - verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), - isA(ReadyTransaction.SERIALIZABLE_CLASS)); + verifyBatchedModifications(batchedModifications.get(0), true); } @Test - public void testReadyWithWriteOnlyAndLastBatchEmpty() throws Exception { - dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true); + public void testReadyWithWriteOnlyAndLastBatchPending() throws Exception { + dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModificationsReady(actorRef, 1); + expectBatchedModificationsReady(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -544,34 +532,26 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); List batchedModifications = captureBatchedModifications(actorRef); - assertEquals("Captured BatchedModifications count", 2, batchedModifications.size()); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), false, + verifyBatchedModifications(batchedModifications.get(0), true, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - verifyBatchedModifications(batchedModifications.get(1), true); - verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); } @Test - public void testReadyWithRecordingOperationFailure() throws Exception { + public void testReadyWithWriteOnlyAndLastBatchEmpty() throws Exception { dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true); - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectFailedBatchedModifications(actorRef); - - doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); + expectBatchedModificationsReady(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -583,9 +563,18 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortFutures(proxy, TestException.class); + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 2, batchedModifications.size()); - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), TestException.class); + verifyBatchedModifications(batchedModifications.get(0), false, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); + + verifyBatchedModifications(batchedModifications.get(1), true); + + verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), + isA(ReadyTransaction.SERIALIZABLE_CLASS)); } @Test @@ -757,18 +746,13 @@ 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); 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); @@ -1196,8 +1180,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, shardBatchedModificationCount); - expectReadyTransaction(actorRef); - YangInstanceIdentifier writePath1 = TestModel.TEST_PATH; NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -1242,17 +1224,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), + verifyBatchedModifications(batchedModifications.get(2), true, new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2)); - if(optimizedWriteOnly) { - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class); - } else { - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); - } + assertEquals("getTotalMessageCount", 3, batchedModifications.get(2).getTotalMessagesSent()); } @Test @@ -1357,9 +1332,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { inOrder.verify(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); } @Test