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=8278d3cffceaa6b82357c08c4ab62e33ff53fafe;hp=7407897dfac504872e37d35c374397369a8c5576;hb=12a411738c5d9030a1a196f04fdf5ec0f4c3395e;hpb=d80bf0f81bdeed907b290b67f26f1a3541ad3ea4 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 7407897dfa..8278d3cffc 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 @@ -1,76 +1,62 @@ 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; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +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 akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.ActorSystem; import akka.actor.Props; import akka.dispatch.Futures; -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.typesafe.config.Config; -import com.typesafe.config.ConfigFactory; -import org.junit.AfterClass; -import org.junit.Before; -import org.junit.BeforeClass; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Uninterruptibles; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Assert; import org.junit.Test; -import org.mockito.ArgumentMatcher; -import org.mockito.Mock; -import org.mockito.MockitoAnnotations; -import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; +import org.mockito.InOrder; +import org.mockito.Mockito; 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; -import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; -import org.opendaylight.controller.cluster.datastore.messages.DeleteData; -import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply; -import org.opendaylight.controller.cluster.datastore.messages.MergeData; -import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply; -import org.opendaylight.controller.cluster.datastore.messages.ReadData; -import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; 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.DeleteModification; +import org.opendaylight.controller.cluster.datastore.modification.MergeModification; +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; import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor; -import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration; +import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException; 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; -import java.io.IOException; -import java.util.List; -import java.util.concurrent.TimeUnit; -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.Mockito.argThat; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.isA; -import static org.mockito.Mockito.times; -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; @SuppressWarnings("resource") -public class TransactionProxyTest { +public class TransactionProxyTest extends AbstractTransactionProxyTest { @SuppressWarnings("serial") static class TestException extends RuntimeException { @@ -80,311 +66,11 @@ public class TransactionProxyTest { CheckedFuture invoke(TransactionProxy proxy) throws Exception; } - private static ActorSystem system; - - private final Configuration configuration = new MockConfiguration(); - - @Mock - private ActorContext mockActorContext; - - private SchemaContext schemaContext; - - @Mock - private ClusterWrapper mockClusterWrapper; - - String memberName = "mock-member"; - - @BeforeClass - public static void setUpClass() throws IOException { - - Config config = ConfigFactory.parseMap(ImmutableMap.builder(). - put("akka.actor.default-dispatcher.type", - "akka.testkit.CallingThreadDispatcherConfigurator").build()). - withFallback(ConfigFactory.load()); - system = ActorSystem.create("test", config); - } - - @AfterClass - public static void tearDownClass() throws IOException { - JavaTestKit.shutdownActorSystem(system); - system = null; - } - - @Before - public void setUp(){ - MockitoAnnotations.initMocks(this); - - schemaContext = TestModel.createTestContext(); - - DatastoreContext dataStoreContext = DatastoreContext.newBuilder().build(); - - doReturn(getSystem()).when(mockActorContext).getActorSystem(); - 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(); - - ShardStrategyFactory.setConfiguration(configuration); - } - - private ActorSystem getSystem() { - return system; - } - - private CreateTransaction eqCreateTransaction(final String memberName, - final TransactionType type) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - CreateTransaction obj = CreateTransaction.fromSerializable(argument); - return obj.getTransactionId().startsWith(memberName) && - obj.getTransactionType() == type.ordinal(); - } - }; - - return argThat(matcher); - } - - private DataExists eqSerializedDataExists() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) && - DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private DataExists eqDataExists() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return (argument instanceof DataExists) && - ((DataExists)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private ReadData eqSerializedReadData() { - 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); - } - }; - - return argThat(matcher); - } - - private ReadData eqReadData() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return (argument instanceof ReadData) && - ((ReadData)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private WriteData eqSerializedWriteData(final NormalizedNode nodeToWrite) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(!WriteData.SERIALIZABLE_CLASS.equals(argument.getClass())) { - return false; - } - - WriteData obj = WriteData.fromSerializable(argument, schemaContext); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); - } - }; - - 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) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(!MergeData.SERIALIZABLE_CLASS.equals(argument.getClass())) { - return false; - } - - MergeData obj = MergeData.fromSerializable(argument, schemaContext); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); - } - }; - - 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() { - 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 argThat(matcher); - } - - private Future readySerializedTxReply(String path) { - return Futures.successful((Object)new ReadyTransactionReply(path).toSerializable()); - } - - private Future readyTxReply(String path) { - return Futures.successful((Object)new ReadyTransactionReply(path)); - } - - - private Future readSerializedDataReply(NormalizedNode data) { - return Futures.successful(new ReadDataReply(schemaContext, data).toSerializable()); - } - - private Future readDataReply(NormalizedNode data) { - return Futures.successful(new ReadDataReply(schemaContext, data)); - } - - private Future dataExistsSerializedReply(boolean exists) { - return Futures.successful(new DataExistsReply(exists).toSerializable()); - } - - private Future dataExistsReply(boolean exists) { - return Futures.successful(new DataExistsReply(exists)); - } - - private Future writeSerializedDataReply() { - return Futures.successful(new WriteDataReply().toSerializable()); - } - - private Future writeDataReply() { - return Futures.successful(new WriteDataReply()); - } - - private Future mergeSerializedDataReply() { - return Futures.successful(new MergeDataReply().toSerializable()); - } - - private Future mergeDataReply() { - return Futures.successful(new MergeDataReply()); - } - - private Future deleteSerializedDataReply() { - return Futures.successful(new DeleteDataReply().toSerializable()); - } - - private Future deleteDataReply() { - return Futures.successful(new DeleteDataReply()); - } - - private ActorSelection actorSelection(ActorRef actorRef) { - return getSystem().actorSelection(actorRef.path()); - } - - private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ - return CreateTransactionReply.newBuilder() - .setTransactionActorPath(actorRef.path().toString()) - .setTransactionId("txn-1") - .setMessageVersion(transactionVersion) - .build(); - } - - private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type, int transactionVersion) { - ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); - doReturn(actorSystem.actorSelection(actorRef.path())). - when(mockActorContext).actorSelection(actorRef.path().toString()); - - 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()); - - return actorRef; - } - - private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { - return setupActorContextWithInitialCreateTransaction(actorSystem, type, CreateTransaction.CURRENT_VERSION); - } - - - private void propagateReadFailedExceptionCause(CheckedFuture future) - throws Throwable { - - try { - future.checkedGet(5, TimeUnit.SECONDS); - fail("Expected ReadFailedException"); - } catch(ReadFailedException e) { - throw e.getCause(); - } - } - @Test 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()); @@ -413,8 +99,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); } @@ -426,8 +111,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)); } @@ -478,21 +162,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); @@ -512,14 +194,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); @@ -527,16 +207,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); } @@ -562,8 +245,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()); @@ -610,23 +292,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()); @@ -652,28 +332,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); } @@ -694,7 +376,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()); } } @@ -706,41 +388,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.SERIALIZABLE_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)); + + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); - transactionProxy.write(TestModel.TEST_PATH, - ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + 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 @@ -749,62 +483,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.SERIALIZABLE_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()); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - DeleteDataReply.SERIALIZABLE_CLASS); - } - - private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy, - Object... expReplies) throws Exception { - assertEquals("getReadyOperationFutures size", expReplies.length, - proxy.getCohortFutures().size()); + // This sends the batched modification. + transactionProxy.ready(); - int i = 0; - for( Future future: proxy.getCohortFutures()) { - assertNotNull("Ready operation Future is null", future); + verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH)); - Object expReply = expReplies[i++]; - if(expReply instanceof ActorSelection) { - ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - assertEquals("Cohort actor path", expReply, actual); - } else { - // Expecting exception. - try { - Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - fail("Expected exception from ready operation Future"); - } catch(Exception e) { - // Expected - } - } - } + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class); } @Test @@ -816,14 +528,10 @@ public class TransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + expectBatchedModifications(actorRef, 1); + expectReadyTransaction(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -836,72 +544,27 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.SERIALIZABLE_CLASS); + BatchedModificationsReply.class); verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + + verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), + isA(BatchedModifications.class)); } @Test - public void testReadyForwardCompatibility() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE, 0); + public void testReadyWithRecordingOperationFailure() throws Exception { + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedReadData()); - - doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + expectFailedBatchedModifications(actorRef); - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + expectReadyTransaction(actorRef); - doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()), - eq(actorRef.path().toString())); - - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_WRITE); + doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); - transactionProxy.read(TestModel.TEST_PATH); - - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - - DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.SERIALIZABLE_CLASS); - - verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); - - verify(mockActorContext).resolvePath(eq(actorRef.path().toString()), - eq(actorRef.path().toString())); - } - - @Test - public void testReadyWithRecordingOperationFailure() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - - NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - - doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); - - 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)); - - 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); @@ -913,8 +576,7 @@ public class TransactionProxyTest { verifyCohortFutures(proxy, TestException.class); - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.SERIALIZABLE_CLASS, TestException.class); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), TestException.class); } @Test @@ -923,15 +585,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); @@ -942,7 +602,7 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.SERIALIZABLE_CLASS); + BatchedModificationsReply.class); verifyCohortFutures(proxy, TestException.class); } @@ -953,8 +613,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); @@ -979,15 +638,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); @@ -1000,6 +657,17 @@ public class TransactionProxyTest { verifyCohortFutures(proxy, IllegalArgumentException.class); } + @Test + public void testUnusedTransaction() throws Exception { + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertEquals("canCommit", true, ready.canCommit().get()); + ready.preCommit().get(); + ready.commit().get(); + } + @Test public void testGetIdentifier() { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); @@ -1018,8 +686,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); @@ -1055,9 +722,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())), @@ -1098,7 +763,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)); @@ -1109,59 +774,593 @@ 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); + + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - verify(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqWriteData(nodeToWrite)); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class); + + // testing ready + doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), isA(ReadyTransaction.class)); - //testing local merge - doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToWrite)); + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - verify(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqMergeData(nodeToWrite)); + ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path())); + } - //testing local delete - doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), eqDeleteData()); + private static interface TransactionProxyOperation { + void run(TransactionProxy transactionProxy); + } - transactionProxy.delete(TestModel.TEST_PATH); + private void throttleOperation(TransactionProxyOperation operation) { + throttleOperation(operation, 1, true); + } + + private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){ + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + + doReturn(outstandingOpsLimit).when(mockActorContext).getTransactionOutstandingOperationLimit(); + + doReturn(actorSystem.actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + if(shardFound) { + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + } else { + doReturn(Futures.failed(new Exception("not found"))) + .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). + setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); + + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(memberName, READ_WRITE)); + + doReturn(true).when(mockActorContext).isPathLocal(actorPath); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + long start = System.nanoTime(); + + operation.run(transactionProxy); + + long end = System.nanoTime(); + + long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()); + Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", + expected, (end-start)), (end - start) > expected); + + } + + private void completeOperation(TransactionProxyOperation operation){ + completeOperation(operation, true); + } + + private void completeOperation(TransactionProxyOperation operation, boolean shardFound){ + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + + doReturn(1).when(mockActorContext).getTransactionOutstandingOperationLimit(); + + doReturn(actorSystem.actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + if(shardFound) { + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + } else { + doReturn(Futures.failed(new Exception("not found"))) + .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). + setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); + + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(memberName, READ_WRITE)); + + doReturn(true).when(mockActorContext).isPathLocal(actorPath); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + long start = System.nanoTime(); + + operation.run(transactionProxy); + + long end = System.nanoTime(); + + 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){ + + 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(){ + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectIncompleteBatchedModifications(); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + } + }); + } + + @Test + public void testWriteThrottlingWhenShardNotFound(){ + // Confirm that there is no throttling when the Shard is not found + completeOperation(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); + } + }, false); + + } + + + @Test + public void testWriteCompletion(){ + completeOperation(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); + } + }); + } + + @Test + public void testMergeThrottlingWhenShardFound(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectIncompleteBatchedModifications(); - verify(mockActorContext).executeOperationAsync(any(ActorSelection.class), eqDeleteData()); + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + } + }); + } + + @Test + public void testMergeThrottlingWhenShardNotFound(){ + + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectBatchedModifications(2); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + } + }, false); + } + + @Test + public void testMergeCompletion(){ + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectBatchedModifications(2); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge); + } + }); + + } + + @Test + public void testDeleteThrottlingWhenShardFound(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + expectIncompleteBatchedModifications(); + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }); + } + + + @Test + public void testDeleteThrottlingWhenShardNotFound(){ + + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + expectBatchedModifications(2); + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }, false); + } + + @Test + public void testDeleteCompletion(){ + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + expectBatchedModifications(2); + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }); + + } + + @Test + public void testReadThrottlingWhenShardFound(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqReadData()); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + } + }); + } + + @Test + public void testReadThrottlingWhenShardNotFound(){ + + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqReadData()); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + } + }, false); + } + + + @Test + public void testReadCompletion(){ + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqReadData()); + + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + } + }); + + } + + @Test + public void testExistsThrottlingWhenShardFound(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDataExists()); + + transactionProxy.exists(TestModel.TEST_PATH); + + transactionProxy.exists(TestModel.TEST_PATH); + } + }); + } + + @Test + public void testExistsThrottlingWhenShardNotFound(){ + + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDataExists()); + + transactionProxy.exists(TestModel.TEST_PATH); + + transactionProxy.exists(TestModel.TEST_PATH); + } + }, false); + } + + + @Test + public void testExistsCompletion(){ + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDataExists()); + + transactionProxy.exists(TestModel.TEST_PATH); + + transactionProxy.exists(TestModel.TEST_PATH); + } + }); + + } + + @Test + public void testReadyThrottling(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectBatchedModifications(1); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), any(ReadyTransaction.class)); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + + transactionProxy.ready(); + } + }); + } + + @Test + public void testReadyThrottlingWithTwoTransactionContexts(){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + NormalizedNode carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME); + + expectBatchedModifications(2); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), any(ReadyTransaction.class)); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); + + transactionProxy.write(TestModel.TEST_PATH, carsNode); + + transactionProxy.ready(); + } + }, 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(), - WriteDataReply.class, MergeDataReply.class, DeleteDataReply.class); + BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); + } - // testing ready - doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(ReadyTransaction.class)); + @Test + public void testModificationOperationBatchingWithInterleavedReads() throws Throwable { + int shardBatchedModificationCount = 10; + doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()). + when(mockActorContext).getDatastoreContext(); - DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); - assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + expectBatchedModifications(actorRef, shardBatchedModificationCount); - ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + YangInstanceIdentifier writePath1 = TestModel.TEST_PATH; + NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path())); + 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); } }