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=6573308c12100914badbedc5d0296b90e096a2b7;hb=64db87e8fb5de25e68ba824ec2eaa17e3f217c96;hp=6d80bbb5b158eb2d962a87c0bb6bf08654be2bf8;hpb=e0e940ce65e9ae8845b94da03fa559c5980a2876;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 6d80bbb5b1..6573308c12 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 @@ -25,23 +25,33 @@ import akka.testkit.JavaTestKit; import com.google.common.base.Optional; import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.CheckedFuture; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Uninterruptibles; import com.typesafe.config.Config; import com.typesafe.config.ConfigFactory; import java.io.IOException; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.junit.AfterClass; import org.junit.Assert; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.ArgumentMatcher; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; +import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; 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.CreateTransaction; import org.opendaylight.controller.cluster.datastore.messages.DataExists; @@ -56,6 +66,11 @@ import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.WriteData; import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply; +import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; +import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; +import org.opendaylight.controller.cluster.datastore.modification.MergeModification; +import org.opendaylight.controller.cluster.datastore.modification.Modification; +import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy; import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; @@ -67,11 +82,13 @@ import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages; import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; 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") @@ -97,7 +114,10 @@ public class TransactionProxyTest { @Mock private ClusterWrapper mockClusterWrapper; - String memberName = "mock-member"; + private final String memberName = "mock-member"; + + private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().operationTimeoutInSeconds(2). + shardBatchedModificationCount(1); @BeforeClass public static void setUpClass() throws IOException { @@ -121,14 +141,13 @@ public class TransactionProxyTest { schemaContext = TestModel.createTestContext(); - DatastoreContext dataStoreContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(2).build(); - doReturn(getSystem()).when(mockActorContext).getActorSystem(); + doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); doReturn(memberName).when(mockActorContext).getCurrentMemberName(); doReturn(schemaContext).when(mockActorContext).getSchemaContext(); doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); doReturn(mockClusterWrapper).when(mockActorContext).getClusterWrapper(); - doReturn(dataStoreContext).when(mockActorContext).getDatastoreContext(); + doReturn(dataStoreContextBuilder.build()).when(mockActorContext).getDatastoreContext(); doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); ShardStrategyFactory.setConfiguration(configuration); @@ -181,11 +200,15 @@ public class TransactionProxyTest { } private ReadData eqSerializedReadData() { + return eqSerializedReadData(TestModel.TEST_PATH); + } + + private ReadData eqSerializedReadData(final YangInstanceIdentifier path) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) && - ReadData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH); + ReadData.fromSerializable(argument).getPath().equals(path); } }; @@ -204,23 +227,13 @@ public class TransactionProxyTest { return argThat(matcher); } - private WriteData eqSerializedWriteData(final NormalizedNode nodeToWrite) { - return eqSerializedWriteData(nodeToWrite, DataStoreVersions.CURRENT_VERSION); - } - - private WriteData eqSerializedWriteData(final NormalizedNode nodeToWrite, - final int transactionVersion) { + private WriteData eqLegacyWriteData(final NormalizedNode nodeToWrite) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - if((transactionVersion >= DataStoreVersions.LITHIUM_VERSION && - WriteData.SERIALIZABLE_CLASS.equals(argument.getClass())) || - (transactionVersion < DataStoreVersions.LITHIUM_VERSION && - ShardTransactionMessages.WriteData.class.equals(argument.getClass()))) { - + if(ShardTransactionMessages.WriteData.class.equals(argument.getClass())) { WriteData obj = WriteData.fromSerializable(argument); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); + return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite); } return false; @@ -230,39 +243,13 @@ public class TransactionProxyTest { return argThat(matcher); } - private WriteData eqWriteData(final NormalizedNode nodeToWrite) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(argument instanceof WriteData) { - WriteData obj = (WriteData) argument; - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); - } - return false; - } - }; - - return argThat(matcher); - } - - private MergeData eqSerializedMergeData(final NormalizedNode nodeToWrite) { - return eqSerializedMergeData(nodeToWrite, DataStoreVersions.CURRENT_VERSION); - } - - private MergeData eqSerializedMergeData(final NormalizedNode nodeToWrite, - final int transactionVersion) { + private MergeData eqLegacyMergeData(final NormalizedNode nodeToWrite) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - if((transactionVersion >= DataStoreVersions.LITHIUM_VERSION && - MergeData.SERIALIZABLE_CLASS.equals(argument.getClass())) || - (transactionVersion < DataStoreVersions.LITHIUM_VERSION && - ShardTransactionMessages.MergeData.class.equals(argument.getClass()))) { - + if(ShardTransactionMessages.MergeData.class.equals(argument.getClass())) { MergeData obj = MergeData.fromSerializable(argument); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); + return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite); } return false; @@ -272,41 +259,12 @@ public class TransactionProxyTest { return argThat(matcher); } - private MergeData eqMergeData(final NormalizedNode nodeToWrite) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(argument instanceof MergeData) { - MergeData obj = ((MergeData) argument); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); - } - - return false; - } - }; - - return argThat(matcher); - } - - private DeleteData eqSerializedDeleteData() { + private DeleteData eqLegacyDeleteData(final YangInstanceIdentifier expPath) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - return DeleteData.SERIALIZABLE_CLASS.equals(argument.getClass()) && - DeleteData.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private DeleteData eqDeleteData() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return argument instanceof DeleteData && - ((DeleteData)argument).getPath().equals(TestModel.TEST_PATH); + return ShardTransactionMessages.DeleteData.class.equals(argument.getClass()) && + DeleteData.fromSerializable(argument).getPath().equals(expPath); } }; @@ -323,7 +281,7 @@ public class TransactionProxyTest { private Future readSerializedDataReply(NormalizedNode data, short transactionVersion) { - return Futures.successful(new ReadDataReply(data).toSerializable(transactionVersion)); + return Futures.successful(new ReadDataReply(data, transactionVersion).toSerializable()); } private Future readSerializedDataReply(NormalizedNode data) { @@ -331,7 +289,7 @@ public class TransactionProxyTest { } private Future readDataReply(NormalizedNode data) { - return Futures.successful(new ReadDataReply(data)); + return Futures.successful(new ReadDataReply(data, DataStoreVersions.CURRENT_VERSION)); } private Future dataExistsSerializedReply(boolean exists) { @@ -342,48 +300,41 @@ public class TransactionProxyTest { return Futures.successful(new DataExistsReply(exists)); } - private Future writeSerializedDataReply(short version) { - return Futures.successful(new WriteDataReply().toSerializable(version)); - } - - private Future writeSerializedDataReply() { - return writeSerializedDataReply(DataStoreVersions.CURRENT_VERSION); - } - - private Future writeDataReply() { - return Futures.successful(new WriteDataReply()); - } - - private Future mergeSerializedDataReply(short version) { - return Futures.successful(new MergeDataReply().toSerializable(version)); - } - - private Future mergeSerializedDataReply() { - return mergeSerializedDataReply(DataStoreVersions.CURRENT_VERSION); + private Future batchedModificationsReply(int count) { + return Futures.successful(new BatchedModificationsReply(count)); } private Future incompleteFuture(){ return mock(Future.class); } - private Future mergeDataReply() { - return Futures.successful(new MergeDataReply()); + private ActorSelection actorSelection(ActorRef actorRef) { + return getSystem().actorSelection(actorRef.path()); } - private Future deleteSerializedDataReply(short version) { - return Futures.successful(new DeleteDataReply().toSerializable(version)); + private void expectBatchedModifications(ActorRef actorRef, int count) { + doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); } - private Future deleteSerializedDataReply() { - return deleteSerializedDataReply(DataStoreVersions.CURRENT_VERSION); + private void expectBatchedModifications(int count) { + doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), isA(BatchedModifications.class)); } - private Future deleteDataReply() { - return Futures.successful(new DeleteDataReply()); + private void expectIncompleteBatchedModifications() { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), isA(BatchedModifications.class)); } - private ActorSelection actorSelection(ActorRef actorRef) { - return getSystem().actorSelection(actorRef.path()); + private void expectReadyTransaction(ActorRef actorRef) { + doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + } + + private void expectFailedBatchedModifications(ActorRef actorRef) { + doReturn(Futures.failed(new TestException())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); } private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ @@ -394,8 +345,7 @@ public class TransactionProxyTest { .build(); } - private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, - TransactionType type, int transactionVersion) { + private ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem) { ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); doReturn(actorSystem.actorSelection(actorRef.path())). when(mockActorContext).actorSelection(actorRef.path().toString()); @@ -403,10 +353,6 @@ public class TransactionProxyTest { doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); - doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext). - executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())), - eqCreateTransaction(memberName, type)); - doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); @@ -414,6 +360,17 @@ public class TransactionProxyTest { return actorRef; } + private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, + TransactionType type, int transactionVersion) { + ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(actorSystem); + + doReturn(Futures.successful(createTransactionReply(actorRef, transactionVersion))).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(actorRef.path())), + eqCreateTransaction(memberName, type)); + + return actorRef; + } + private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION); } @@ -434,8 +391,7 @@ public class TransactionProxyTest { public void testRead() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); @@ -464,8 +420,7 @@ public class TransactionProxyTest { doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS); } @@ -477,8 +432,7 @@ public class TransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } @@ -529,21 +483,19 @@ public class TransactionProxyTest { @Test(expected = TestException.class) public void testReadWithPriorRecordingOperationFailure() throws Throwable { + doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()). + when(mockActorContext).getDatastoreContext(); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - - doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData()); + expectFailedBatchedModifications(actorRef); doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -563,14 +515,12 @@ public class TransactionProxyTest { NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(expectedNode)); + expectBatchedModifications(actorRef, 1); doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, expectedNode); @@ -578,16 +528,19 @@ public class TransactionProxyTest { TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); - assertEquals("Response NormalizedNode", expectedNode, readOptional.get()); + + InOrder inOrder = Mockito.inOrder(mockActorContext); + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); } @Test(expected=IllegalStateException.class) public void testReadPreConditionCheck() { - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); - + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.read(TestModel.TEST_PATH); } @@ -613,8 +566,7 @@ public class TransactionProxyTest { public void testExists() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); @@ -661,23 +613,21 @@ public class TransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); } @Test(expected = TestException.class) public void testExistsWithPriorRecordingOperationFailure() throws Throwable { + doReturn(dataStoreContextBuilder.shardBatchedModificationCount(2).build()). + when(mockActorContext).getDatastoreContext(); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - - doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData()); + expectFailedBatchedModifications(actorRef); doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); @@ -703,28 +653,30 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet(); assertEquals("Exists response", true, exists); + + InOrder inOrder = Mockito.inOrder(mockActorContext); + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); } @Test(expected=IllegalStateException.class) public void testExistsPreConditionCheck() { - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); - + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.exists(TestModel.TEST_PATH); } @@ -745,7 +697,7 @@ public class TransactionProxyTest { // Expected } } else { - assertEquals("Recording operation Future result type", expResultType, + assertEquals(String.format("Recording operation %d Future result type", i +1 ), expResultType, Await.result(future, Duration.create(5, TimeUnit.SECONDS)).getClass()); } } @@ -757,41 +709,93 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - verify(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + // This sends the batched modification. + transactionProxy.ready(); + + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.class); + BatchedModificationsReply.class); } - @Test(expected=IllegalStateException.class) - public void testWritePreConditionCheck() { + @Test + public void testWriteAfterAsyncRead() throws Throwable { + ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + Promise createTxPromise = akka.dispatch.Futures.promise(); + doReturn(createTxPromise).when(mockActorContext).executeOperationAsync( + eq(getSystem().actorSelection(actorRef.path())), + eqCreateTransaction(memberName, READ_WRITE)); - transactionProxy.write(TestModel.TEST_PATH, - ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); + + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); + + final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + final TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + final CountDownLatch readComplete = new CountDownLatch(1); + final AtomicReference caughtEx = new AtomicReference<>(); + com.google.common.util.concurrent.Futures.addCallback(transactionProxy.read(TestModel.TEST_PATH), + new FutureCallback>>() { + @Override + public void onSuccess(Optional> result) { + try { + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + } catch (Exception e) { + caughtEx.set(e); + } finally { + readComplete.countDown(); + } + } + + @Override + public void onFailure(Throwable t) { + caughtEx.set(t); + readComplete.countDown(); + } + }); + + createTxPromise.success(createTransactionReply(actorRef, DataStoreVersions.CURRENT_VERSION)); + + Uninterruptibles.awaitUninterruptibly(readComplete, 5, TimeUnit.SECONDS); + + if(caughtEx.get() != null) { + throw caughtEx.get(); + } + + // This sends the batched modification. + transactionProxy.ready(); + + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class); } @Test(expected=IllegalStateException.class) - public void testWriteAfterReadyPreConditionCheck() { + public void testWritePreConditionCheck() { + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + } - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + @Test(expected=IllegalStateException.class) + public void testWriteAfterReadyPreConditionCheck() { + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.ready(); - transactionProxy.write(TestModel.TEST_PATH, - ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); } @Test @@ -800,37 +804,40 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); - verify(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); + // This sends the batched modification. + transactionProxy.ready(); + + verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite)); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.class); + BatchedModificationsReply.class); } @Test public void testDelete() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - doReturn(deleteSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedDeleteData()); + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.delete(TestModel.TEST_PATH); - verify(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedDeleteData()); + // This sends the batched modification. + transactionProxy.ready(); + + verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH)); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - DeleteDataReply.class); + BatchedModificationsReply.class); } private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy, @@ -867,14 +874,10 @@ public class TransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -887,9 +890,12 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.class); + BatchedModificationsReply.class); verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + + verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), + isA(BatchedModifications.class)); } private ActorRef testCompatibilityWithHeliumVersion(short version) throws Exception { @@ -901,14 +907,16 @@ public class TransactionProxyTest { doReturn(readSerializedDataReply(testNode, version)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - doReturn(writeSerializedDataReply(version)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(testNode, version)); + doReturn(Futures.successful(new WriteDataReply().toSerializable(version))).when(mockActorContext). + executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyWriteData(testNode)); - doReturn(mergeSerializedDataReply(version)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(testNode, version)); + doReturn(Futures.successful(new MergeDataReply().toSerializable(version))).when(mockActorContext). + executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyMergeData(testNode)); - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + doReturn(Futures.successful(new DeleteDataReply().toSerializable(version))).when(mockActorContext). + executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyDeleteData(TestModel.TEST_PATH)); + + expectReadyTransaction(actorRef); doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()), eq(actorRef.path().toString())); @@ -925,6 +933,8 @@ public class TransactionProxyTest { transactionProxy.merge(TestModel.TEST_PATH, testNode); + transactionProxy.delete(TestModel.TEST_PATH); + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); assertTrue(ready instanceof ThreePhaseCommitCohortProxy); @@ -932,7 +942,8 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - ShardTransactionMessages.WriteDataReply.class, ShardTransactionMessages.MergeDataReply.class); + ShardTransactionMessages.WriteDataReply.class, ShardTransactionMessages.MergeDataReply.class, + ShardTransactionMessages.DeleteDataReply.class); verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); @@ -961,21 +972,13 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); + expectFailedBatchedModifications(actorRef); - doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + expectReadyTransaction(actorRef); doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); - - transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -987,8 +990,7 @@ public class TransactionProxyTest { verifyCohortFutures(proxy, TestException.class); - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.class, TestException.class); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), TestException.class); } @Test @@ -997,15 +999,13 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); @@ -1016,7 +1016,7 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.class); + BatchedModificationsReply.class); verifyCohortFutures(proxy, TestException.class); } @@ -1027,8 +1027,7 @@ public class TransactionProxyTest { doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when( mockActorContext).findPrimaryShardAsync(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -1053,15 +1052,13 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + expectBatchedModifications(actorRef, 1); doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -1092,8 +1089,7 @@ public class TransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -1129,9 +1125,7 @@ public class TransactionProxyTest { String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() - .setTransactionId("txn-1") - .setTransactionActorPath(actorPath) - .build(); + .setTransactionId("txn-1").setTransactionActorPath(actorPath).build(); doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), @@ -1172,7 +1166,7 @@ public class TransactionProxyTest { } @Test - public void testLocalTxActorWrite() throws Exception { + public void testLocalTxActorReady() throws Exception { ActorSystem actorSystem = getSystem(); ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); @@ -1183,48 +1177,26 @@ public class TransactionProxyTest { when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; - CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() - .setTransactionId("txn-1") - .setTransactionActorPath(actorPath) - .build(); + CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder(). + setTransactionId("txn-1").setTransactionActorPath(actorPath). + setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). - executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, WRITE_ONLY)); doReturn(true).when(mockActorContext).isPathLocal(actorPath); - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - doReturn(writeDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + doReturn(batchedModificationsReply(1)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), isA(BatchedModifications.class)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - verify(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); - - //testing local merge - doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToWrite)); - - transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); - - verify(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToWrite)); - - //testing local delete - doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqDeleteData()); - - transactionProxy.delete(TestModel.TEST_PATH); - - verify(mockActorContext).executeOperationAsync(any(ActorSelection.class), eqDeleteData()); + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.class, MergeDataReply.class, DeleteDataReply.class); + BatchedModificationsReply.class); // testing ready doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync( @@ -1265,10 +1237,9 @@ public class TransactionProxyTest { } String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; - CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() - .setTransactionId("txn-1") - .setTransactionActorPath(actorPath) - .build(); + CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder(). + setTransactionId("txn-1").setTransactionActorPath(actorPath). + setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), @@ -1278,15 +1249,15 @@ public class TransactionProxyTest { TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); - long start = System.currentTimeMillis(); + long start = System.nanoTime(); operation.run(transactionProxy); - long end = System.currentTimeMillis(); + long end = System.nanoTime(); - Assert.assertTrue(String.format("took less time than expected %s was %s", - mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()*1000, - (end-start)), (end - start) > mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()*1000); + long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()); + Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", + expected, (end-start)), (end - start) > expected); } @@ -1312,10 +1283,9 @@ public class TransactionProxyTest { } String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; - CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder() - .setTransactionId("txn-1") - .setTransactionActorPath(actorPath) - .build(); + CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder(). + setTransactionId("txn-1").setTransactionActorPath(actorPath). + setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), @@ -1325,15 +1295,15 @@ public class TransactionProxyTest { TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); - long start = System.currentTimeMillis(); + long start = System.nanoTime(); operation.run(transactionProxy); - long end = System.currentTimeMillis(); + long end = System.nanoTime(); - Assert.assertTrue(String.format("took more time than expected %s was %s", - mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()*1000, - (end-start)), (end - start) <= mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()*1000); + long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()); + Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", + expected, (end-start)), (end - start) <= expected); } public void testWriteThrottling(boolean shardFound){ @@ -1343,8 +1313,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + expectBatchedModifications(2); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -1360,15 +1329,13 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + expectIncompleteBatchedModifications(); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); } }); - } @Test @@ -1379,8 +1346,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + expectBatchedModifications(2); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -1398,15 +1364,13 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqSerializedWriteData(nodeToWrite)); + expectBatchedModifications(2); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); } }); - } @Test @@ -1417,8 +1381,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToMerge)); + expectIncompleteBatchedModifications(); transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); @@ -1435,8 +1398,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToMerge)); + expectBatchedModifications(2); transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); @@ -1452,8 +1414,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToMerge)); + expectBatchedModifications(2); transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); @@ -1469,8 +1430,7 @@ public class TransactionProxyTest { throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqDeleteData()); + expectIncompleteBatchedModifications(); transactionProxy.delete(TestModel.TEST_PATH); @@ -1486,8 +1446,7 @@ public class TransactionProxyTest { completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqDeleteData()); + expectBatchedModifications(2); transactionProxy.delete(TestModel.TEST_PATH); @@ -1501,8 +1460,7 @@ public class TransactionProxyTest { completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { - doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqDeleteData()); + expectBatchedModifications(2); transactionProxy.delete(TestModel.TEST_PATH); @@ -1620,8 +1578,7 @@ public class TransactionProxyTest { public void run(TransactionProxy transactionProxy) { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + expectBatchedModifications(1); doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( any(ActorSelection.class), any(ReadyTransaction.class)); @@ -1642,11 +1599,7 @@ public class TransactionProxyTest { NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); NormalizedNode carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); - - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(carsNode)); + expectBatchedModifications(2); doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( any(ActorSelection.class), any(ReadyTransaction.class)); @@ -1659,4 +1612,203 @@ public class TransactionProxyTest { } }, 2, true); } + + @Test + public void testModificationOperationBatching() throws Throwable { + int shardBatchedModificationCount = 3; + doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()). + when(mockActorContext).getDatastoreContext(); + + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + + expectBatchedModifications(actorRef, shardBatchedModificationCount); + + expectReadyTransaction(actorRef); + + YangInstanceIdentifier writePath1 = TestModel.TEST_PATH; + NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH; + NormalizedNode writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME); + + YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH; + NormalizedNode writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME); + + YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH; + NormalizedNode mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH; + NormalizedNode mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME); + + YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH; + NormalizedNode mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME); + + YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH; + YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH; + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + transactionProxy.write(writePath1, writeNode1); + transactionProxy.write(writePath2, writeNode2); + transactionProxy.delete(deletePath1); + transactionProxy.merge(mergePath1, mergeNode1); + transactionProxy.merge(mergePath2, mergeNode2); + transactionProxy.write(writePath3, writeNode3); + transactionProxy.merge(mergePath3, mergeNode3); + transactionProxy.delete(deletePath2); + + // This sends the last batch. + transactionProxy.ready(); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 3, batchedModifications.size()); + + verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1), + new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1)); + + verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1), + new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3)); + + verifyBatchedModifications(batchedModifications.get(2), new MergeModification(mergePath3, mergeNode3), + new DeleteModification(deletePath2)); + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); + } + + @Test + public void testModificationOperationBatchingWithInterleavedReads() throws Throwable { + int shardBatchedModificationCount = 10; + doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()). + when(mockActorContext).getDatastoreContext(); + + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + + expectBatchedModifications(actorRef, shardBatchedModificationCount); + + YangInstanceIdentifier writePath1 = TestModel.TEST_PATH; + NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH; + NormalizedNode writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME); + + YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH; + NormalizedNode mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH; + NormalizedNode mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME); + + YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH; + + doReturn(readSerializedDataReply(writeNode2)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData(writePath2)); + + doReturn(readSerializedDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2)); + + doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + transactionProxy.write(writePath1, writeNode1); + transactionProxy.write(writePath2, writeNode2); + + Optional> readOptional = transactionProxy.read(writePath2). + get(5, TimeUnit.SECONDS); + + assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); + assertEquals("Response NormalizedNode", writeNode2, readOptional.get()); + + transactionProxy.merge(mergePath1, mergeNode1); + transactionProxy.merge(mergePath2, mergeNode2); + + readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS); + + transactionProxy.delete(deletePath); + + Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet(); + assertEquals("Exists response", true, exists); + + assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); + assertEquals("Response NormalizedNode", mergeNode2, readOptional.get()); + + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 3, batchedModifications.size()); + + verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1), + new WriteModification(writePath2, writeNode2)); + + verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1), + new MergeModification(mergePath2, mergeNode2)); + + verifyBatchedModifications(batchedModifications.get(2), new DeleteModification(deletePath)); + + InOrder inOrder = Mockito.inOrder(mockActorContext); + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData(writePath2)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData(mergePath2)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + inOrder.verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); + } + + private List captureBatchedModifications(ActorRef actorRef) { + ArgumentCaptor batchedModificationsCaptor = + ArgumentCaptor.forClass(BatchedModifications.class); + verify(mockActorContext, Mockito.atLeastOnce()).executeOperationAsync( + eq(actorSelection(actorRef)), batchedModificationsCaptor.capture()); + + List batchedModifications = filterCaptured( + batchedModificationsCaptor, BatchedModifications.class); + return batchedModifications; + } + + private List filterCaptured(ArgumentCaptor captor, Class type) { + List captured = new ArrayList<>(); + for(T c: captor.getAllValues()) { + if(type.isInstance(c)) { + captured.add(c); + } + } + + return captured; + } + + private void verifyOneBatchedModification(ActorRef actorRef, Modification expected) { + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); + + verifyBatchedModifications(batchedModifications.get(0), expected); + } + + private void verifyBatchedModifications(Object message, Modification... expected) { + assertEquals("Message type", BatchedModifications.class, message.getClass()); + BatchedModifications batchedModifications = (BatchedModifications)message; + assertEquals("BatchedModifications size", expected.length, batchedModifications.getModifications().size()); + for(int i = 0; i < batchedModifications.getModifications().size(); i++) { + Modification actual = batchedModifications.getModifications().get(i); + assertEquals("Modification type", expected[i].getClass(), actual.getClass()); + assertEquals("getPath", ((AbstractModification)expected[i]).getPath(), + ((AbstractModification)actual).getPath()); + if(actual instanceof WriteModification) { + assertEquals("getData", ((WriteModification)expected[i]).getData(), + ((WriteModification)actual).getData()); + } + } + } }