opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/TransactionProxyTest.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertFalse;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.junit.Assert.assertThrows;
  14 import static org.junit.Assert.assertTrue;
  15 import static org.mockito.ArgumentMatchers.any;
  16 import static org.mockito.ArgumentMatchers.anyString;
  17 import static org.mockito.ArgumentMatchers.eq;
  18 import static org.mockito.ArgumentMatchers.isA;
  19 import static org.mockito.Mockito.doReturn;
  20 import static org.mockito.Mockito.doThrow;
  21 import static org.mockito.Mockito.mock;
  22 import static org.mockito.Mockito.verify;
  23 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_ONLY;
  24 import static org.opendaylight.controller.cluster.datastore.TransactionType.READ_WRITE;
  25 import static org.opendaylight.controller.cluster.datastore.TransactionType.WRITE_ONLY;
  26
  27 import akka.actor.ActorRef;
  28 import akka.actor.ActorSelection;
  29 import akka.actor.ActorSystem;
  30 import akka.actor.Props;
  31 import akka.dispatch.Futures;
  32 import akka.util.Timeout;
  33 import com.google.common.base.Throwables;
  34 import com.google.common.collect.ImmutableSortedSet;
  35 import com.google.common.collect.Sets;
  36 import com.google.common.util.concurrent.FluentFuture;
  37 import com.google.common.util.concurrent.FutureCallback;
  38 import com.google.common.util.concurrent.MoreExecutors;
  39 import com.google.common.util.concurrent.Uninterruptibles;
  40 import java.util.Collection;
  41 import java.util.List;
  42 import java.util.Optional;
  43 import java.util.SortedSet;
  44 import java.util.concurrent.CountDownLatch;
  45 import java.util.concurrent.ExecutionException;
  46 import java.util.concurrent.TimeUnit;
  47 import java.util.concurrent.atomic.AtomicReference;
  48 import org.junit.Assert;
  49 import org.junit.Test;
  50 import org.mockito.ArgumentCaptor;
  51 import org.mockito.InOrder;
  52 import org.mockito.Mockito;
  53 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  54 import org.opendaylight.controller.cluster.datastore.config.Configuration;
  55 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  56 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  57 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  58 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
  59 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
  60 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
  61 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  62 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  63 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  64 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
  65 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
  66 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
  67 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
  68 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
  69 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregatorTest;
  70 import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
  71 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  72 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  73 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  74 import org.opendaylight.mdsal.common.api.ReadFailedException;
  75 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  76 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  77 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  78 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  79 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  80 import org.opendaylight.yangtools.yang.data.tree.api.DataTree;
  81 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
  82 import org.opendaylight.yangtools.yang.data.tree.api.DataTreeSnapshot;
  83 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  84 import scala.concurrent.Promise;
  85
  86 @SuppressWarnings({"resource", "checkstyle:IllegalThrows", "checkstyle:AvoidHidingCauseException"})
  87 public class TransactionProxyTest extends AbstractTransactionProxyTest {
  88
  89     @SuppressWarnings("serial")
  90     static class TestException extends RuntimeException {
  91     }
  92
  93     interface Invoker {
  94         FluentFuture<?> invoke(TransactionProxy proxy);
  95     }
  96
  97     @Test
  98     public void testRead() throws Exception {
  99         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 100
 101         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 102
 103         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 104                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 105
 106         assertEquals(Optional.empty(), transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS));
 107
 108         NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 109
 110         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 111                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 112
 113         assertEquals(Optional.of(expectedNode), transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS));
 114     }
 115
 116     @Test(expected = ReadFailedException.class)
 117     public void testReadWithInvalidReplyMessageType() throws Throwable {
 118         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 119
 120         doReturn(Futures.successful(new Object())).when(mockActorContext)
 121                 .executeOperationAsync(eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 122
 123         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 124
 125         try {
 126             transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 127         } catch (ExecutionException e) {
 128             throw e.getCause();
 129         }
 130     }
 131
 132     @Test(expected = TestException.class)
 133     public void testReadWithAsyncRemoteOperatonFailure() throws Throwable {
 134         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 135
 136         doReturn(Futures.failed(new TestException())).when(mockActorContext)
 137                 .executeOperationAsync(eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 138
 139         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 140
 141         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 142     }
 143
 144     private void testExceptionOnInitialCreateTransaction(final Exception exToThrow, final Invoker invoker)
 145             throws Throwable {
 146         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 147
 148         if (exToThrow instanceof PrimaryNotFoundException) {
 149             doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 150         } else {
 151             doReturn(primaryShardInfoReply(getSystem(), actorRef)).when(mockActorContext)
 152                     .findPrimaryShardAsync(anyString());
 153         }
 154
 155         doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
 156                 any(ActorSelection.class), any(), any(Timeout.class));
 157
 158         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 159
 160         propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
 161     }
 162
 163     private void testReadWithExceptionOnInitialCreateTransaction(final Exception exToThrow) throws Throwable {
 164         testExceptionOnInitialCreateTransaction(exToThrow, proxy -> proxy.read(TestModel.TEST_PATH));
 165     }
 166
 167     @Test(expected = PrimaryNotFoundException.class)
 168     public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 169         testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
 170     }
 171
 172     @Test(expected = TestException.class)
 173     public void testReadWhenATimeoutExceptionIsThrown() throws Throwable {
 174         testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
 175                 new TestException()));
 176     }
 177
 178     @Test(expected = TestException.class)
 179     public void testReadWhenAnyOtherExceptionIsThrown() throws Throwable {
 180         testReadWithExceptionOnInitialCreateTransaction(new TestException());
 181     }
 182
 183     @Test
 184     public void testReadWithPriorRecordingOperationSuccessful() throws Exception {
 185         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 186
 187         NormalizedNode expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 188
 189         expectBatchedModifications(actorRef, 1);
 190
 191         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
 192                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 193
 194         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 195
 196         transactionProxy.write(TestModel.TEST_PATH, expectedNode);
 197
 198         assertEquals(Optional.of(expectedNode), transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS));
 199
 200         InOrder inOrder = Mockito.inOrder(mockActorContext);
 201         inOrder.verify(mockActorContext).executeOperationAsync(
 202                 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
 203
 204         inOrder.verify(mockActorContext).executeOperationAsync(
 205                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 206     }
 207
 208     @Test
 209     public void testReadPreConditionCheck() {
 210         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 211         assertThrows(IllegalStateException.class, () -> transactionProxy.read(TestModel.TEST_PATH));
 212     }
 213
 214     @Test(expected = IllegalArgumentException.class)
 215     public void testInvalidCreateTransactionReply() throws Throwable {
 216         ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 217
 218         doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext)
 219                 .actorSelection(actorRef.path().toString());
 220
 221         doReturn(primaryShardInfoReply(getSystem(), actorRef)).when(mockActorContext)
 222                 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 223
 224         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
 225             eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY),
 226             any(Timeout.class));
 227
 228         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 229
 230         propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
 231     }
 232
 233     @Test
 234     public void testExists() throws Exception {
 235         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 236
 237         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 238
 239         doReturn(dataExistsReply(false)).when(mockActorContext).executeOperationAsync(
 240                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 241
 242         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).get();
 243
 244         assertEquals("Exists response", Boolean.FALSE, exists);
 245
 246         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
 247                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 248
 249         exists = transactionProxy.exists(TestModel.TEST_PATH).get();
 250
 251         assertEquals("Exists response", Boolean.TRUE, exists);
 252     }
 253
 254     @Test(expected = PrimaryNotFoundException.class)
 255     public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Throwable {
 256         testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"),
 257             proxy -> proxy.exists(TestModel.TEST_PATH));
 258     }
 259
 260     @Test(expected = ReadFailedException.class)
 261     public void testExistsWithInvalidReplyMessageType() throws Throwable {
 262         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 263
 264         doReturn(Futures.successful(new Object())).when(mockActorContext)
 265                 .executeOperationAsync(eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 266
 267         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 268
 269         try {
 270             transactionProxy.exists(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 271         } catch (ExecutionException e) {
 272             throw e.getCause();
 273         }
 274     }
 275
 276     @Test(expected = TestException.class)
 277     public void testExistsWithAsyncRemoteOperatonFailure() throws Throwable {
 278         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 279
 280         doReturn(Futures.failed(new TestException())).when(mockActorContext)
 281                 .executeOperationAsync(eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 282
 283         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 284
 285         propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
 286     }
 287
 288     @Test
 289     public void testExistsWithPriorRecordingOperationSuccessful() throws Exception {
 290         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 291
 292         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 293
 294         expectBatchedModifications(actorRef, 1);
 295
 296         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
 297                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 298
 299         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 300
 301         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 302
 303         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).get();
 304
 305         assertEquals("Exists response", Boolean.TRUE, exists);
 306
 307         InOrder inOrder = Mockito.inOrder(mockActorContext);
 308         inOrder.verify(mockActorContext).executeOperationAsync(
 309                 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
 310
 311         inOrder.verify(mockActorContext).executeOperationAsync(
 312                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
 313     }
 314
 315     @Test(expected = IllegalStateException.class)
 316     public void testExistsPreConditionCheck() {
 317         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 318         transactionProxy.exists(TestModel.TEST_PATH);
 319     }
 320
 321     @Test
 322     public void testWrite() {
 323         dataStoreContextBuilder.shardBatchedModificationCount(1);
 324         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 325
 326         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 327
 328         expectBatchedModifications(actorRef, 1);
 329
 330         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 331
 332         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 333
 334         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false);
 335     }
 336
 337     @Test
 338     @SuppressWarnings("checkstyle:IllegalCatch")
 339     public void testWriteAfterAsyncRead() throws Exception {
 340         ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem(),
 341                 DefaultShardStrategy.DEFAULT_SHARD);
 342
 343         Promise<Object> createTxPromise = akka.dispatch.Futures.promise();
 344         doReturn(createTxPromise).when(mockActorContext).executeOperationAsync(
 345                 eq(getSystem().actorSelection(actorRef.path())),
 346                 eqCreateTransaction(memberName, READ_WRITE), any(Timeout.class));
 347
 348         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 349                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 350
 351         expectBatchedModificationsReady(actorRef);
 352
 353         final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 354
 355         final TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 356
 357         final CountDownLatch readComplete = new CountDownLatch(1);
 358         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 359         com.google.common.util.concurrent.Futures.addCallback(transactionProxy.read(TestModel.TEST_PATH),
 360                 new  FutureCallback<Optional<NormalizedNode>>() {
 361                     @Override
 362                     public void onSuccess(final Optional<NormalizedNode> result) {
 363                         try {
 364                             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 365                         } catch (Exception e) {
 366                             caughtEx.set(e);
 367                         } finally {
 368                             readComplete.countDown();
 369                         }
 370                     }
 371
 372                     @Override
 373                     public void onFailure(final Throwable failure) {
 374                         caughtEx.set(failure);
 375                         readComplete.countDown();
 376                     }
 377                 }, MoreExecutors.directExecutor());
 378
 379         createTxPromise.success(createTransactionReply(actorRef, DataStoreVersions.CURRENT_VERSION));
 380
 381         Uninterruptibles.awaitUninterruptibly(readComplete, 5, TimeUnit.SECONDS);
 382
 383         final Throwable t = caughtEx.get();
 384         if (t != null) {
 385             Throwables.propagateIfPossible(t, Exception.class);
 386             throw new RuntimeException(t);
 387         }
 388
 389         // This sends the batched modification.
 390         transactionProxy.ready();
 391
 392         verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), true);
 393     }
 394
 395     @Test(expected = IllegalStateException.class)
 396     public void testWritePreConditionCheck() {
 397         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 398         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 399     }
 400
 401     @Test(expected = IllegalStateException.class)
 402     public void testWriteAfterReadyPreConditionCheck() {
 403         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 404
 405         transactionProxy.ready();
 406
 407         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 408     }
 409
 410     @Test
 411     public void testMerge() {
 412         dataStoreContextBuilder.shardBatchedModificationCount(1);
 413         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 414
 415         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 416
 417         expectBatchedModifications(actorRef, 1);
 418
 419         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 420
 421         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 422
 423         verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite), false);
 424     }
 425
 426     @Test
 427     public void testDelete() {
 428         dataStoreContextBuilder.shardBatchedModificationCount(1);
 429         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 430
 431         expectBatchedModifications(actorRef, 1);
 432
 433         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 434
 435         transactionProxy.delete(TestModel.TEST_PATH);
 436
 437         verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH), false);
 438     }
 439
 440     @Test
 441     public void testReadWrite() {
 442         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 443
 444         final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 445
 446         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 447                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 448
 449         expectBatchedModifications(actorRef, 1);
 450
 451         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 452
 453         transactionProxy.read(TestModel.TEST_PATH);
 454
 455         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 456
 457         transactionProxy.read(TestModel.TEST_PATH);
 458
 459         transactionProxy.read(TestModel.TEST_PATH);
 460
 461         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 462         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 463
 464         verifyBatchedModifications(batchedModifications.get(0), false,
 465                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 466     }
 467
 468     @Test
 469     public void testReadyWithReadWrite() {
 470         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 471
 472         final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 473
 474         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 475                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 476
 477         expectBatchedModificationsReady(actorRef, true);
 478
 479         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 480
 481         transactionProxy.read(TestModel.TEST_PATH);
 482
 483         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 484
 485         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 486
 487         assertTrue(ready instanceof SingleCommitCohortProxy);
 488
 489         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 490
 491         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 492         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 493
 494         verifyBatchedModifications(batchedModifications.get(0), true, true,
 495                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 496
 497         assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent());
 498     }
 499
 500     @Test
 501     public void testReadyWithNoModifications() {
 502         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 503
 504         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 505                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 506
 507         expectBatchedModificationsReady(actorRef, true);
 508
 509         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 510
 511         transactionProxy.read(TestModel.TEST_PATH);
 512
 513         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 514
 515         assertTrue(ready instanceof SingleCommitCohortProxy);
 516
 517         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 518
 519         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 520         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 521
 522         verifyBatchedModifications(batchedModifications.get(0), true, true);
 523     }
 524
 525     @Test
 526     public void testReadyWithMultipleShardWrites() {
 527         ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 528
 529         ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY,
 530                 TestModel.JUNK_QNAME.getLocalName());
 531
 532         expectBatchedModificationsReady(actorRef1);
 533         expectBatchedModificationsReady(actorRef2);
 534
 535         ActorRef actorRef3 = getSystem().actorOf(Props.create(DoNothingActor.class));
 536
 537         doReturn(getSystem().actorSelection(actorRef3.path())).when(mockActorContext)
 538                 .actorSelection(actorRef3.path().toString());
 539
 540         doReturn(Futures.successful(newPrimaryShardInfo(actorRef3, createDataTree()))).when(mockActorContext)
 541                 .findPrimaryShardAsync(eq(CarsModel.BASE_QNAME.getLocalName()));
 542
 543         expectReadyLocalTransaction(actorRef3, false);
 544
 545         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 546
 547         transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 548         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 549         transactionProxy.write(CarsModel.BASE_PATH, ImmutableNodes.containerNode(CarsModel.BASE_QNAME));
 550
 551         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 552
 553         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 554
 555         verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef1),
 556                 actorSelection(actorRef2), actorSelection(actorRef3));
 557
 558         SortedSet<String> expShardNames =
 559                 ImmutableSortedSet.of(DefaultShardStrategy.DEFAULT_SHARD,
 560                         TestModel.JUNK_QNAME.getLocalName(), CarsModel.BASE_QNAME.getLocalName());
 561
 562         ArgumentCaptor<BatchedModifications> batchedMods = ArgumentCaptor.forClass(BatchedModifications.class);
 563         verify(mockActorContext).executeOperationAsync(
 564                 eq(actorSelection(actorRef1)), batchedMods.capture(), any(Timeout.class));
 565         assertEquals("Participating shards", Optional.of(expShardNames),
 566             batchedMods.getValue().getParticipatingShardNames());
 567
 568         batchedMods = ArgumentCaptor.forClass(BatchedModifications.class);
 569         verify(mockActorContext).executeOperationAsync(
 570                 eq(actorSelection(actorRef2)), batchedMods.capture(), any(Timeout.class));
 571         assertEquals("Participating shards", Optional.of(expShardNames),
 572             batchedMods.getValue().getParticipatingShardNames());
 573
 574         ArgumentCaptor<ReadyLocalTransaction> readyLocalTx = ArgumentCaptor.forClass(ReadyLocalTransaction.class);
 575         verify(mockActorContext).executeOperationAsync(
 576                 eq(actorSelection(actorRef3)), readyLocalTx.capture(), any(Timeout.class));
 577         assertEquals("Participating shards", Optional.of(expShardNames),
 578             readyLocalTx.getValue().getParticipatingShardNames());
 579     }
 580
 581     @Test
 582     public void testReadyWithWriteOnlyAndLastBatchPending() {
 583         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 584
 585         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 586
 587         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 588
 589         expectBatchedModificationsReady(actorRef, true);
 590
 591         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 592
 593         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 594
 595         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 596
 597         assertTrue(ready instanceof SingleCommitCohortProxy);
 598
 599         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 600
 601         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 602         assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
 603
 604         verifyBatchedModifications(batchedModifications.get(0), true, true,
 605                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 606     }
 607
 608     @Test
 609     public void testReadyWithWriteOnlyAndLastBatchEmpty() {
 610         dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true);
 611         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 612
 613         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 614
 615         expectBatchedModificationsReady(actorRef, true);
 616
 617         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 618
 619         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 620
 621         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 622
 623         assertTrue(ready instanceof SingleCommitCohortProxy);
 624
 625         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 626
 627         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
 628         assertEquals("Captured BatchedModifications count", 2, batchedModifications.size());
 629
 630         verifyBatchedModifications(batchedModifications.get(0), false,
 631                 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
 632
 633         verifyBatchedModifications(batchedModifications.get(1), true, true);
 634     }
 635
 636     @Test
 637     public void testReadyWithReplyFailure() {
 638         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 639
 640         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 641
 642         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 643
 644         expectFailedBatchedModifications(actorRef);
 645
 646         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 647
 648         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 649
 650         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 651
 652         assertTrue(ready instanceof SingleCommitCohortProxy);
 653
 654         verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class);
 655     }
 656
 657     @Test
 658     public void testReadyWithDebugContextEnabled() {
 659         dataStoreContextBuilder.transactionDebugContextEnabled(true);
 660
 661         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 662
 663         expectBatchedModificationsReady(actorRef, true);
 664
 665         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 666
 667         transactionProxy.merge(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 668
 669         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 670
 671         assertTrue(ready instanceof DebugThreePhaseCommitCohort);
 672
 673         verifyCohortFutures((DebugThreePhaseCommitCohort)ready, new CommitTransactionReply().toSerializable());
 674     }
 675
 676     @Test
 677     public void testReadyWithLocalTransaction() {
 678         ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 679
 680         doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
 681                 .actorSelection(shardActorRef.path().toString());
 682
 683         doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, createDataTree()))).when(mockActorContext)
 684                 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 685
 686         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 687
 688         expectReadyLocalTransaction(shardActorRef, true);
 689
 690         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 691         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 692
 693         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 694         assertTrue(ready instanceof SingleCommitCohortProxy);
 695         verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
 696
 697         ArgumentCaptor<ReadyLocalTransaction> readyLocalTx = ArgumentCaptor.forClass(ReadyLocalTransaction.class);
 698         verify(mockActorContext).executeOperationAsync(
 699                 eq(actorSelection(shardActorRef)), readyLocalTx.capture(), any(Timeout.class));
 700         assertFalse("Participating shards present", readyLocalTx.getValue().getParticipatingShardNames().isPresent());
 701     }
 702
 703     @Test
 704     public void testReadyWithLocalTransactionWithFailure() {
 705         ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
 706
 707         doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
 708                 .actorSelection(shardActorRef.path().toString());
 709
 710         DataTree mockDataTree = createDataTree();
 711         DataTreeModification mockModification = mockDataTree.takeSnapshot().newModification();
 712         doThrow(new RuntimeException("mock")).when(mockModification).ready();
 713
 714         doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, mockDataTree))).when(mockActorContext)
 715                 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 716
 717         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 718
 719         expectReadyLocalTransaction(shardActorRef, true);
 720
 721         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 722         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 723
 724         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 725         assertTrue(ready instanceof SingleCommitCohortProxy);
 726         verifyCohortFutures((SingleCommitCohortProxy)ready, RuntimeException.class);
 727     }
 728
 729     private void testWriteOnlyTxWithFindPrimaryShardFailure(final Exception toThrow) {
 730         doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
 731
 732         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 733
 734         NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 735
 736         transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
 737
 738         transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 739
 740         transactionProxy.delete(TestModel.TEST_PATH);
 741
 742         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 743
 744         assertTrue(ready instanceof SingleCommitCohortProxy);
 745
 746         verifyCohortFutures((SingleCommitCohortProxy)ready, toThrow.getClass());
 747     }
 748
 749     @Test
 750     public void testWriteOnlyTxWithPrimaryNotFoundException() {
 751         testWriteOnlyTxWithFindPrimaryShardFailure(new PrimaryNotFoundException("mock"));
 752     }
 753
 754     @Test
 755     public void testWriteOnlyTxWithNotInitializedException() {
 756         testWriteOnlyTxWithFindPrimaryShardFailure(new NotInitializedException("mock"));
 757     }
 758
 759     @Test
 760     public void testWriteOnlyTxWithNoShardLeaderException() {
 761         testWriteOnlyTxWithFindPrimaryShardFailure(new NoShardLeaderException("mock"));
 762     }
 763
 764     @Test
 765     public void testReadyWithInvalidReplyMessageType() {
 766         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
 767         ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
 768
 769         ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY,
 770                 TestModel.JUNK_QNAME.getLocalName());
 771
 772         doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
 773                 eq(actorSelection(actorRef1)), isA(BatchedModifications.class), any(Timeout.class));
 774
 775         expectBatchedModificationsReady(actorRef2);
 776
 777         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
 778
 779         transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
 780         transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 781
 782         DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
 783
 784         assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
 785
 786         verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef2),
 787                 IllegalArgumentException.class);
 788     }
 789
 790     @Test
 791     public void testGetIdentifier() {
 792         setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
 793         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
 794
 795         Object id = transactionProxy.getIdentifier();
 796         assertNotNull("getIdentifier returned null", id);
 797         assertTrue("Invalid identifier: " + id, id.toString().contains(memberName));
 798     }
 799
 800     @Test
 801     public void testClose() {
 802         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
 803
 804         doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
 805                 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
 806
 807         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 808
 809         transactionProxy.read(TestModel.TEST_PATH);
 810
 811         transactionProxy.close();
 812
 813         verify(mockActorContext).sendOperationAsync(
 814                 eq(actorSelection(actorRef)), isA(CloseTransaction.class));
 815     }
 816
 817     private interface TransactionProxyOperation {
 818         void run(TransactionProxy transactionProxy);
 819     }
 820
 821     private PrimaryShardInfo newPrimaryShardInfo(final ActorRef actorRef) {
 822         return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION);
 823     }
 824
 825     private PrimaryShardInfo newPrimaryShardInfo(final ActorRef actorRef, final DataTree dataTree) {
 826         return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION,
 827                 dataTree);
 828     }
 829
 830     private void throttleOperation(final TransactionProxyOperation operation) {
 831         throttleOperation(operation, 1, true);
 832     }
 833
 834     private void throttleOperation(final TransactionProxyOperation operation, final int outstandingOpsLimit,
 835             final boolean shardFound) {
 836         throttleOperation(operation, outstandingOpsLimit, shardFound, TimeUnit.MILLISECONDS.toNanos(
 837                 mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()));
 838     }
 839
 840     private void throttleOperation(final TransactionProxyOperation operation, final int outstandingOpsLimit,
 841             final boolean shardFound, final long expectedCompletionTime) {
 842         ActorSystem actorSystem = getSystem();
 843         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 844
 845         // Note that we setting batchedModificationCount to one less than what we need because in TransactionProxy
 846         // we now allow one extra permit to be allowed for ready
 847         doReturn(dataStoreContextBuilder.operationTimeoutInSeconds(2)
 848                 .shardBatchedModificationCount(outstandingOpsLimit - 1).build()).when(mockActorContext)
 849                         .getDatastoreContext();
 850
 851         doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
 852                 .actorSelection(shardActorRef.path().toString());
 853
 854         if (shardFound) {
 855             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
 856                     .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 857             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
 858                     .findPrimaryShardAsync(eq("cars"));
 859
 860         } else {
 861             doReturn(Futures.failed(new Exception("not found")))
 862                     .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 863         }
 864
 865         doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
 866                 eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_WRITE),
 867                 any(Timeout.class));
 868
 869         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 870
 871         long start = System.nanoTime();
 872
 873         operation.run(transactionProxy);
 874
 875         long end = System.nanoTime();
 876
 877         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
 878                 expectedCompletionTime, end - start),
 879                 end - start > expectedCompletionTime && end - start < expectedCompletionTime * 2);
 880
 881     }
 882
 883     private void completeOperation(final TransactionProxyOperation operation) {
 884         completeOperation(operation, true);
 885     }
 886
 887     private void completeOperation(final TransactionProxyOperation operation, final boolean shardFound) {
 888         ActorSystem actorSystem = getSystem();
 889         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 890
 891         doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
 892                 .actorSelection(shardActorRef.path().toString());
 893
 894         if (shardFound) {
 895             doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
 896                     .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 897         } else {
 898             doReturn(Futures.failed(new PrimaryNotFoundException("test"))).when(mockActorContext)
 899                     .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 900         }
 901
 902         ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 903         String actorPath = txActorRef.path().toString();
 904         CreateTransactionReply createTransactionReply = new CreateTransactionReply(actorPath, nextTransactionId(),
 905                 DataStoreVersions.CURRENT_VERSION);
 906
 907         doReturn(actorSystem.actorSelection(actorPath)).when(mockActorContext).actorSelection(actorPath);
 908
 909         doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).executeOperationAsync(
 910                 eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_WRITE),
 911                 any(Timeout.class));
 912
 913         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 914
 915         long start = System.nanoTime();
 916
 917         operation.run(transactionProxy);
 918
 919         long end = System.nanoTime();
 920
 921         long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
 922                 .getOperationTimeoutInMillis());
 923         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
 924                 expected, end - start), end - start <= expected);
 925     }
 926
 927     private void completeOperationLocal(final TransactionProxyOperation operation, final DataTree dataTree) {
 928         ActorSystem actorSystem = getSystem();
 929         ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
 930
 931         doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
 932                 .actorSelection(shardActorRef.path().toString());
 933
 934         doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, dataTree))).when(mockActorContext)
 935                 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
 936
 937         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
 938
 939         long start = System.nanoTime();
 940
 941         operation.run(transactionProxy);
 942
 943         long end = System.nanoTime();
 944
 945         long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
 946                 .getOperationTimeoutInMillis());
 947         Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", expected, end - start),
 948                 end - start <= expected);
 949     }
 950
 951     private static DataTree createDataTree() {
 952         DataTree dataTree = mock(DataTree.class);
 953         DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
 954         DataTreeModification dataTreeModification = mock(DataTreeModification.class);
 955
 956         doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
 957         doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
 958
 959         return dataTree;
 960     }
 961
 962     private static DataTree createDataTree(final NormalizedNode readResponse) {
 963         DataTree dataTree = mock(DataTree.class);
 964         DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
 965         DataTreeModification dataTreeModification = mock(DataTreeModification.class);
 966
 967         doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
 968         doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
 969         doReturn(Optional.of(readResponse)).when(dataTreeModification).readNode(any(YangInstanceIdentifier.class));
 970
 971         return dataTree;
 972     }
 973
 974
 975     @Test
 976     public void testWriteCompletionForLocalShard() {
 977         completeOperationLocal(transactionProxy -> {
 978             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 979
 980             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 981
 982             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 983
 984         }, createDataTree());
 985     }
 986
 987     @Test
 988     public void testWriteThrottlingWhenShardFound() {
 989         throttleOperation(transactionProxy -> {
 990             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 991
 992             expectIncompleteBatchedModifications();
 993
 994             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 995
 996             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
 997         });
 998     }
 999
