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=fa2f9187d6059f1585a1475531556d8563db3c5a;hp=e5392e025158704f44d152306aaa727b64d460e8;hb=cccc148d3cd2b5a2a7c86da2e74f2ea43fa00b93;hpb=2dc333588d0c15eb7f2df6223dcdcc15e05b077e 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 e5392e0251..fa2f9187d6 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,28 +1,48 @@ package org.opendaylight.controller.cluster.datastore; -import static org.junit.Assert.assertTrue; 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 akka.actor.ActorPath; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.argThat; +import static org.mockito.Matchers.eq; +import static org.mockito.Matchers.isA; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +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.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.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.ArgumentMatcher; import org.mockito.Mock; +import org.mockito.Mockito; import org.mockito.MockitoAnnotations; - -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY; -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.WRITE_ONLY; -import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_WRITE; - import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; @@ -45,32 +65,22 @@ import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy 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; 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.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.util.List; -import java.util.concurrent.TimeUnit; - -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyString; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.doThrow; -import static org.mockito.Mockito.argThat; -import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.isA; -import static org.mockito.Mockito.times; @SuppressWarnings("resource") -public class TransactionProxyTest extends AbstractActorTest { +public class TransactionProxyTest { @SuppressWarnings("serial") static class TestException extends RuntimeException { @@ -80,6 +90,8 @@ public class TransactionProxyTest extends AbstractActorTest { CheckedFuture invoke(TransactionProxy proxy) throws Exception; } + private static ActorSystem system; + private final Configuration configuration = new MockConfiguration(); @Mock @@ -87,36 +99,70 @@ public class TransactionProxyTest extends AbstractActorTest { 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().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(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); 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(); + if(CreateTransaction.SERIALIZABLE_CLASS.equals(argument.getClass())) { + CreateTransaction obj = CreateTransaction.fromSerializable(argument); + return obj.getTransactionId().startsWith(memberName) && + obj.getTransactionType() == type.ordinal(); + } + + return false; } }; return argThat(matcher); } - private DataExists eqDataExists() { + private DataExists eqSerializedDataExists() { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { @@ -128,7 +174,19 @@ public class TransactionProxyTest extends AbstractActorTest { return argThat(matcher); } - private ReadData eqReadData() { + 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) { @@ -140,17 +198,80 @@ public class TransactionProxyTest extends AbstractActorTest { 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) { + return eqSerializedWriteData(nodeToWrite, DataStoreVersions.CURRENT_VERSION); + } + + private WriteData eqSerializedWriteData(final NormalizedNode nodeToWrite, + final int transactionVersion) { + 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()))) { + + WriteData obj = WriteData.fromSerializable(argument); + return obj.getPath().equals(TestModel.TEST_PATH) && + obj.getData().equals(nodeToWrite); + } + + return false; + } + }; + + return argThat(matcher); + } + private WriteData eqWriteData(final NormalizedNode nodeToWrite) { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - if(!WriteData.SERIALIZABLE_CLASS.equals(argument.getClass())) { - return false; + 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); + } - WriteData obj = WriteData.fromSerializable(argument, schemaContext); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); + private MergeData eqSerializedMergeData(final NormalizedNode nodeToWrite, + final int transactionVersion) { + 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()))) { + + MergeData obj = MergeData.fromSerializable(argument); + return obj.getPath().equals(TestModel.TEST_PATH) && + obj.getData().equals(nodeToWrite); + } + + return false; } }; @@ -161,20 +282,20 @@ public class TransactionProxyTest extends AbstractActorTest { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { - if(!MergeData.SERIALIZABLE_CLASS.equals(argument.getClass())) { - return false; + if(argument instanceof MergeData) { + MergeData obj = ((MergeData) argument); + return obj.getPath().equals(TestModel.TEST_PATH) && + obj.getData().equals(nodeToWrite); } - MergeData obj = MergeData.fromSerializable(argument, schemaContext); - return obj.getPath().equals(TestModel.TEST_PATH) && - obj.getData().equals(nodeToWrite); + return false; } }; return argThat(matcher); } - private DeleteData eqDeleteData() { + private DeleteData eqSerializedDeleteData() { ArgumentMatcher matcher = new ArgumentMatcher() { @Override public boolean matches(Object argument) { @@ -186,54 +307,130 @@ public class TransactionProxyTest extends AbstractActorTest { return argThat(matcher); } - private Future readyTxReply(ActorPath path) { + 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 readDataReply(NormalizedNode data) { - return Futures.successful(new ReadDataReply(schemaContext, data).toSerializable()); + private Future readyTxReply(String path) { + return Futures.successful((Object)new ReadyTransactionReply(path)); + } + + private Future readSerializedDataReply(NormalizedNode data, + short transactionVersion) { + return Futures.successful(new ReadDataReply(data).toSerializable(transactionVersion)); + } + + private Future readSerializedDataReply(NormalizedNode data) { + return readSerializedDataReply(data, DataStoreVersions.CURRENT_VERSION); } - private Future dataExistsReply(boolean exists) { + private Future readDataReply(NormalizedNode data) { + return Futures.successful(new ReadDataReply(data)); + } + + private Future dataExistsSerializedReply(boolean exists) { return Futures.successful(new DataExistsReply(exists).toSerializable()); } - private Future writeDataReply() { - return Futures.successful(new WriteDataReply().toSerializable()); + private Future dataExistsReply(boolean exists) { + 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 incompleteFuture(){ + return mock(Future.class); } - private Future mergeDataReply() { - return Futures.successful(new MergeDataReply().toSerializable()); + private Future mergeDataReply() { + return Futures.successful(new MergeDataReply()); } - private Future deleteDataReply() { - return Futures.successful(new DeleteDataReply().toSerializable()); + private Future deleteSerializedDataReply(short version) { + return Futures.successful(new DeleteDataReply().toSerializable(version)); + } + + private Future deleteSerializedDataReply() { + return deleteSerializedDataReply(DataStoreVersions.CURRENT_VERSION); + } + + private Future deleteDataReply() { + return Futures.successful(new DeleteDataReply()); } private ActorSelection actorSelection(ActorRef actorRef) { return getSystem().actorSelection(actorRef.path()); } - private CreateTransactionReply createTransactionReply(ActorRef actorRef){ + private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ return CreateTransactionReply.newBuilder() .setTransactionActorPath(actorRef.path().toString()) - .setTransactionId("txn-1").build(); + .setTransactionId("txn-1") + .setMessageVersion(transactionVersion) + .build(); } - private ActorRef setupActorContextWithInitialCreateTransaction(TransactionType type) { - ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); - doReturn(getSystem().actorSelection(actorRef.path())). + private ActorRef setupActorContextWithoutInitialCreateTransaction(ActorSystem actorSystem) { + ActorRef actorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + doReturn(actorSystem.actorSelection(actorRef.path())). when(mockActorContext).actorSelection(actorRef.path().toString()); - doReturn(createTransactionReply(actorRef)).when(mockActorContext). - executeShardOperation(eq(DefaultShardStrategy.DEFAULT_SHARD), + + doReturn(Futures.successful(actorSystem.actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + + doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); + + doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); + + 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)); - doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath( - anyString(), eq(actorRef.path().toString())); - doReturn(actorRef.path()).when(mockActorContext).actorFor(actorRef.path().toString()); return actorRef; } + private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { + return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION); + } + + private void propagateReadFailedExceptionCause(CheckedFuture future) throws Throwable { @@ -247,13 +444,13 @@ public class TransactionProxyTest extends AbstractActorTest { @Test public void testRead() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); - doReturn(readDataReply(null)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); Optional> readOptional = transactionProxy.read( TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); @@ -262,8 +459,8 @@ public class TransactionProxyTest extends AbstractActorTest { NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(readDataReply(expectedNode)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); @@ -274,10 +471,10 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = ReadFailedException.class) public void testReadWithInvalidReplyMessageType() throws Exception { - setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeRemoteOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -287,10 +484,10 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = TestException.class) public void testReadWithAsyncRemoteOperatonFailure() throws Throwable { - setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -300,12 +497,19 @@ public class TransactionProxyTest extends AbstractActorTest { private void testExceptionOnInitialCreateTransaction(Exception exToThrow, Invoker invoker) throws Throwable { + ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); - doThrow(exToThrow).when(mockActorContext).executeShardOperation( - anyString(), any()); + if (exToThrow instanceof PrimaryNotFoundException) { + doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); + } else { + doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(anyString()); + } - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), any()); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); propagateReadFailedExceptionCause(invoker.invoke(transactionProxy)); } @@ -337,18 +541,18 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = TestException.class) public void testReadWithPriorRecordingOperationFailure() throws Throwable { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(eq(actorSelection(actorRef)), eqDeleteData()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData()); - doReturn(readDataReply(null)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -360,22 +564,22 @@ public class TransactionProxyTest extends AbstractActorTest { try { propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } finally { - verify(mockActorContext, times(0)).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + verify(mockActorContext, times(0)).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); } } @Test public void testReadWithPriorRecordingOperationSuccessful() throws Throwable { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(expectedNode)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(expectedNode)); - doReturn(readDataReply(expectedNode)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -399,22 +603,40 @@ public class TransactionProxyTest extends AbstractActorTest { transactionProxy.read(TestModel.TEST_PATH); } + @Test(expected=IllegalArgumentException.class) + public void testInvalidCreateTransactionReply() throws Throwable { + ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class)); + + doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext). + actorSelection(actorRef.path().toString()); + + doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + + doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync( + eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY)); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + + propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); + } + @Test public void testExists() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); - doReturn(dataExistsReply(false)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDataExists()); + doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet(); assertEquals("Exists response", false, exists); - doReturn(dataExistsReply(true)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDataExists()); + doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet(); @@ -433,10 +655,10 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = ReadFailedException.class) public void testExistsWithInvalidReplyMessageType() throws Exception { - setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeRemoteOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -446,10 +668,10 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = TestException.class) public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable { - setupActorContextWithInitialCreateTransaction(READ_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(any(ActorSelection.class), any()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); @@ -459,18 +681,18 @@ public class TransactionProxyTest extends AbstractActorTest { @Test(expected = TestException.class) public void testExistsWithPriorRecordingOperationFailure() throws Throwable { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(eq(actorSelection(actorRef)), eqDeleteData()); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDeleteData()); - doReturn(dataExistsReply(false)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDataExists()); + doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -482,22 +704,22 @@ public class TransactionProxyTest extends AbstractActorTest { try { propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); } finally { - verify(mockActorContext, times(0)).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDataExists()); + verify(mockActorContext, times(0)).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); } } @Test public void testExistsWithPriorRecordingOperationSuccessful() throws Throwable { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - doReturn(dataExistsReply(true)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDataExists()); + doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDataExists()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -510,7 +732,7 @@ public class TransactionProxyTest extends AbstractActorTest { } @Test(expected=IllegalStateException.class) - public void testxistsPreConditionCheck() { + public void testExistsPreConditionCheck() { TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -543,23 +765,79 @@ public class TransactionProxyTest extends AbstractActorTest { @Test public void testWrite() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - verify(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + WriteDataReply.class); + } + + @Test + public void testWriteAfterAsyncRead() throws Throwable { + ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem()); + + 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()); + + final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); + + 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(); + } + + verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.SERIALIZABLE_CLASS); + WriteDataReply.class); } @Test(expected=IllegalStateException.class) @@ -586,56 +864,55 @@ public class TransactionProxyTest extends AbstractActorTest { @Test public void testMerge() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqMergeData(nodeToWrite)); + doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); - verify(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqMergeData(nodeToWrite)); + verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.SERIALIZABLE_CLASS); + MergeDataReply.class); } @Test public void testDelete() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - doReturn(deleteDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDeleteData()); + doReturn(deleteSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDeleteData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.delete(TestModel.TEST_PATH); - verify(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqDeleteData()); + verify(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedDeleteData()); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - DeleteDataReply.SERIALIZABLE_CLASS); + DeleteDataReply.class); } - private void verifyCohortPathFutures(ThreePhaseCommitCohortProxy proxy, - Object... expReplies) throws Exception { + private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy, + Object... expReplies) throws Exception { assertEquals("getReadyOperationFutures size", expReplies.length, - proxy.getCohortPathFutures().size()); + proxy.getCohortFutures().size()); int i = 0; - for( Future future: proxy.getCohortPathFutures()) { + for( Future future: proxy.getCohortFutures()) { assertNotNull("Ready operation Future is null", future); Object expReply = expReplies[i++]; - if(expReply instanceof ActorPath) { - ActorPath actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS)); + if(expReply instanceof ActorSelection) { + ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS)); assertEquals("Cohort actor path", expReply, actual); } else { // Expecting exception. @@ -649,20 +926,19 @@ public class TransactionProxyTest extends AbstractActorTest { } } - @SuppressWarnings("unchecked") @Test public void testReady() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(readDataReply(null)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - doReturn(readyTxReply(actorRef.path())).when(mockActorContext).executeRemoteOperationAsync( + doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, @@ -679,27 +955,91 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - WriteDataReply.SERIALIZABLE_CLASS); + WriteDataReply.class); - verifyCohortPathFutures(proxy, actorRef.path()); + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + } + + private ActorRef testCompatibilityWithHeliumVersion(short version) throws Exception { + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), + READ_WRITE, version); + + NormalizedNode testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(readSerializedDataReply(testNode, version)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); + + doReturn(writeSerializedDataReply(version)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(testNode, version)); + + doReturn(mergeSerializedDataReply(version)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedMergeData(testNode, version)); + + doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); + + doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()), + eq(actorRef.path().toString())); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + + Optional> readOptional = transactionProxy.read(TestModel.TEST_PATH). + get(5, TimeUnit.SECONDS); + + assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); + assertEquals("Response NormalizedNode", testNode, readOptional.get()); + + transactionProxy.write(TestModel.TEST_PATH, testNode); + + transactionProxy.merge(TestModel.TEST_PATH, testNode); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + + ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + ShardTransactionMessages.WriteDataReply.class, ShardTransactionMessages.MergeDataReply.class); + + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); + + return actorRef; + } + + @Test + public void testCompatibilityWithBaseHeliumVersion() throws Exception { + ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.BASE_HELIUM_VERSION); + + verify(mockActorContext).resolvePath(eq(actorRef.path().toString()), + eq(actorRef.path().toString())); + } + + @Test + public void testCompatibilityWithHeliumR1Version() throws Exception { + ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.HELIUM_1_VERSION); + + verify(mockActorContext, Mockito.never()).resolvePath(eq(actorRef.path().toString()), + eq(actorRef.path().toString())); } - @SuppressWarnings("unchecked") @Test public void testReadyWithRecordingOperationFailure() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqMergeData(nodeToWrite)); + doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); - doReturn(readyTxReply(actorRef.path())).when(mockActorContext).executeRemoteOperationAsync( + 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); @@ -713,24 +1053,23 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.SERIALIZABLE_CLASS, TestException.class); + verifyCohortFutures(proxy, TestException.class); - verifyCohortPathFutures(proxy, TestException.class); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + MergeDataReply.class, TestException.class); } - @SuppressWarnings("unchecked") @Test public void testReadyWithReplyFailure() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(mergeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqMergeData(nodeToWrite)); + doReturn(mergeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedMergeData(nodeToWrite)); doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeRemoteOperationAsync(eq(actorSelection(actorRef)), + executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, @@ -745,16 +1084,16 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - MergeDataReply.SERIALIZABLE_CLASS); + MergeDataReply.class); - verifyCohortPathFutures(proxy, TestException.class); + verifyCohortFutures(proxy, TestException.class); } @Test public void testReadyWithInitialCreateTransactionFailure() throws Exception { - doThrow(new PrimaryNotFoundException("mock")).when(mockActorContext).executeShardOperation( - anyString(), any()); + doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when( + mockActorContext).findPrimaryShardAsync(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -773,21 +1112,20 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortPathFutures(proxy, PrimaryNotFoundException.class); + verifyCohortFutures(proxy, PrimaryNotFoundException.class); } - @SuppressWarnings("unchecked") @Test public void testReadyWithInvalidReplyMessageType() throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(WRITE_ONLY); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(writeDataReply()).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqWriteData(nodeToWrite)); + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedWriteData(nodeToWrite)); doReturn(Futures.successful(new Object())).when(mockActorContext). - executeRemoteOperationAsync(eq(actorSelection(actorRef)), + executeOperationAsync(eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, @@ -801,12 +1139,12 @@ public class TransactionProxyTest extends AbstractActorTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortPathFutures(proxy, IllegalArgumentException.class); + verifyCohortFutures(proxy, IllegalArgumentException.class); } @Test public void testGetIdentifier() { - setupActorContextWithInitialCreateTransaction(READ_ONLY); + setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, TransactionProxy.TransactionType.READ_ONLY); @@ -815,13 +1153,12 @@ public class TransactionProxyTest extends AbstractActorTest { assertTrue("Invalid identifier: " + id, id.toString().startsWith(memberName)); } - @SuppressWarnings("unchecked") @Test public void testClose() throws Exception{ - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); - doReturn(readDataReply(null)).when(mockActorContext).executeRemoteOperationAsync( - eq(actorSelection(actorRef)), eqReadData()); + doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), eqSerializedReadData()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -830,7 +1167,564 @@ public class TransactionProxyTest extends AbstractActorTest { transactionProxy.close(); - verify(mockActorContext).sendRemoteOperationAsync( + verify(mockActorContext).sendOperationAsync( eq(actorSelection(actorRef)), isA(CloseTransaction.SERIALIZABLE_CLASS)); } + + + /** + * Method to test a local Tx actor. The Tx paths are matched to decide if the + * Tx actor is local or not. This is done by mocking the Tx actor path + * and the caller paths and ensuring that the paths have the remote-address format + * + * Note: Since the default akka provider for test is not a RemoteActorRefProvider, + * the paths returned for the actors for all the tests are not qualified remote paths. + * Hence are treated as non-local/remote actors. In short, all tests except + * few below run for remote actors + * + * @throws Exception + */ + @Test + public void testLocalTxActorRead() throws Exception { + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + + doReturn(actorSystem.actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + 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(); + + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(memberName, READ_ONLY)); + + doReturn(true).when(mockActorContext).isPathLocal(actorPath); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY); + + // negative test case with null as the reply + doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqReadData()); + + Optional> readOptional = transactionProxy.read( + TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); + + assertEquals("NormalizedNode isPresent", false, readOptional.isPresent()); + + // test case with node as read data reply + NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqReadData()); + + readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); + + assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); + + assertEquals("Response NormalizedNode", expectedNode, readOptional.get()); + + // test for local data exists + doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDataExists()); + + boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet(); + + assertEquals("Exists response", true, exists); + } + + @Test + public void testLocalTxActorWrite() throws Exception { + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + + doReturn(actorSystem.actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + 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(); + + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). + 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)); + + 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()); + + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + WriteDataReply.class, MergeDataReply.class, DeleteDataReply.class); + + // testing ready + doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), isA(ReadyTransaction.class)); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); + + ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; + + verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path())); + } + + private static interface TransactionProxyOperation { + void run(TransactionProxy transactionProxy); + } + + 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) + .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(); + + Assert.assertTrue(String.format("took less time than expected %s was %s", + TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()), + (end-start)), (end - start) > TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds())); + + } + + 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) + .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(); + + Assert.assertTrue(String.format("took more time than expected %s was %s", + TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()), + (end-start)), (end - start) <= TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds())); + } + + public void testWriteThrottling(boolean shardFound){ + + throttleOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(nodeToWrite)); + + 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); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(nodeToWrite)); + + 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); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(nodeToWrite)); + + 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); + + doReturn(writeSerializedDataReply()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqSerializedWriteData(nodeToWrite)); + + 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); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqMergeData(nodeToMerge)); + + 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); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqMergeData(nodeToMerge)); + + 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); + + doReturn(mergeDataReply()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqMergeData(nodeToMerge)); + + 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) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDeleteData()); + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }); + } + + + @Test + public void testDeleteThrottlingWhenShardNotFound(){ + + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDeleteData()); + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }, false); + } + + @Test + public void testDeleteCompletion(){ + completeOperation(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + doReturn(deleteDataReply()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqDeleteData()); + + 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); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(nodeToWrite)); + + 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); + + doReturn(writeDataReply()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(nodeToWrite)); + + doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( + any(ActorSelection.class), eqWriteData(carsNode)); + + 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); + } }