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