1000     @Test
1001     public void testWriteThrottlingWhenShardNotFound() {
1002         // Confirm that there is no throttling when the Shard is not found
1003         completeOperation(transactionProxy -> {
1004             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1005
1006             expectBatchedModifications(2);
1007
1008             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1009
1010             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1011         }, false);
1012
1013     }
1014
1015
1016     @Test
1017     public void testWriteCompletion() {
1018         completeOperation(transactionProxy -> {
1019             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1020
1021             expectBatchedModifications(2);
1022
1023             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1024
1025             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1026         });
1027     }
1028
1029     @Test
1030     public void testMergeThrottlingWhenShardFound() {
1031         throttleOperation(transactionProxy -> {
1032             NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1033
1034             expectIncompleteBatchedModifications();
1035
1036             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1037
1038             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1039         });
1040     }
1041
1042     @Test
1043     public void testMergeThrottlingWhenShardNotFound() {
1044         completeOperation(transactionProxy -> {
1045             NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1046
1047             expectBatchedModifications(2);
1048
1049             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1050
1051             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1052         }, false);
1053     }
1054
1055     @Test
1056     public void testMergeCompletion() {
1057         completeOperation(transactionProxy -> {
1058             NormalizedNode nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1059
1060             expectBatchedModifications(2);
1061
1062             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1063
1064             transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1065         });
1066
1067     }
1068
1069     @Test
1070     public void testMergeCompletionForLocalShard() {
1071         completeOperationLocal(transactionProxy -> {
1072             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1073
1074             transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1075
1076             transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1077
1078         }, createDataTree());
1079     }
1080
1081
1082     @Test
1083     public void testDeleteThrottlingWhenShardFound() {
1084
1085         throttleOperation(transactionProxy -> {
1086             expectIncompleteBatchedModifications();
1087
1088             transactionProxy.delete(TestModel.TEST_PATH);
1089
1090             transactionProxy.delete(TestModel.TEST_PATH);
1091         });
1092     }
1093
1094
1095     @Test
1096     public void testDeleteThrottlingWhenShardNotFound() {
1097
1098         completeOperation(transactionProxy -> {
1099             expectBatchedModifications(2);
1100
1101             transactionProxy.delete(TestModel.TEST_PATH);
1102
1103             transactionProxy.delete(TestModel.TEST_PATH);
1104         }, false);
1105     }
1106
1107     @Test
1108     public void testDeleteCompletionForLocalShard() {
1109         completeOperationLocal(transactionProxy -> {
1110
1111             transactionProxy.delete(TestModel.TEST_PATH);
1112
1113             transactionProxy.delete(TestModel.TEST_PATH);
1114         }, createDataTree());
1115
1116     }
1117
1118     @Test
1119     public void testDeleteCompletion() {
1120         completeOperation(transactionProxy -> {
1121             expectBatchedModifications(2);
1122
1123             transactionProxy.delete(TestModel.TEST_PATH);
1124
1125             transactionProxy.delete(TestModel.TEST_PATH);
1126         });
1127
1128     }
1129
1130     @Test
1131     public void testReadThrottlingWhenShardFound() {
1132
1133         throttleOperation(transactionProxy -> {
1134             doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1135                     any(ActorSelection.class), eqReadData());
1136
1137             transactionProxy.read(TestModel.TEST_PATH);
1138
1139             transactionProxy.read(TestModel.TEST_PATH);
1140         });
1141     }
1142
1143     @Test
1144     public void testReadThrottlingWhenShardNotFound() {
1145
1146         completeOperation(transactionProxy -> {
1147             doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1148                     any(ActorSelection.class), eqReadData());
1149
1150             transactionProxy.read(TestModel.TEST_PATH);
1151
1152             transactionProxy.read(TestModel.TEST_PATH);
1153         }, false);
1154     }
1155
1156
1157     @Test
1158     public void testReadCompletion() {
1159         completeOperation(transactionProxy -> {
1160             NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1161
1162             doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync(
1163                     any(ActorSelection.class), eqReadData(), any(Timeout.class));
1164
1165             transactionProxy.read(TestModel.TEST_PATH);
1166
1167             transactionProxy.read(TestModel.TEST_PATH);
1168         });
1169
1170     }
1171
1172     @Test
1173     public void testReadCompletionForLocalShard() {
1174         final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1175         completeOperationLocal(transactionProxy -> {
1176             transactionProxy.read(TestModel.TEST_PATH);
1177
1178             transactionProxy.read(TestModel.TEST_PATH);
1179         }, createDataTree(nodeToRead));
1180
1181     }
1182
1183     @Test
1184     public void testReadCompletionForLocalShardWhenExceptionOccurs() {
1185         completeOperationLocal(transactionProxy -> {
1186             transactionProxy.read(TestModel.TEST_PATH);
1187
1188             transactionProxy.read(TestModel.TEST_PATH);
1189         }, createDataTree());
1190
1191     }
1192
1193     @Test
1194     public void testExistsThrottlingWhenShardFound() {
1195
1196         throttleOperation(transactionProxy -> {
1197             doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1198                     any(ActorSelection.class), eqDataExists());
1199
1200             transactionProxy.exists(TestModel.TEST_PATH);
1201
1202             transactionProxy.exists(TestModel.TEST_PATH);
1203         });
1204     }
1205
1206     @Test
1207     public void testExistsThrottlingWhenShardNotFound() {
1208
1209         completeOperation(transactionProxy -> {
1210             doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1211                     any(ActorSelection.class), eqDataExists());
1212
1213             transactionProxy.exists(TestModel.TEST_PATH);
1214
1215             transactionProxy.exists(TestModel.TEST_PATH);
1216         }, false);
1217     }
1218
1219
1220     @Test
1221     public void testExistsCompletion() {
1222         completeOperation(transactionProxy -> {
1223             doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1224                     any(ActorSelection.class), eqDataExists(), any(Timeout.class));
1225
1226             transactionProxy.exists(TestModel.TEST_PATH);
1227
1228             transactionProxy.exists(TestModel.TEST_PATH);
1229         });
1230
1231     }
1232
1233     @Test
1234     public void testExistsCompletionForLocalShard() {
1235         final NormalizedNode nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1236         completeOperationLocal(transactionProxy -> {
1237             transactionProxy.exists(TestModel.TEST_PATH);
1238
1239             transactionProxy.exists(TestModel.TEST_PATH);
1240         }, createDataTree(nodeToRead));
1241
1242     }
1243
1244     @Test
1245     public void testExistsCompletionForLocalShardWhenExceptionOccurs() {
1246         completeOperationLocal(transactionProxy -> {
1247             transactionProxy.exists(TestModel.TEST_PATH);
1248
1249             transactionProxy.exists(TestModel.TEST_PATH);
1250         }, createDataTree());
1251
1252     }
1253
1254     @Test
1255     public void testReadyThrottling() {
1256
1257         throttleOperation(transactionProxy -> {
1258             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1259
1260             expectBatchedModifications(1);
1261
1262             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1263
1264             transactionProxy.ready();
1265         });
1266     }
1267
1268     @Test
1269     public void testReadyThrottlingWithTwoTransactionContexts() {
1270         throttleOperation(transactionProxy -> {
1271             NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1272             NormalizedNode carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME);
1273
1274             expectBatchedModifications(2);
1275
1276             transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1277
1278             // Trying to write to Cars will cause another transaction context to get created
1279             transactionProxy.write(CarsModel.BASE_PATH, carsNode);
1280
1281             // Now ready should block for both transaction contexts
1282             transactionProxy.ready();
1283         }, 1, true, TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
1284                 .getOperationTimeoutInMillis()) * 2);
1285     }
1286
1287     private void testModificationOperationBatching(final TransactionType type) {
1288         int shardBatchedModificationCount = 3;
1289         dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1290
1291         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), type);
1292
1293         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1294
1295         YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1296         NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1297
1298         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1299         NormalizedNode writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1300
1301         YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH;
1302         NormalizedNode writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1303
1304         YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1305         NormalizedNode mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1306
1307         YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH;
1308         NormalizedNode mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1309
1310         YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH;
1311         NormalizedNode mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1312
1313         YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH;
1314         YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH;
1315
1316         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, type);
1317
1318         transactionProxy.write(writePath1, writeNode1);
1319         transactionProxy.write(writePath2, writeNode2);
1320         transactionProxy.delete(deletePath1);
1321         transactionProxy.merge(mergePath1, mergeNode1);
1322         transactionProxy.merge(mergePath2, mergeNode2);
1323         transactionProxy.write(writePath3, writeNode3);
1324         transactionProxy.merge(mergePath3, mergeNode3);
1325         transactionProxy.delete(deletePath2);
1326
1327         // This sends the last batch.
1328         transactionProxy.ready();
1329
1330         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1331         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1332
1333         verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1334                 new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1));
1335
1336         verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1337                 new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3));
1338
1339         verifyBatchedModifications(batchedModifications.get(2), true, true,
1340                 new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2));
1341
1342         assertEquals("getTotalMessageCount", 3, batchedModifications.get(2).getTotalMessagesSent());
1343     }
1344
1345     @Test
1346     public void testReadWriteModificationOperationBatching() {
1347         testModificationOperationBatching(READ_WRITE);
1348     }
1349
1350     @Test
1351     public void testWriteOnlyModificationOperationBatching() {
1352         testModificationOperationBatching(WRITE_ONLY);
1353     }
1354
1355     @Test
1356     public void testOptimizedWriteOnlyModificationOperationBatching() {
1357         dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
1358         testModificationOperationBatching(WRITE_ONLY);
1359     }
1360
1361     @Test
1362     public void testModificationOperationBatchingWithInterleavedReads() throws Exception {
1363
1364         int shardBatchedModificationCount = 10;
1365         dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1366
1367         ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1368
1369         expectBatchedModifications(actorRef, shardBatchedModificationCount);
1370
1371         final YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1372         final NormalizedNode writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1373
1374         YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1375         NormalizedNode writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1376
1377         final YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1378         final NormalizedNode mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1379
1380         YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH;
1381         NormalizedNode mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1382
1383         final YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH;
1384
1385         doReturn(readDataReply(writeNode2)).when(mockActorContext).executeOperationAsync(
1386                 eq(actorSelection(actorRef)), eqReadData(writePath2), any(Timeout.class));
1387
1388         doReturn(readDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync(
1389                 eq(actorSelection(actorRef)), eqReadData(mergePath2), any(Timeout.class));
1390
1391         doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1392                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
1393
1394         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
1395
1396         transactionProxy.write(writePath1, writeNode1);
1397         transactionProxy.write(writePath2, writeNode2);
1398
1399         Optional<NormalizedNode> readOptional = transactionProxy.read(writePath2).get(5, TimeUnit.SECONDS);
1400
1401         assertEquals("Response NormalizedNode", Optional.of(writeNode2), readOptional);
1402
1403         transactionProxy.merge(mergePath1, mergeNode1);
1404         transactionProxy.merge(mergePath2, mergeNode2);
1405
1406         readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS);
1407
1408         transactionProxy.delete(deletePath);
1409
1410         Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).get();
1411         assertEquals("Exists response", Boolean.TRUE, exists);
1412
1413         assertEquals("Response NormalizedNode", Optional.of(mergeNode2), readOptional);
1414
1415         List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1416         assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1417
1418         verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1419                 new WriteModification(writePath2, writeNode2));
1420
1421         verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1422                 new MergeModification(mergePath2, mergeNode2));
1423
1424         verifyBatchedModifications(batchedModifications.get(2), false, new DeleteModification(deletePath));
1425
1426         InOrder inOrder = Mockito.inOrder(mockActorContext);
1427         inOrder.verify(mockActorContext).executeOperationAsync(
1428                 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1429
1430         inOrder.verify(mockActorContext).executeOperationAsync(
1431                 eq(actorSelection(actorRef)), eqReadData(writePath2), any(Timeout.class));
1432
1433         inOrder.verify(mockActorContext).executeOperationAsync(
1434                 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1435
1436         inOrder.verify(mockActorContext).executeOperationAsync(
1437                 eq(actorSelection(actorRef)), eqReadData(mergePath2), any(Timeout.class));
1438
1439         inOrder.verify(mockActorContext).executeOperationAsync(
1440                 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1441
1442         inOrder.verify(mockActorContext).executeOperationAsync(
1443                 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
1444     }
1445
1446     @Test
1447     public void testReadRoot() throws InterruptedException, ExecutionException, java.util.concurrent.TimeoutException {
1448         EffectiveModelContext schemaContext = SchemaContextHelper.full();
1449         Configuration configuration = mock(Configuration.class);
1450         doReturn(configuration).when(mockActorContext).getConfiguration();
1451         doReturn(schemaContext).when(mockActorContext).getSchemaContext();
1452         doReturn(Sets.newHashSet("test", "cars")).when(configuration).getAllShardNames();
1453
1454         NormalizedNode expectedNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1455         NormalizedNode expectedNode2 = ImmutableNodes.containerNode(CarsModel.CARS_QNAME);
1456
1457         setUpReadData("test", NormalizedNodeAggregatorTest.getRootNode(expectedNode1, schemaContext));
1458         setUpReadData("cars", NormalizedNodeAggregatorTest.getRootNode(expectedNode2, schemaContext));
1459
1460         doReturn(MemberName.forName(memberName)).when(mockActorContext).getCurrentMemberName();
1461
1462         doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher();
1463
1464         TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
1465
1466         Optional<NormalizedNode> readOptional = transactionProxy.read(
1467                 YangInstanceIdentifier.empty()).get(5, TimeUnit.SECONDS);
1468
1469         assertTrue("NormalizedNode isPresent", readOptional.isPresent());
1470
1471         NormalizedNode normalizedNode = readOptional.orElseThrow();
1472
1473         assertTrue("Expect value to be a Collection", normalizedNode.body() instanceof Collection);
1474
1475         @SuppressWarnings("unchecked")
1476         Collection<NormalizedNode> collection = (Collection<NormalizedNode>) normalizedNode.body();
1477
1478         for (NormalizedNode node : collection) {
1479             assertTrue("Expected " + node + " to be a ContainerNode", node instanceof ContainerNode);
1480         }
1481
1482         assertTrue("Child with QName = " + TestModel.TEST_QNAME + " not found",
1483                 NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME) != null);
1484
1485         assertEquals(expectedNode1, NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME));
1486
1487         assertTrue("Child with QName = " + CarsModel.BASE_QNAME + " not found",
1488                 NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME) != null);
1489
1490         assertEquals(expectedNode2, NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME));
1491     }
1492
1493
1494     private void setUpReadData(final String shardName, final NormalizedNode expectedNode) {
1495         ActorSystem actorSystem = getSystem();
1496         ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
1497
1498         doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
1499                 .actorSelection(shardActorRef.path().toString());
1500
1501         doReturn(primaryShardInfoReply(getSystem(), shardActorRef)).when(mockActorContext)
1502                 .findPrimaryShardAsync(eq(shardName));
1503
1504         ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1505
1506         doReturn(actorSystem.actorSelection(txActorRef.path())).when(mockActorContext)
1507                 .actorSelection(txActorRef.path().toString());
1508
1509         doReturn(Futures.successful(createTransactionReply(txActorRef, DataStoreVersions.CURRENT_VERSION)))
1510                 .when(mockActorContext).executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1511                         eqCreateTransaction(memberName, TransactionType.READ_ONLY), any(Timeout.class));
1512
1513         doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1514                 eq(actorSelection(txActorRef)), eqReadData(YangInstanceIdentifier.empty()), any(Timeout.class));
1515     }
1516 }