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=a5625637f4b5720d2d8084dcd415ab94b31a0c7a;hp=93c6ddbe7398be9b2054bb67a5b9b69bac47c09c;hb=482caffb6cc503ab1d156109a5dba8c2fe675f3d;hpb=228af4aa1ef1a802fd24e7e010f3bba959ee03dd 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 93c6ddbe73..a5625637f4 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,3 +1,11 @@ +/* + * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved. + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v1.0 which accompanies this distribution, + * and is available at http://www.eclipse.org/legal/epl-v10.html + */ + package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; @@ -11,14 +19,15 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; 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 static org.opendaylight.controller.cluster.datastore.TransactionType.READ_ONLY; +import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE; +import static org.opendaylight.controller.cluster.datastore.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.util.Timeout; import com.google.common.base.Optional; import com.google.common.collect.Sets; import com.google.common.util.concurrent.CheckedFuture; @@ -34,7 +43,7 @@ import org.junit.Assert; import org.junit.Test; import org.mockito.InOrder; import org.mockito.Mockito; -import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; +import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; @@ -42,13 +51,14 @@ import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; +import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; 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.utils.DoNothingActor; import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregatorTest; +import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; @@ -58,6 +68,9 @@ import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCoh import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Promise; @@ -77,7 +90,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { public void testRead() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); @@ -106,7 +119,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS); } @@ -118,7 +131,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } @@ -130,14 +143,14 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { if (exToThrow instanceof PrimaryNotFoundException) { doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); } else { - doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), actorRef)). when(mockActorContext).findPrimaryShardAsync(anyString()); } doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), any()); + any(ActorSelection.class), any(), any(Timeout.class)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(invoker.invoke(transactionProxy)); } @@ -178,7 +191,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, expectedNode); @@ -198,7 +211,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testReadPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.read(TestModel.TEST_PATH); } @@ -209,13 +222,14 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext). actorSelection(actorRef.path().toString()); - doReturn(Futures.successful(getSystem().actorSelection(actorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), actorRef)). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync( - eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY)); + eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY), + any(Timeout.class)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH)); } @@ -224,7 +238,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { public void testExists() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); doReturn(dataExistsSerializedReply(false)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); @@ -258,8 +272,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS); } @@ -271,7 +284,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.failed(new TestException())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH)); } @@ -287,7 +300,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -305,7 +318,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testExistsPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.exists(TestModel.TEST_PATH); } @@ -318,7 +331,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -332,7 +345,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { Promise createTxPromise = akka.dispatch.Futures.promise(); doReturn(createTxPromise).when(mockActorContext).executeOperationAsync( eq(getSystem().actorSelection(actorRef.path())), - eqCreateTransaction(memberName, READ_WRITE)); + eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class)); doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); @@ -341,7 +354,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - final TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + final TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); final CountDownLatch readComplete = new CountDownLatch(1); final AtomicReference caughtEx = new AtomicReference<>(); @@ -381,13 +394,13 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test(expected=IllegalStateException.class) public void testWritePreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); } @Test(expected=IllegalStateException.class) public void testWriteAfterReadyPreConditionCheck() { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.ready(); @@ -403,7 +416,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); @@ -417,7 +430,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.delete(TestModel.TEST_PATH); @@ -435,7 +448,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(actorRef, 1); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -463,7 +476,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -493,7 +506,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -518,7 +531,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef1); expectBatchedModificationsReady(actorRef2); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME)); transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); @@ -541,7 +554,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -570,7 +583,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -602,7 +615,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectFailedBatchedModifications(actorRef); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); @@ -613,10 +626,29 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class); } + @Test + public void testReadyWithDebugContextEnabled() throws Exception { + dataStoreContextBuilder.transactionDebugContextEnabled(true); + + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + + expectBatchedModificationsReady(actorRef, true); + + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); + + transactionProxy.merge(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); + + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof DebugThreePhaseCommitCohort); + + verifyCohortFutures((DebugThreePhaseCommitCohort)ready, new CommitTransactionReply().toSerializable()); + } + private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception { doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -660,7 +692,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef2); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY); transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME)); transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)); @@ -676,8 +708,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testGetIdentifier() { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, - TransactionProxy.TransactionType.READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); Object id = transactionProxy.getIdentifier(); assertNotNull("getIdentifier returned null", id); @@ -691,7 +722,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(readSerializedDataReply(null)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedReadData()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.read(TestModel.TEST_PATH); @@ -719,7 +750,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); doReturn(true).when(mockActorContext).isPathLocal(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); // negative test case with null as the reply doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync( @@ -758,7 +789,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModificationsReady(actorRef, true); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); @@ -779,34 +810,53 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { } private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound){ + throttleOperation(operation, outstandingOpsLimit, shardFound, TimeUnit.MILLISECONDS.toNanos( + mockActorContext.getDatastoreContext().getOperationTimeoutInMillis())); + } + + private PrimaryShardInfo newPrimaryShardInfo(ActorRef actorRef){ + return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION, + Optional.absent()); + } + + private PrimaryShardInfo newPrimaryShardInfo(ActorRef actorRef, Optional dataTreeOptional){ + return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION, + dataTreeOptional); + } + + + private void throttleOperation(TransactionProxyOperation operation, int outstandingOpsLimit, boolean shardFound, long expectedCompletionTime){ ActorSystem actorSystem = getSystem(); ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); - doReturn(outstandingOpsLimit).when(mockActorContext).getTransactionOutstandingOperationLimit(); + // Note that we setting batchedModificationCount to one less than what we need because in TransactionProxy + // we now allow one extra permit to be allowed for ready + doReturn(dataStoreContextBuilder.operationTimeoutInSeconds(2). + shardBatchedModificationCount(outstandingOpsLimit-1).build()).when(mockActorContext).getDatastoreContext(); doReturn(actorSystem.actorSelection(shardActorRef.path())). when(mockActorContext).actorSelection(shardActorRef.path().toString()); if(shardFound) { - doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))). + when(mockActorContext).findPrimaryShardAsync(eq("cars")); + } 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(incompleteFuture()).when(mockActorContext). + executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), + eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class)); doReturn(true).when(mockActorContext).isPathLocal(actorPath); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); long start = System.nanoTime(); @@ -814,9 +864,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { 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); + expectedCompletionTime, (end-start)), + ((end - start) > expectedCompletionTime) && ((end - start) < expectedCompletionTime*2)); } @@ -828,13 +878,11 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { 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()))). + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } else { doReturn(Futures.failed(new PrimaryNotFoundException("test"))) @@ -851,11 +899,11 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), - eqCreateTransaction(memberName, READ_WRITE)); + eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class)); doReturn(true).when(mockActorContext).isPathLocal(anyString()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); long start = System.nanoTime(); @@ -863,30 +911,77 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { long end = System.nanoTime(); - long expected = TimeUnit.SECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInSeconds()); + long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()); Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", expected, (end-start)), (end - start) <= expected); } - public void testWriteThrottling(boolean shardFound){ + private void completeOperationLocal(TransactionProxyOperation operation, Optional dataTreeOptional){ + ActorSystem actorSystem = getSystem(); + ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); - throttleOperation(new TransactionProxyOperation() { + doReturn(actorSystem.actorSelection(shardActorRef.path())). + when(mockActorContext).actorSelection(shardActorRef.path().toString()); + + doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, dataTreeOptional))). + when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); + + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); + + long start = System.nanoTime(); + + operation.run(transactionProxy); + + long end = System.nanoTime(); + + long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()); + Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", + expected, (end-start)), (end - start) <= expected); + } + + private static Optional createDataTree(){ + DataTree dataTree = mock(DataTree.class); + Optional dataTreeOptional = Optional.of(dataTree); + DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class); + DataTreeModification dataTreeModification = mock(DataTreeModification.class); + + doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot(); + doReturn(dataTreeModification).when(dataTreeSnapshot).newModification(); + + return dataTreeOptional; + } + + private static Optional createDataTree(NormalizedNode readResponse){ + DataTree dataTree = mock(DataTree.class); + Optional dataTreeOptional = Optional.of(dataTree); + DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class); + DataTreeModification dataTreeModification = mock(DataTreeModification.class); + + doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot(); + doReturn(dataTreeModification).when(dataTreeSnapshot).newModification(); + doReturn(Optional.of(readResponse)).when(dataTreeModification).readNode(any(YangInstanceIdentifier.class)); + + return dataTreeOptional; + } + + + @Test + public void testWriteCompletionForLocalShard(){ + completeOperationLocal(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); + }, createDataTree()); } @Test public void testWriteThrottlingWhenShardFound(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -904,7 +999,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testWriteThrottlingWhenShardNotFound(){ // Confirm that there is no throttling when the Shard is not found - dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -923,7 +1017,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testWriteCompletion(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -940,7 +1033,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testMergeThrottlingWhenShardFound(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -957,7 +1049,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testMergeThrottlingWhenShardNotFound(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -974,7 +1065,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testMergeCompletion(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -990,6 +1080,22 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { } + @Test + public void testMergeCompletionForLocalShard(){ + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); + + transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); + + } + }, createDataTree()); + } + + @Test public void testDeleteThrottlingWhenShardFound(){ @@ -1021,9 +1127,22 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { }, false); } + @Test + public void testDeleteCompletionForLocalShard(){ + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + + transactionProxy.delete(TestModel.TEST_PATH); + + transactionProxy.delete(TestModel.TEST_PATH); + } + }, createDataTree()); + + } + @Test public void testDeleteCompletion(){ - dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1088,6 +1207,33 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { } + @Test + public void testReadCompletionForLocalShard(){ + final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + } + }, createDataTree(nodeToRead)); + + } + + @Test + public void testReadCompletionForLocalShardWhenExceptionOccurs(){ + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + transactionProxy.read(TestModel.TEST_PATH); + + transactionProxy.read(TestModel.TEST_PATH); + } + }, createDataTree()); + + } + @Test public void testExistsThrottlingWhenShardFound(){ @@ -1137,6 +1283,32 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { } + @Test + public void testExistsCompletionForLocalShard(){ + final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + transactionProxy.exists(TestModel.TEST_PATH); + + transactionProxy.exists(TestModel.TEST_PATH); + } + }, createDataTree(nodeToRead)); + + } + + @Test + public void testExistsCompletionForLocalShardWhenExceptionOccurs(){ + completeOperationLocal(new TransactionProxyOperation() { + @Override + public void run(TransactionProxy transactionProxy) { + transactionProxy.exists(TestModel.TEST_PATH); + + transactionProxy.exists(TestModel.TEST_PATH); + } + }, createDataTree()); + + } @Test public void testReadyThrottling(){ @@ -1147,9 +1319,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { expectBatchedModifications(1); - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), any(ReadyTransaction.class)); - transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); transactionProxy.ready(); @@ -1159,7 +1328,6 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { @Test public void testReadyThrottlingWithTwoTransactionContexts(){ - throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1173,11 +1341,13 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - transactionProxy.write(TestModel.TEST_PATH, carsNode); + // Trying to write to Cars will cause another transaction context to get created + transactionProxy.write(CarsModel.BASE_PATH, carsNode); + // Now ready should block for both transaction contexts transactionProxy.ready(); } - }, 2, true); + }, 1, true, TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()) * 2); } private void testModificationOperationBatching(TransactionType type) throws Exception { @@ -1209,7 +1379,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH; YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH; - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, type); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, type); transactionProxy.write(writePath1, writeNode1); transactionProxy.write(writePath2, writeNode2); @@ -1287,7 +1457,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(dataExistsSerializedReply(true)).when(mockActorContext).executeOperationAsync( eq(actorSelection(actorRef)), eqSerializedDataExists()); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE); transactionProxy.write(writePath1, writeNode1); transactionProxy.write(writePath2, writeNode2); @@ -1359,11 +1529,9 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(memberName).when(mockActorContext).getCurrentMemberName(); - doReturn(10).when(mockActorContext).getTransactionOutstandingOperationLimit(); - doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher(); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY); Optional> readOptional = transactionProxy.read( YangInstanceIdentifier.builder().build()).get(5, TimeUnit.SECONDS); @@ -1399,7 +1567,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(getSystem().actorSelection(shardActorRef.path())). when(mockActorContext).actorSelection(shardActorRef.path().toString()); - doReturn(Futures.successful(getSystem().actorSelection(shardActorRef.path()))). + doReturn(primaryShardInfoReply(getSystem(), shardActorRef)). when(mockActorContext).findPrimaryShardAsync(eq(shardName)); doReturn(true).when(mockActorContext).isPathLocal(shardActorRef.path().toString()); @@ -1411,7 +1579,7 @@ public class TransactionProxyTest extends AbstractTransactionProxyTest { doReturn(Futures.successful(createTransactionReply(txActorRef, DataStoreVersions.CURRENT_VERSION))).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), - eqCreateTransaction(memberName, TransactionType.READ_ONLY)); + eqCreateTransaction(memberName, TransactionType.READ_ONLY), any(Timeout.class)); doReturn(readSerializedDataReply(expectedNode)).when(mockActorContext).executeOperationAsync( eq(actorSelection(txActorRef)), eqSerializedReadData(YangInstanceIdentifier.builder().build()));