X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxyTest.java;h=265ec59f1cd324889627281850ffae9ed456c52a;hb=95115ca49f3b16b936e0f6c88aedfc17cd0ee92c;hp=abfe7eae22a15b69fd4dc1c71df46befe5059e31;hpb=e638203390314beb735c5d54f7936122e8d01d42;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java index abfe7eae22..265ec59f1c 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 @@ -6,11 +6,10 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; -import static org.mockito.Matchers.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.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType.READ_ONLY; @@ -21,78 +20,47 @@ 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.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import org.junit.AfterClass; import org.junit.Assert; -import org.junit.Before; -import org.junit.BeforeClass; import org.junit.Test; -import org.mockito.ArgumentCaptor; -import org.mockito.ArgumentMatcher; import org.mockito.InOrder; -import org.mockito.Mock; import org.mockito.Mockito; -import org.mockito.MockitoAnnotations; -import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.TransactionProxy.TransactionType; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; +import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException; import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException; import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications; import org.opendaylight.controller.cluster.datastore.messages.BatchedModificationsReply; import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction; -import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction; -import org.opendaylight.controller.cluster.datastore.messages.DataExists; -import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply; -import org.opendaylight.controller.cluster.datastore.messages.DeleteData; -import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply; -import org.opendaylight.controller.cluster.datastore.messages.MergeData; -import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply; -import org.opendaylight.controller.cluster.datastore.messages.ReadData; -import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply; import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction; -import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply; -import org.opendaylight.controller.cluster.datastore.messages.WriteData; -import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply; -import org.opendaylight.controller.cluster.datastore.modification.AbstractModification; import org.opendaylight.controller.cluster.datastore.modification.DeleteModification; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; -import org.opendaylight.controller.cluster.datastore.modification.Modification; import org.opendaylight.controller.cluster.datastore.modification.WriteModification; import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy; -import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory; -import org.opendaylight.controller.cluster.datastore.utils.ActorContext; 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.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.Promise; import scala.concurrent.duration.Duration; @SuppressWarnings("resource") -public class TransactionProxyTest { +public class TransactionProxyTest extends AbstractTransactionProxyTest { @SuppressWarnings("serial") static class TestException extends RuntimeException { @@ -102,291 +70,6 @@ public class TransactionProxyTest { CheckedFuture invoke(TransactionProxy proxy) throws Exception; } - private static ActorSystem system; - - private final Configuration configuration = new MockConfiguration(); - - @Mock - private ActorContext mockActorContext; - - private SchemaContext schemaContext; - - @Mock - private ClusterWrapper mockClusterWrapper; - - private final String memberName = "mock-member"; - - private final Builder dataStoreContextBuilder = DatastoreContext.newBuilder().operationTimeoutInSeconds(2). - shardBatchedModificationCount(1); - - @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(); - - 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(dataStoreContextBuilder.build()).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) { - 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 eqSerializedDataExists() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return DataExists.SERIALIZABLE_CLASS.equals(argument.getClass()) && - DataExists.fromSerializable(argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private DataExists eqDataExists() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return (argument instanceof DataExists) && - ((DataExists)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private ReadData eqSerializedReadData() { - return eqSerializedReadData(TestModel.TEST_PATH); - } - - private ReadData eqSerializedReadData(final YangInstanceIdentifier path) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return ReadData.SERIALIZABLE_CLASS.equals(argument.getClass()) && - ReadData.fromSerializable(argument).getPath().equals(path); - } - }; - - return argThat(matcher); - } - - private ReadData eqReadData() { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return (argument instanceof ReadData) && - ((ReadData)argument).getPath().equals(TestModel.TEST_PATH); - } - }; - - return argThat(matcher); - } - - private WriteData eqLegacyWriteData(final NormalizedNode nodeToWrite) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(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 MergeData eqLegacyMergeData(final NormalizedNode nodeToWrite) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - if(ShardTransactionMessages.MergeData.class.equals(argument.getClass())) { - MergeData obj = MergeData.fromSerializable(argument); - return obj.getPath().equals(TestModel.TEST_PATH) && obj.getData().equals(nodeToWrite); - } - - return false; - } - }; - - return argThat(matcher); - } - - private DeleteData eqLegacyDeleteData(final YangInstanceIdentifier expPath) { - ArgumentMatcher matcher = new ArgumentMatcher() { - @Override - public boolean matches(Object argument) { - return ShardTransactionMessages.DeleteData.class.equals(argument.getClass()) && - DeleteData.fromSerializable(argument).getPath().equals(expPath); - } - }; - - return argThat(matcher); - } - - private Future readySerializedTxReply(String path) { - return Futures.successful((Object)new ReadyTransactionReply(path).toSerializable()); - } - - private Future readyTxReply(String path) { - return Futures.successful((Object)new ReadyTransactionReply(path)); - } - - private Future readSerializedDataReply(NormalizedNode data, - short transactionVersion) { - return Futures.successful(new ReadDataReply(data, transactionVersion).toSerializable()); - } - - private Future readSerializedDataReply(NormalizedNode data) { - return readSerializedDataReply(data, DataStoreVersions.CURRENT_VERSION); - } - - private Future readDataReply(NormalizedNode data) { - return Futures.successful(new ReadDataReply(data, DataStoreVersions.CURRENT_VERSION)); - } - - private Future dataExistsSerializedReply(boolean exists) { - return Futures.successful(new DataExistsReply(exists).toSerializable()); - } - - private Future dataExistsReply(boolean exists) { - return Futures.successful(new DataExistsReply(exists)); - } - - private Future batchedModificationsReply(int count) { - return Futures.successful(new BatchedModificationsReply(count)); - } - - private Future incompleteFuture(){ - return mock(Future.class); - } - - private ActorSelection actorSelection(ActorRef actorRef) { - return getSystem().actorSelection(actorRef.path()); - } - - private void expectBatchedModifications(ActorRef actorRef, int count) { - doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(BatchedModifications.class)); - } - - private void expectBatchedModifications(int count) { - doReturn(batchedModificationsReply(count)).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(BatchedModifications.class)); - } - - private void expectIncompleteBatchedModifications() { - doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(BatchedModifications.class)); - } - - private void expectReadyTransaction(ActorRef actorRef) { - doReturn(readySerializedTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(ReadyTransaction.SERIALIZABLE_CLASS)); - } - - private void expectFailedBatchedModifications(ActorRef actorRef) { - doReturn(Futures.failed(new TestException())).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), isA(BatchedModifications.class)); - } - - private CreateTransactionReply createTransactionReply(ActorRef actorRef, int transactionVersion){ - return CreateTransactionReply.newBuilder() - .setTransactionActorPath(actorRef.path().toString()) - .setTransactionId("txn-1") - .setMessageVersion(transactionVersion) - .build(); - } - - 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(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)); - - return actorRef; - } - - private ActorRef setupActorContextWithInitialCreateTransaction(ActorSystem actorSystem, TransactionType type) { - return setupActorContextWithInitialCreateTransaction(actorSystem, type, DataStoreVersions.CURRENT_VERSION); - } - - - private void propagateReadFailedExceptionCause(CheckedFuture future) - throws Throwable { - - try { - future.checkedGet(5, TimeUnit.SECONDS); - fail("Expected ReadFailedException"); - } catch(ReadFailedException e) { - throw e.getCause(); - } - } - @Test public void testRead() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); @@ -705,24 +388,18 @@ public class TransactionProxyTest { @Test public void testWrite() throws Exception { + dataStoreContextBuilder.shardBatchedModificationCount(1); ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); expectBatchedModifications(actorRef, 1); - expectReadyTransaction(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - // This sends the batched modification. - transactionProxy.ready(); - - verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false); } @Test @@ -777,7 +454,7 @@ public class TransactionProxyTest { // This sends the batched modification. transactionProxy.ready(); - verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite)); + verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false); verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), BatchedModificationsReply.class); @@ -800,73 +477,36 @@ public class TransactionProxyTest { @Test public void testMerge() throws Exception { + dataStoreContextBuilder.shardBatchedModificationCount(1); ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); expectBatchedModifications(actorRef, 1); - expectReadyTransaction(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite); - // This sends the batched modification. - transactionProxy.ready(); - - verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite)); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); + verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite), false); } @Test public void testDelete() throws Exception { + dataStoreContextBuilder.shardBatchedModificationCount(1); ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); expectBatchedModifications(actorRef, 1); - expectReadyTransaction(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); transactionProxy.delete(TestModel.TEST_PATH); - // This sends the batched modification. - transactionProxy.ready(); - - verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH)); - - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - } - - private void verifyCohortFutures(ThreePhaseCommitCohortProxy proxy, - Object... expReplies) throws Exception { - assertEquals("getReadyOperationFutures size", expReplies.length, - proxy.getCohortFutures().size()); - - int i = 0; - for( Future future: proxy.getCohortFutures()) { - assertNotNull("Ready operation Future is null", future); - - Object expReply = expReplies[i++]; - if(expReply instanceof ActorSelection) { - ActorSelection actual = Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - assertEquals("Cohort actor path", expReply, actual); - } else { - // Expecting exception. - try { - Await.result(future, Duration.create(5, TimeUnit.SECONDS)); - fail("Expected exception from ready operation Future"); - } catch(Exception e) { - // Expected - } - } - } + verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH), false); } @Test - public void testReady() throws Exception { + public void testReadyWithReadWrite() throws Exception { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); @@ -896,44 +536,57 @@ public class TransactionProxyTest { verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), isA(BatchedModifications.class)); + + verify(mockActorContext).executeOperationAsync(eq(actorSelection(actorRef)), + isA(ReadyTransaction.SERIALIZABLE_CLASS)); } - private ActorRef testCompatibilityWithHeliumVersion(short version) throws Exception { - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), - READ_WRITE, version); + @Test + public void testReadyWithWriteOnlyAndLastBatchPending() throws Exception { + dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); - NormalizedNode testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); - doReturn(readSerializedDataReply(testNode, version)).when(mockActorContext).executeOperationAsync( - eq(actorSelection(actorRef)), eqSerializedReadData()); + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - doReturn(Futures.successful(new WriteDataReply().toSerializable(version))).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyWriteData(testNode)); + expectBatchedModificationsReady(actorRef, 1); - doReturn(Futures.successful(new MergeDataReply().toSerializable(version))).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyMergeData(testNode)); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); - doReturn(Futures.successful(new DeleteDataReply().toSerializable(version))).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), eqLegacyDeleteData(TestModel.TEST_PATH)); + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - expectReadyTransaction(actorRef); + DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); + + assertTrue(ready instanceof ThreePhaseCommitCohortProxy); - doReturn(actorRef.path().toString()).when(mockActorContext).resolvePath(eq(actorRef.path().toString()), - eq(actorRef.path().toString())); + ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures()); - Optional> readOptional = transactionProxy.read(TestModel.TEST_PATH). - get(5, TimeUnit.SECONDS); + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); - assertEquals("NormalizedNode isPresent", true, readOptional.isPresent()); - assertEquals("Response NormalizedNode", testNode, readOptional.get()); + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - transactionProxy.write(TestModel.TEST_PATH, testNode); + verifyBatchedModifications(batchedModifications.get(0), true, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - transactionProxy.merge(TestModel.TEST_PATH, testNode); + verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), + isA(ReadyTransaction.SERIALIZABLE_CLASS)); + } - transactionProxy.delete(TestModel.TEST_PATH); + @Test + public void testReadyWithWriteOnlyAndLastBatchEmpty() throws Exception { + dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); + + NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + + expectBatchedModificationsReady(actorRef, 1); + + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + + transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); @@ -942,40 +595,32 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - ShardTransactionMessages.WriteDataReply.class, ShardTransactionMessages.MergeDataReply.class, - ShardTransactionMessages.DeleteDataReply.class); + BatchedModificationsReply.class); verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); - return actorRef; - } - - @Test - public void testCompatibilityWithBaseHeliumVersion() throws Exception { - ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.BASE_HELIUM_VERSION); + List batchedModifications = captureBatchedModifications(actorRef); + assertEquals("Captured BatchedModifications count", 2, batchedModifications.size()); - verify(mockActorContext).resolvePath(eq(actorRef.path().toString()), - eq(actorRef.path().toString())); - } + verifyBatchedModifications(batchedModifications.get(0), false, + new WriteModification(TestModel.TEST_PATH, nodeToWrite)); - @Test - public void testCompatibilityWithHeliumR1Version() throws Exception { - ActorRef actorRef = testCompatibilityWithHeliumVersion(DataStoreVersions.HELIUM_1_VERSION); + verifyBatchedModifications(batchedModifications.get(1), true); - verify(mockActorContext, Mockito.never()).resolvePath(eq(actorRef.path().toString()), - eq(actorRef.path().toString())); + verify(mockActorContext, never()).executeOperationAsync(eq(actorSelection(actorRef)), + isA(ReadyTransaction.SERIALIZABLE_CLASS)); } @Test public void testReadyWithRecordingOperationFailure() throws Exception { + dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); expectFailedBatchedModifications(actorRef); - expectReadyTransaction(actorRef); - doReturn(false).when(mockActorContext).isPathLocal(actorRef.path().toString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -995,15 +640,13 @@ public class TransactionProxyTest { @Test public void testReadyWithReplyFailure() throws Exception { + dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModifications(actorRef, 1); - - doReturn(Futures.failed(new TestException())).when(mockActorContext). - executeOperationAsync(eq(actorSelection(actorRef)), - isA(ReadyTransaction.SERIALIZABLE_CLASS)); + expectFailedBatchedModifications(actorRef); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -1015,17 +658,11 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - verifyCohortFutures(proxy, TestException.class); } - @Test - public void testReadyWithInitialCreateTransactionFailure() throws Exception { - - doReturn(Futures.failed(new PrimaryNotFoundException("mock"))).when( - mockActorContext).findPrimaryShardAsync(anyString()); + private void testWriteOnlyTxWithFindPrimaryShardFailure(Exception toThrow) throws Exception { + doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -1043,20 +680,36 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortFutures(proxy, PrimaryNotFoundException.class); + verifyCohortFutures(proxy, toThrow.getClass()); + } + + @Test + public void testWriteOnlyTxWithPrimaryNotFoundException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new PrimaryNotFoundException("mock")); + } + + @Test + public void testWriteOnlyTxWithNotInitializedException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new NotInitializedException("mock")); + } + + @Test + public void testWriteOnlyTxWithNoShardLeaderException() throws Exception { + testWriteOnlyTxWithFindPrimaryShardFailure(new NoShardLeaderException("mock")); } @Test public void testReadyWithInvalidReplyMessageType() throws Exception { + dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); - expectBatchedModifications(actorRef, 1); + //expectBatchedModifications(actorRef, 1); doReturn(Futures.successful(new Object())).when(mockActorContext). executeOperationAsync(eq(actorSelection(actorRef)), - isA(ReadyTransaction.SERIALIZABLE_CLASS)); + isA(BatchedModifications.class)); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); @@ -1071,17 +724,6 @@ public class TransactionProxyTest { verifyCohortFutures(proxy, IllegalArgumentException.class); } - @Test - public void testUnusedTransaction() throws Exception { - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); - - DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); - - assertEquals("canCommit", true, ready.canCommit().get()); - ready.preCommit().get(); - ready.commit().get(); - } - @Test public void testGetIdentifier() { setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); @@ -1125,24 +767,8 @@ public class TransactionProxyTest { */ @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); + setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY); + doReturn(true).when(mockActorContext).isPathLocal(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext,READ_ONLY); @@ -1178,40 +804,20 @@ public class TransactionProxyTest { @Test public void testLocalTxActorReady() 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). - setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); - - doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). - executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), - eqCreateTransaction(memberName, WRITE_ONLY)); - - doReturn(true).when(mockActorContext).isPathLocal(actorPath); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + doReturn(true).when(mockActorContext).isPathLocal(anyString()); doReturn(batchedModificationsReply(1)).when(mockActorContext).executeOperationAsync( any(ActorSelection.class), isA(BatchedModifications.class)); - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, WRITE_ONLY); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); transactionProxy.write(TestModel.TEST_PATH, nodeToWrite); - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class); - // testing ready - doReturn(readyTxReply(shardActorRef.path().toString())).when(mockActorContext).executeOperationAsync( - any(ActorSelection.class), isA(ReadyTransaction.class)); + doReturn(readyTxReply(actorRef.path().toString())).when(mockActorContext).executeOperationAsync( + eq(actorSelection(actorRef)), isA(ReadyTransaction.class)); DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready(); @@ -1219,7 +825,7 @@ public class TransactionProxyTest { ThreePhaseCommitCohortProxy proxy = (ThreePhaseCommitCohortProxy) ready; - verifyCohortFutures(proxy, getSystem().actorSelection(shardActorRef.path())); + verifyCohortFutures(proxy, getSystem().actorSelection(actorRef.path())); } private static interface TransactionProxyOperation { @@ -1289,20 +895,23 @@ public class TransactionProxyTest { doReturn(Futures.successful(actorSystem.actorSelection(shardActorRef.path()))). when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } else { - doReturn(Futures.failed(new Exception("not found"))) + doReturn(Futures.failed(new PrimaryNotFoundException("test"))) .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } - String actorPath = "akka.tcp://system@127.0.0.1:2550/user/tx-actor"; + ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class)); + String actorPath = txActorRef.path().toString(); CreateTransactionReply createTransactionReply = CreateTransactionReply.newBuilder(). setTransactionId("txn-1").setTransactionActorPath(actorPath). setMessageVersion(DataStoreVersions.CURRENT_VERSION).build(); + doReturn(actorSystem.actorSelection(actorPath)).when(mockActorContext).actorSelection(actorPath); + doReturn(Futures.successful(createTransactionReply)).when(mockActorContext). executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_WRITE)); - doReturn(true).when(mockActorContext).isPathLocal(actorPath); + doReturn(true).when(mockActorContext).isPathLocal(anyString()); TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); @@ -1335,6 +944,7 @@ public class TransactionProxyTest { @Test public void testWriteThrottlingWhenShardFound(){ + dataStoreContextBuilder.shardBatchedModificationCount(1); throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1352,6 +962,7 @@ public class TransactionProxyTest { @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) { @@ -1370,6 +981,7 @@ public class TransactionProxyTest { @Test public void testWriteCompletion(){ + dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1386,7 +998,7 @@ public class TransactionProxyTest { @Test public void testMergeThrottlingWhenShardFound(){ - + dataStoreContextBuilder.shardBatchedModificationCount(1); throttleOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1403,7 +1015,7 @@ public class TransactionProxyTest { @Test public void testMergeThrottlingWhenShardNotFound(){ - + dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1420,6 +1032,7 @@ public class TransactionProxyTest { @Test public void testMergeCompletion(){ + dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1468,6 +1081,7 @@ public class TransactionProxyTest { @Test public void testDeleteCompletion(){ + dataStoreContextBuilder.shardBatchedModificationCount(1); completeOperation(new TransactionProxyOperation() { @Override public void run(TransactionProxy transactionProxy) { @@ -1624,13 +1238,11 @@ public class TransactionProxyTest { }, 2, true); } - @Test - public void testModificationOperationBatching() throws Throwable { + private void testModificationOperationBatching(TransactionType type) throws Exception { int shardBatchedModificationCount = 3; - doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()). - when(mockActorContext).getDatastoreContext(); + dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount); - ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); + ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), type); expectBatchedModifications(actorRef, shardBatchedModificationCount); @@ -1657,7 +1269,7 @@ public class TransactionProxyTest { YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH; YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH; - TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, READ_WRITE); + TransactionProxy transactionProxy = new TransactionProxy(mockActorContext, type); transactionProxy.write(writePath1, writeNode1); transactionProxy.write(writePath2, writeNode2); @@ -1674,24 +1286,46 @@ public class TransactionProxyTest { List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 3, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1), + verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1), new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1)); - verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1), + verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1), new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3)); - verifyBatchedModifications(batchedModifications.get(2), new MergeModification(mergePath3, mergeNode3), + boolean optimizedWriteOnly = type == WRITE_ONLY && dataStoreContextBuilder.build().isWriteOnlyTransactionOptimizationsEnabled(); + verifyBatchedModifications(batchedModifications.get(2), optimizedWriteOnly, new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2)); - verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), - BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); + if(optimizedWriteOnly) { + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class, BatchedModificationsReply.class); + } else { + verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), + BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); + } + } + + @Test + public void testReadWriteModificationOperationBatching() throws Throwable { + testModificationOperationBatching(READ_WRITE); + } + + @Test + public void testWriteOnlyModificationOperationBatching() throws Throwable { + testModificationOperationBatching(WRITE_ONLY); + } + + @Test + public void testOptimizedWriteOnlyModificationOperationBatching() throws Throwable { + dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); + testModificationOperationBatching(WRITE_ONLY); } @Test public void testModificationOperationBatchingWithInterleavedReads() throws Throwable { + int shardBatchedModificationCount = 10; - doReturn(dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount).build()). - when(mockActorContext).getDatastoreContext(); + dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount); ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); @@ -1747,13 +1381,13 @@ public class TransactionProxyTest { List batchedModifications = captureBatchedModifications(actorRef); assertEquals("Captured BatchedModifications count", 3, batchedModifications.size()); - verifyBatchedModifications(batchedModifications.get(0), new WriteModification(writePath1, writeNode1), + verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1), new WriteModification(writePath2, writeNode2)); - verifyBatchedModifications(batchedModifications.get(1), new MergeModification(mergePath1, mergeNode1), + verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1), new MergeModification(mergePath2, mergeNode2)); - verifyBatchedModifications(batchedModifications.get(2), new DeleteModification(deletePath)); + verifyBatchedModifications(batchedModifications.get(2), false, new DeleteModification(deletePath)); InOrder inOrder = Mockito.inOrder(mockActorContext); inOrder.verify(mockActorContext).executeOperationAsync( @@ -1777,49 +1411,4 @@ public class TransactionProxyTest { verifyRecordingOperationFutures(transactionProxy.getRecordedOperationFutures(), BatchedModificationsReply.class, BatchedModificationsReply.class, BatchedModificationsReply.class); } - - private List captureBatchedModifications(ActorRef actorRef) { - ArgumentCaptor batchedModificationsCaptor = - ArgumentCaptor.forClass(BatchedModifications.class); - verify(mockActorContext, Mockito.atLeastOnce()).executeOperationAsync( - eq(actorSelection(actorRef)), batchedModificationsCaptor.capture()); - - List batchedModifications = filterCaptured( - batchedModificationsCaptor, BatchedModifications.class); - return batchedModifications; - } - - private List filterCaptured(ArgumentCaptor captor, Class type) { - List captured = new ArrayList<>(); - for(T c: captor.getAllValues()) { - if(type.isInstance(c)) { - captured.add(c); - } - } - - return captured; - } - - private void verifyOneBatchedModification(ActorRef actorRef, Modification expected) { - List batchedModifications = captureBatchedModifications(actorRef); - assertEquals("Captured BatchedModifications count", 1, batchedModifications.size()); - - verifyBatchedModifications(batchedModifications.get(0), expected); - } - - private void verifyBatchedModifications(Object message, Modification... expected) { - assertEquals("Message type", BatchedModifications.class, message.getClass()); - BatchedModifications batchedModifications = (BatchedModifications)message; - assertEquals("BatchedModifications size", expected.length, batchedModifications.getModifications().size()); - for(int i = 0; i < batchedModifications.getModifications().size(); i++) { - Modification actual = batchedModifications.getModifications().get(i); - assertEquals("Modification type", expected[i].getClass(), actual.getClass()); - assertEquals("getPath", ((AbstractModification)expected[i]).getPath(), - ((AbstractModification)actual).getPath()); - if(actual instanceof WriteModification) { - assertEquals("getData", ((WriteModification)expected[i]).getData(), - ((WriteModification)actual).getData()); - } - } - } }