2 * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.datastore;
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;
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;
81 @SuppressWarnings("resource")
82 public class TransactionProxyTest extends AbstractTransactionProxyTest {
84 @SuppressWarnings("serial")
85 static class TestException extends RuntimeException {
89 CheckedFuture<?, ReadFailedException> invoke(TransactionProxy proxy) throws Exception;
93 public void testRead() throws Exception {
94 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
96 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
98 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
99 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
101 Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
102 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
104 assertEquals("NormalizedNode isPresent", false, readOptional.isPresent());
106 NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
108 doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
109 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
111 readOptional = transactionProxy.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
113 assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
115 assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
118 @Test(expected = ReadFailedException.class)
119 public void testReadWithInvalidReplyMessageType() throws Exception {
120 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
122 doReturn(Futures.successful(new Object())).when(mockActorContext)
123 .executeOperationAsync(eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
125 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
127 transactionProxy.read(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
130 @Test(expected = TestException.class)
131 public void testReadWithAsyncRemoteOperatonFailure() throws Exception {
132 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
134 doReturn(Futures.failed(new TestException())).when(mockActorContext)
135 .executeOperationAsync(eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
137 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
139 propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
142 private void testExceptionOnInitialCreateTransaction(final Exception exToThrow, final Invoker invoker)
144 ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
146 if (exToThrow instanceof PrimaryNotFoundException) {
147 doReturn(Futures.failed(exToThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
149 doReturn(primaryShardInfoReply(getSystem(), actorRef)).when(mockActorContext)
150 .findPrimaryShardAsync(anyString());
153 doReturn(Futures.failed(exToThrow)).when(mockActorContext).executeOperationAsync(
154 any(ActorSelection.class), any(), any(Timeout.class));
156 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
158 propagateReadFailedExceptionCause(invoker.invoke(transactionProxy));
161 private void testReadWithExceptionOnInitialCreateTransaction(final Exception exToThrow) throws Exception {
162 testExceptionOnInitialCreateTransaction(exToThrow, proxy -> proxy.read(TestModel.TEST_PATH));
165 @Test(expected = PrimaryNotFoundException.class)
166 public void testReadWhenAPrimaryNotFoundExceptionIsThrown() throws Exception {
167 testReadWithExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"));
170 @Test(expected = TimeoutException.class)
171 public void testReadWhenATimeoutExceptionIsThrown() throws Exception {
172 testReadWithExceptionOnInitialCreateTransaction(new TimeoutException("test",
173 new Exception("reason")));
176 @Test(expected = TestException.class)
177 public void testReadWhenAnyOtherExceptionIsThrown() throws Exception {
178 testReadWithExceptionOnInitialCreateTransaction(new TestException());
182 public void testReadWithPriorRecordingOperationSuccessful() throws Exception {
183 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
185 NormalizedNode<?, ?> expectedNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
187 expectBatchedModifications(actorRef, 1);
189 doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
190 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
192 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
194 transactionProxy.write(TestModel.TEST_PATH, expectedNode);
196 Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
197 TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
199 assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
200 assertEquals("Response NormalizedNode", expectedNode, readOptional.get());
202 InOrder inOrder = Mockito.inOrder(mockActorContext);
203 inOrder.verify(mockActorContext).executeOperationAsync(
204 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
206 inOrder.verify(mockActorContext).executeOperationAsync(
207 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
210 @Test(expected = IllegalStateException.class)
211 public void testReadPreConditionCheck() {
212 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
213 transactionProxy.read(TestModel.TEST_PATH);
216 @Test(expected = IllegalArgumentException.class)
217 public void testInvalidCreateTransactionReply() throws Exception {
218 ActorRef actorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
220 doReturn(getSystem().actorSelection(actorRef.path())).when(mockActorContext)
221 .actorSelection(actorRef.path().toString());
223 doReturn(primaryShardInfoReply(getSystem(), actorRef)).when(mockActorContext)
224 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
226 doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
227 eq(getSystem().actorSelection(actorRef.path())), eqCreateTransaction(memberName, READ_ONLY),
230 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
232 propagateReadFailedExceptionCause(transactionProxy.read(TestModel.TEST_PATH));
236 public void testExists() throws Exception {
237 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
239 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
241 doReturn(dataExistsReply(false)).when(mockActorContext).executeOperationAsync(
242 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
244 Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
246 assertEquals("Exists response", false, exists);
248 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
249 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
251 exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
253 assertEquals("Exists response", true, exists);
256 @Test(expected = PrimaryNotFoundException.class)
257 public void testExistsWhenAPrimaryNotFoundExceptionIsThrown() throws Exception {
258 testExceptionOnInitialCreateTransaction(new PrimaryNotFoundException("test"),
259 proxy -> proxy.exists(TestModel.TEST_PATH));
262 @Test(expected = ReadFailedException.class)
263 public void testExistsWithInvalidReplyMessageType() throws Exception {
264 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
266 doReturn(Futures.successful(new Object())).when(mockActorContext)
267 .executeOperationAsync(eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
269 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
271 transactionProxy.exists(TestModel.TEST_PATH).checkedGet(5, TimeUnit.SECONDS);
274 @Test(expected = TestException.class)
275 public void testExistsWithAsyncRemoteOperatonFailure() throws Exception {
276 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
278 doReturn(Futures.failed(new TestException())).when(mockActorContext)
279 .executeOperationAsync(eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
281 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
283 propagateReadFailedExceptionCause(transactionProxy.exists(TestModel.TEST_PATH));
287 public void testExistsWithPriorRecordingOperationSuccessful() throws Exception {
288 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
290 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
292 expectBatchedModifications(actorRef, 1);
294 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
295 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
297 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
299 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
301 Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
303 assertEquals("Exists response", true, exists);
305 InOrder inOrder = Mockito.inOrder(mockActorContext);
306 inOrder.verify(mockActorContext).executeOperationAsync(
307 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
309 inOrder.verify(mockActorContext).executeOperationAsync(
310 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
313 @Test(expected = IllegalStateException.class)
314 public void testExistsPreConditionCheck() {
315 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
316 transactionProxy.exists(TestModel.TEST_PATH);
320 public void testWrite() throws Exception {
321 dataStoreContextBuilder.shardBatchedModificationCount(1);
322 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
324 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
326 expectBatchedModifications(actorRef, 1);
328 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
330 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
332 verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), false);
336 @SuppressWarnings("checkstyle:IllegalCatch")
337 public void testWriteAfterAsyncRead() throws Exception {
338 ActorRef actorRef = setupActorContextWithoutInitialCreateTransaction(getSystem(),
339 DefaultShardStrategy.DEFAULT_SHARD);
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));
346 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
347 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
349 expectBatchedModificationsReady(actorRef);
351 final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
353 final TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
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<?, ?>>>() {
360 public void onSuccess(final Optional<NormalizedNode<?, ?>> result) {
362 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
363 } catch (Exception e) {
366 readComplete.countDown();
371 public void onFailure(final Throwable failure) {
372 caughtEx.set(failure);
373 readComplete.countDown();
375 }, MoreExecutors.directExecutor());
377 createTxPromise.success(createTransactionReply(actorRef, DataStoreVersions.CURRENT_VERSION));
379 Uninterruptibles.awaitUninterruptibly(readComplete, 5, TimeUnit.SECONDS);
381 if (caughtEx.get() != null) {
382 Throwables.propagateIfInstanceOf(caughtEx.get(), Exception.class);
383 Throwables.propagate(caughtEx.get());
386 // This sends the batched modification.
387 transactionProxy.ready();
389 verifyOneBatchedModification(actorRef, new WriteModification(TestModel.TEST_PATH, nodeToWrite), true);
392 @Test(expected = IllegalStateException.class)
393 public void testWritePreConditionCheck() {
394 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
395 transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
398 @Test(expected = IllegalStateException.class)
399 public void testWriteAfterReadyPreConditionCheck() {
400 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
402 transactionProxy.ready();
404 transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
408 public void testMerge() throws Exception {
409 dataStoreContextBuilder.shardBatchedModificationCount(1);
410 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
412 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
414 expectBatchedModifications(actorRef, 1);
416 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
418 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
420 verifyOneBatchedModification(actorRef, new MergeModification(TestModel.TEST_PATH, nodeToWrite), false);
424 public void testDelete() throws Exception {
425 dataStoreContextBuilder.shardBatchedModificationCount(1);
426 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
428 expectBatchedModifications(actorRef, 1);
430 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
432 transactionProxy.delete(TestModel.TEST_PATH);
434 verifyOneBatchedModification(actorRef, new DeleteModification(TestModel.TEST_PATH), false);
438 public void testReadWrite() throws Exception {
439 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
441 final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
443 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
444 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
446 expectBatchedModifications(actorRef, 1);
448 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
450 transactionProxy.read(TestModel.TEST_PATH);
452 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
454 transactionProxy.read(TestModel.TEST_PATH);
456 transactionProxy.read(TestModel.TEST_PATH);
458 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
459 assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
461 verifyBatchedModifications(batchedModifications.get(0), false,
462 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
466 public void testReadyWithReadWrite() throws Exception {
467 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
469 final NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
471 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
472 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
474 expectBatchedModificationsReady(actorRef, true);
476 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
478 transactionProxy.read(TestModel.TEST_PATH);
480 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
482 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
484 assertTrue(ready instanceof SingleCommitCohortProxy);
486 verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
488 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
489 assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
491 verifyBatchedModifications(batchedModifications.get(0), true, true,
492 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
494 assertEquals("getTotalMessageCount", 1, batchedModifications.get(0).getTotalMessagesSent());
498 public void testReadyWithNoModifications() throws Exception {
499 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
501 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
502 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
504 expectBatchedModificationsReady(actorRef, true);
506 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
508 transactionProxy.read(TestModel.TEST_PATH);
510 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
512 assertTrue(ready instanceof SingleCommitCohortProxy);
514 verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
516 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
517 assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
519 verifyBatchedModifications(batchedModifications.get(0), true, true);
523 public void testReadyWithMultipleShardWrites() throws Exception {
524 ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
526 ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk");
528 expectBatchedModificationsReady(actorRef1);
529 expectBatchedModificationsReady(actorRef2);
531 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
533 transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
534 transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
536 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
538 assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
540 verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef1),
541 actorSelection(actorRef2));
545 public void testReadyWithWriteOnlyAndLastBatchPending() throws Exception {
546 dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
548 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
550 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
552 expectBatchedModificationsReady(actorRef, true);
554 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
556 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
558 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
560 assertTrue(ready instanceof SingleCommitCohortProxy);
562 verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
564 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
565 assertEquals("Captured BatchedModifications count", 1, batchedModifications.size());
567 verifyBatchedModifications(batchedModifications.get(0), true, true,
568 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
572 public void testReadyWithWriteOnlyAndLastBatchEmpty() throws Exception {
573 dataStoreContextBuilder.shardBatchedModificationCount(1).writeOnlyTransactionOptimizationsEnabled(true);
574 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
576 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
578 expectBatchedModificationsReady(actorRef, true);
580 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
582 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
584 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
586 assertTrue(ready instanceof SingleCommitCohortProxy);
588 verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
590 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
591 assertEquals("Captured BatchedModifications count", 2, batchedModifications.size());
593 verifyBatchedModifications(batchedModifications.get(0), false,
594 new WriteModification(TestModel.TEST_PATH, nodeToWrite));
596 verifyBatchedModifications(batchedModifications.get(1), true, true);
600 public void testReadyWithReplyFailure() throws Exception {
601 dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
603 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
605 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
607 expectFailedBatchedModifications(actorRef);
609 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
611 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
613 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
615 assertTrue(ready instanceof SingleCommitCohortProxy);
617 verifyCohortFutures((SingleCommitCohortProxy)ready, TestException.class);
621 public void testReadyWithDebugContextEnabled() throws Exception {
622 dataStoreContextBuilder.transactionDebugContextEnabled(true);
624 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
626 expectBatchedModificationsReady(actorRef, true);
628 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
630 transactionProxy.merge(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
632 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
634 assertTrue(ready instanceof DebugThreePhaseCommitCohort);
636 verifyCohortFutures((DebugThreePhaseCommitCohort)ready, new CommitTransactionReply().toSerializable());
640 public void testReadyWithLocalTransaction() throws Exception {
641 ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
643 doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
644 .actorSelection(shardActorRef.path().toString());
646 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, createDataTree()))).when(mockActorContext)
647 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
649 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
651 expectReadyLocalTransaction(shardActorRef, true);
653 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
654 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
656 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
657 assertTrue(ready instanceof SingleCommitCohortProxy);
658 verifyCohortFutures((SingleCommitCohortProxy)ready, new CommitTransactionReply().toSerializable());
662 public void testReadyWithLocalTransactionWithFailure() throws Exception {
663 ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
665 doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
666 .actorSelection(shardActorRef.path().toString());
668 DataTree mockDataTree = createDataTree();
669 DataTreeModification mockModification = mockDataTree.takeSnapshot().newModification();
670 doThrow(new RuntimeException("mock")).when(mockModification).ready();
672 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, mockDataTree))).when(mockActorContext)
673 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
675 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
677 expectReadyLocalTransaction(shardActorRef, true);
679 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
680 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
682 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
683 assertTrue(ready instanceof SingleCommitCohortProxy);
684 verifyCohortFutures((SingleCommitCohortProxy)ready, RuntimeException.class);
687 private void testWriteOnlyTxWithFindPrimaryShardFailure(final Exception toThrow) throws Exception {
688 doReturn(Futures.failed(toThrow)).when(mockActorContext).findPrimaryShardAsync(anyString());
690 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
692 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
694 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
696 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
698 transactionProxy.delete(TestModel.TEST_PATH);
700 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
702 assertTrue(ready instanceof SingleCommitCohortProxy);
704 verifyCohortFutures((SingleCommitCohortProxy)ready, toThrow.getClass());
708 public void testWriteOnlyTxWithPrimaryNotFoundException() throws Exception {
709 testWriteOnlyTxWithFindPrimaryShardFailure(new PrimaryNotFoundException("mock"));
713 public void testWriteOnlyTxWithNotInitializedException() throws Exception {
714 testWriteOnlyTxWithFindPrimaryShardFailure(new NotInitializedException("mock"));
718 public void testWriteOnlyTxWithNoShardLeaderException() throws Exception {
719 testWriteOnlyTxWithFindPrimaryShardFailure(new NoShardLeaderException("mock"));
723 public void testReadyWithInvalidReplyMessageType() throws Exception {
724 dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
725 ActorRef actorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY);
727 ActorRef actorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY, "junk");
729 doReturn(Futures.successful(new Object())).when(mockActorContext).executeOperationAsync(
730 eq(actorSelection(actorRef1)), isA(BatchedModifications.class), any(Timeout.class));
732 expectBatchedModificationsReady(actorRef2);
734 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, WRITE_ONLY);
736 transactionProxy.write(TestModel.JUNK_PATH, ImmutableNodes.containerNode(TestModel.JUNK_QNAME));
737 transactionProxy.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
739 DOMStoreThreePhaseCommitCohort ready = transactionProxy.ready();
741 assertTrue(ready instanceof ThreePhaseCommitCohortProxy);
743 verifyCohortFutures((ThreePhaseCommitCohortProxy)ready, actorSelection(actorRef2),
744 IllegalArgumentException.class);
748 public void testGetIdentifier() {
749 setupActorContextWithInitialCreateTransaction(getSystem(), READ_ONLY);
750 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
752 Object id = transactionProxy.getIdentifier();
753 assertNotNull("getIdentifier returned null", id);
754 assertTrue("Invalid identifier: " + id, id.toString().contains(memberName));
758 public void testClose() throws Exception {
759 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
761 doReturn(readDataReply(null)).when(mockActorContext).executeOperationAsync(
762 eq(actorSelection(actorRef)), eqReadData(), any(Timeout.class));
764 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
766 transactionProxy.read(TestModel.TEST_PATH);
768 transactionProxy.close();
770 verify(mockActorContext).sendOperationAsync(
771 eq(actorSelection(actorRef)), isA(CloseTransaction.class));
774 private interface TransactionProxyOperation {
775 void run(TransactionProxy transactionProxy);
778 private PrimaryShardInfo newPrimaryShardInfo(final ActorRef actorRef) {
779 return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION);
782 private PrimaryShardInfo newPrimaryShardInfo(final ActorRef actorRef, final DataTree dataTree) {
783 return new PrimaryShardInfo(getSystem().actorSelection(actorRef.path()), DataStoreVersions.CURRENT_VERSION,
787 private void throttleOperation(final TransactionProxyOperation operation) {
788 throttleOperation(operation, 1, true);
791 private void throttleOperation(final TransactionProxyOperation operation, final int outstandingOpsLimit,
792 final boolean shardFound) {
793 throttleOperation(operation, outstandingOpsLimit, shardFound, TimeUnit.MILLISECONDS.toNanos(
794 mockActorContext.getDatastoreContext().getOperationTimeoutInMillis()));
797 private void throttleOperation(final TransactionProxyOperation operation, final int outstandingOpsLimit,
798 final boolean shardFound, final long expectedCompletionTime) {
799 ActorSystem actorSystem = getSystem();
800 ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
802 // Note that we setting batchedModificationCount to one less than what we need because in TransactionProxy
803 // we now allow one extra permit to be allowed for ready
804 doReturn(dataStoreContextBuilder.operationTimeoutInSeconds(2)
805 .shardBatchedModificationCount(outstandingOpsLimit - 1).build()).when(mockActorContext)
806 .getDatastoreContext();
808 doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
809 .actorSelection(shardActorRef.path().toString());
812 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
813 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
814 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
815 .findPrimaryShardAsync(eq("cars"));
818 doReturn(Futures.failed(new Exception("not found")))
819 .when(mockActorContext).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
822 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
823 eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_WRITE),
826 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
828 long start = System.nanoTime();
830 operation.run(transactionProxy);
832 long end = System.nanoTime();
834 Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
835 expectedCompletionTime, end - start),
836 end - start > expectedCompletionTime && end - start < expectedCompletionTime * 2);
840 private void completeOperation(final TransactionProxyOperation operation) {
841 completeOperation(operation, true);
844 private void completeOperation(final TransactionProxyOperation operation, final boolean shardFound) {
845 ActorSystem actorSystem = getSystem();
846 ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
848 doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
849 .actorSelection(shardActorRef.path().toString());
852 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef))).when(mockActorContext)
853 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
855 doReturn(Futures.failed(new PrimaryNotFoundException("test"))).when(mockActorContext)
856 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
859 ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
860 String actorPath = txActorRef.path().toString();
861 CreateTransactionReply createTransactionReply = new CreateTransactionReply(actorPath, nextTransactionId(),
862 DataStoreVersions.CURRENT_VERSION);
864 doReturn(actorSystem.actorSelection(actorPath)).when(mockActorContext).actorSelection(actorPath);
866 doReturn(Futures.successful(createTransactionReply)).when(mockActorContext).executeOperationAsync(
867 eq(actorSystem.actorSelection(shardActorRef.path())), eqCreateTransaction(memberName, READ_WRITE),
870 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
872 long start = System.nanoTime();
874 operation.run(transactionProxy);
876 long end = System.nanoTime();
878 long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
879 .getOperationTimeoutInMillis());
880 Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s",
881 expected, end - start), end - start <= expected);
884 private void completeOperationLocal(final TransactionProxyOperation operation, final DataTree dataTree) {
885 ActorSystem actorSystem = getSystem();
886 ActorRef shardActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
888 doReturn(actorSystem.actorSelection(shardActorRef.path())).when(mockActorContext)
889 .actorSelection(shardActorRef.path().toString());
891 doReturn(Futures.successful(newPrimaryShardInfo(shardActorRef, dataTree))).when(mockActorContext)
892 .findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD));
894 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
896 long start = System.nanoTime();
898 operation.run(transactionProxy);
900 long end = System.nanoTime();
902 long expected = TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
903 .getOperationTimeoutInMillis());
904 Assert.assertTrue(String.format("Expected elapsed time: %s. Actual: %s", expected, end - start),
905 end - start <= expected);
908 private static DataTree createDataTree() {
909 DataTree dataTree = mock(DataTree.class);
910 DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
911 DataTreeModification dataTreeModification = mock(DataTreeModification.class);
913 doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
914 doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
919 private static DataTree createDataTree(final NormalizedNode<?, ?> readResponse) {
920 DataTree dataTree = mock(DataTree.class);
921 DataTreeSnapshot dataTreeSnapshot = mock(DataTreeSnapshot.class);
922 DataTreeModification dataTreeModification = mock(DataTreeModification.class);
924 doReturn(dataTreeSnapshot).when(dataTree).takeSnapshot();
925 doReturn(dataTreeModification).when(dataTreeSnapshot).newModification();
926 doReturn(Optional.of(readResponse)).when(dataTreeModification).readNode(any(YangInstanceIdentifier.class));
933 public void testWriteCompletionForLocalShard() {
934 completeOperationLocal(transactionProxy -> {
935 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
937 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
939 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
941 }, createDataTree());
945 public void testWriteThrottlingWhenShardFound() {
946 throttleOperation(transactionProxy -> {
947 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
949 expectIncompleteBatchedModifications();
951 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
953 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
958 public void testWriteThrottlingWhenShardNotFound() {
959 // Confirm that there is no throttling when the Shard is not found
960 completeOperation(transactionProxy -> {
961 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
963 expectBatchedModifications(2);
965 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
967 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
974 public void testWriteCompletion() {
975 completeOperation(transactionProxy -> {
976 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
978 expectBatchedModifications(2);
980 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
982 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
987 public void testMergeThrottlingWhenShardFound() {
988 throttleOperation(transactionProxy -> {
989 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
991 expectIncompleteBatchedModifications();
993 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
995 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1000 public void testMergeThrottlingWhenShardNotFound() {
1001 completeOperation(transactionProxy -> {
1002 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1004 expectBatchedModifications(2);
1006 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1008 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1013 public void testMergeCompletion() {
1014 completeOperation(transactionProxy -> {
1015 NormalizedNode<?, ?> nodeToMerge = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1017 expectBatchedModifications(2);
1019 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1021 transactionProxy.merge(TestModel.TEST_PATH, nodeToMerge);
1027 public void testMergeCompletionForLocalShard() {
1028 completeOperationLocal(transactionProxy -> {
1029 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1031 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1033 transactionProxy.merge(TestModel.TEST_PATH, nodeToWrite);
1035 }, createDataTree());
1040 public void testDeleteThrottlingWhenShardFound() {
1042 throttleOperation(transactionProxy -> {
1043 expectIncompleteBatchedModifications();
1045 transactionProxy.delete(TestModel.TEST_PATH);
1047 transactionProxy.delete(TestModel.TEST_PATH);
1053 public void testDeleteThrottlingWhenShardNotFound() {
1055 completeOperation(transactionProxy -> {
1056 expectBatchedModifications(2);
1058 transactionProxy.delete(TestModel.TEST_PATH);
1060 transactionProxy.delete(TestModel.TEST_PATH);
1065 public void testDeleteCompletionForLocalShard() {
1066 completeOperationLocal(transactionProxy -> {
1068 transactionProxy.delete(TestModel.TEST_PATH);
1070 transactionProxy.delete(TestModel.TEST_PATH);
1071 }, createDataTree());
1076 public void testDeleteCompletion() {
1077 completeOperation(transactionProxy -> {
1078 expectBatchedModifications(2);
1080 transactionProxy.delete(TestModel.TEST_PATH);
1082 transactionProxy.delete(TestModel.TEST_PATH);
1088 public void testReadThrottlingWhenShardFound() {
1090 throttleOperation(transactionProxy -> {
1091 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1092 any(ActorSelection.class), eqReadData());
1094 transactionProxy.read(TestModel.TEST_PATH);
1096 transactionProxy.read(TestModel.TEST_PATH);
1101 public void testReadThrottlingWhenShardNotFound() {
1103 completeOperation(transactionProxy -> {
1104 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1105 any(ActorSelection.class), eqReadData());
1107 transactionProxy.read(TestModel.TEST_PATH);
1109 transactionProxy.read(TestModel.TEST_PATH);
1115 public void testReadCompletion() {
1116 completeOperation(transactionProxy -> {
1117 NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1119 doReturn(readDataReply(nodeToRead)).when(mockActorContext).executeOperationAsync(
1120 any(ActorSelection.class), eqReadData(), any(Timeout.class));
1122 transactionProxy.read(TestModel.TEST_PATH);
1124 transactionProxy.read(TestModel.TEST_PATH);
1130 public void testReadCompletionForLocalShard() {
1131 final NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1132 completeOperationLocal(transactionProxy -> {
1133 transactionProxy.read(TestModel.TEST_PATH);
1135 transactionProxy.read(TestModel.TEST_PATH);
1136 }, createDataTree(nodeToRead));
1141 public void testReadCompletionForLocalShardWhenExceptionOccurs() {
1142 completeOperationLocal(transactionProxy -> {
1143 transactionProxy.read(TestModel.TEST_PATH);
1145 transactionProxy.read(TestModel.TEST_PATH);
1146 }, createDataTree());
1151 public void testExistsThrottlingWhenShardFound() {
1153 throttleOperation(transactionProxy -> {
1154 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1155 any(ActorSelection.class), eqDataExists());
1157 transactionProxy.exists(TestModel.TEST_PATH);
1159 transactionProxy.exists(TestModel.TEST_PATH);
1164 public void testExistsThrottlingWhenShardNotFound() {
1166 completeOperation(transactionProxy -> {
1167 doReturn(incompleteFuture()).when(mockActorContext).executeOperationAsync(
1168 any(ActorSelection.class), eqDataExists());
1170 transactionProxy.exists(TestModel.TEST_PATH);
1172 transactionProxy.exists(TestModel.TEST_PATH);
1178 public void testExistsCompletion() {
1179 completeOperation(transactionProxy -> {
1180 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1181 any(ActorSelection.class), eqDataExists(), any(Timeout.class));
1183 transactionProxy.exists(TestModel.TEST_PATH);
1185 transactionProxy.exists(TestModel.TEST_PATH);
1191 public void testExistsCompletionForLocalShard() {
1192 final NormalizedNode<?, ?> nodeToRead = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1193 completeOperationLocal(transactionProxy -> {
1194 transactionProxy.exists(TestModel.TEST_PATH);
1196 transactionProxy.exists(TestModel.TEST_PATH);
1197 }, createDataTree(nodeToRead));
1202 public void testExistsCompletionForLocalShardWhenExceptionOccurs() {
1203 completeOperationLocal(transactionProxy -> {
1204 transactionProxy.exists(TestModel.TEST_PATH);
1206 transactionProxy.exists(TestModel.TEST_PATH);
1207 }, createDataTree());
1212 public void testReadyThrottling() {
1214 throttleOperation(transactionProxy -> {
1215 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1217 expectBatchedModifications(1);
1219 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1221 transactionProxy.ready();
1226 public void testReadyThrottlingWithTwoTransactionContexts() {
1227 throttleOperation(transactionProxy -> {
1228 NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1229 NormalizedNode<?, ?> carsNode = ImmutableNodes.containerNode(CarsModel.BASE_QNAME);
1231 expectBatchedModifications(2);
1233 transactionProxy.write(TestModel.TEST_PATH, nodeToWrite);
1235 // Trying to write to Cars will cause another transaction context to get created
1236 transactionProxy.write(CarsModel.BASE_PATH, carsNode);
1238 // Now ready should block for both transaction contexts
1239 transactionProxy.ready();
1240 }, 1, true, TimeUnit.MILLISECONDS.toNanos(mockActorContext.getDatastoreContext()
1241 .getOperationTimeoutInMillis()) * 2);
1244 private void testModificationOperationBatching(final TransactionType type) throws Exception {
1245 int shardBatchedModificationCount = 3;
1246 dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1248 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), type);
1250 expectBatchedModifications(actorRef, shardBatchedModificationCount);
1252 YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1253 NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1255 YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1256 NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1258 YangInstanceIdentifier writePath3 = TestModel.INNER_LIST_PATH;
1259 NormalizedNode<?, ?> writeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1261 YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1262 NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1264 YangInstanceIdentifier mergePath2 = TestModel.OUTER_LIST_PATH;
1265 NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1267 YangInstanceIdentifier mergePath3 = TestModel.INNER_LIST_PATH;
1268 NormalizedNode<?, ?> mergeNode3 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1270 YangInstanceIdentifier deletePath1 = TestModel.TEST_PATH;
1271 YangInstanceIdentifier deletePath2 = TestModel.OUTER_LIST_PATH;
1273 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, type);
1275 transactionProxy.write(writePath1, writeNode1);
1276 transactionProxy.write(writePath2, writeNode2);
1277 transactionProxy.delete(deletePath1);
1278 transactionProxy.merge(mergePath1, mergeNode1);
1279 transactionProxy.merge(mergePath2, mergeNode2);
1280 transactionProxy.write(writePath3, writeNode3);
1281 transactionProxy.merge(mergePath3, mergeNode3);
1282 transactionProxy.delete(deletePath2);
1284 // This sends the last batch.
1285 transactionProxy.ready();
1287 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1288 assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1290 verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1291 new WriteModification(writePath2, writeNode2), new DeleteModification(deletePath1));
1293 verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1294 new MergeModification(mergePath2, mergeNode2), new WriteModification(writePath3, writeNode3));
1296 verifyBatchedModifications(batchedModifications.get(2), true, true,
1297 new MergeModification(mergePath3, mergeNode3), new DeleteModification(deletePath2));
1299 assertEquals("getTotalMessageCount", 3, batchedModifications.get(2).getTotalMessagesSent());
1303 public void testReadWriteModificationOperationBatching() throws Exception {
1304 testModificationOperationBatching(READ_WRITE);
1308 public void testWriteOnlyModificationOperationBatching() throws Exception {
1309 testModificationOperationBatching(WRITE_ONLY);
1313 public void testOptimizedWriteOnlyModificationOperationBatching() throws Exception {
1314 dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true);
1315 testModificationOperationBatching(WRITE_ONLY);
1319 public void testModificationOperationBatchingWithInterleavedReads() throws Exception {
1321 int shardBatchedModificationCount = 10;
1322 dataStoreContextBuilder.shardBatchedModificationCount(shardBatchedModificationCount);
1324 ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE);
1326 expectBatchedModifications(actorRef, shardBatchedModificationCount);
1328 final YangInstanceIdentifier writePath1 = TestModel.TEST_PATH;
1329 final NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1331 YangInstanceIdentifier writePath2 = TestModel.OUTER_LIST_PATH;
1332 NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME);
1334 final YangInstanceIdentifier mergePath1 = TestModel.TEST_PATH;
1335 final NormalizedNode<?, ?> mergeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1337 YangInstanceIdentifier mergePath2 = TestModel.INNER_LIST_PATH;
1338 NormalizedNode<?, ?> mergeNode2 = ImmutableNodes.containerNode(TestModel.INNER_LIST_QNAME);
1340 final YangInstanceIdentifier deletePath = TestModel.OUTER_LIST_PATH;
1342 doReturn(readDataReply(writeNode2)).when(mockActorContext).executeOperationAsync(
1343 eq(actorSelection(actorRef)), eqReadData(writePath2), any(Timeout.class));
1345 doReturn(readDataReply(mergeNode2)).when(mockActorContext).executeOperationAsync(
1346 eq(actorSelection(actorRef)), eqReadData(mergePath2), any(Timeout.class));
1348 doReturn(dataExistsReply(true)).when(mockActorContext).executeOperationAsync(
1349 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
1351 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_WRITE);
1353 transactionProxy.write(writePath1, writeNode1);
1354 transactionProxy.write(writePath2, writeNode2);
1356 Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(writePath2).get(5, TimeUnit.SECONDS);
1358 assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1359 assertEquals("Response NormalizedNode", writeNode2, readOptional.get());
1361 transactionProxy.merge(mergePath1, mergeNode1);
1362 transactionProxy.merge(mergePath2, mergeNode2);
1364 readOptional = transactionProxy.read(mergePath2).get(5, TimeUnit.SECONDS);
1366 transactionProxy.delete(deletePath);
1368 Boolean exists = transactionProxy.exists(TestModel.TEST_PATH).checkedGet();
1369 assertEquals("Exists response", true, exists);
1371 assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1372 assertEquals("Response NormalizedNode", mergeNode2, readOptional.get());
1374 List<BatchedModifications> batchedModifications = captureBatchedModifications(actorRef);
1375 assertEquals("Captured BatchedModifications count", 3, batchedModifications.size());
1377 verifyBatchedModifications(batchedModifications.get(0), false, new WriteModification(writePath1, writeNode1),
1378 new WriteModification(writePath2, writeNode2));
1380 verifyBatchedModifications(batchedModifications.get(1), false, new MergeModification(mergePath1, mergeNode1),
1381 new MergeModification(mergePath2, mergeNode2));
1383 verifyBatchedModifications(batchedModifications.get(2), false, new DeleteModification(deletePath));
1385 InOrder inOrder = Mockito.inOrder(mockActorContext);
1386 inOrder.verify(mockActorContext).executeOperationAsync(
1387 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1389 inOrder.verify(mockActorContext).executeOperationAsync(
1390 eq(actorSelection(actorRef)), eqReadData(writePath2), any(Timeout.class));
1392 inOrder.verify(mockActorContext).executeOperationAsync(
1393 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1395 inOrder.verify(mockActorContext).executeOperationAsync(
1396 eq(actorSelection(actorRef)), eqReadData(mergePath2), any(Timeout.class));
1398 inOrder.verify(mockActorContext).executeOperationAsync(
1399 eq(actorSelection(actorRef)), isA(BatchedModifications.class), any(Timeout.class));
1401 inOrder.verify(mockActorContext).executeOperationAsync(
1402 eq(actorSelection(actorRef)), eqDataExists(), any(Timeout.class));
1406 public void testReadRoot() throws ReadFailedException, InterruptedException, ExecutionException,
1407 java.util.concurrent.TimeoutException {
1408 SchemaContext schemaContext = SchemaContextHelper.full();
1409 Configuration configuration = mock(Configuration.class);
1410 doReturn(configuration).when(mockActorContext).getConfiguration();
1411 doReturn(schemaContext).when(mockActorContext).getSchemaContext();
1412 doReturn(Sets.newHashSet("test", "cars")).when(configuration).getAllShardNames();
1414 NormalizedNode<?, ?> expectedNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
1415 NormalizedNode<?, ?> expectedNode2 = ImmutableNodes.containerNode(CarsModel.CARS_QNAME);
1417 setUpReadData("test", NormalizedNodeAggregatorTest.getRootNode(expectedNode1, schemaContext));
1418 setUpReadData("cars", NormalizedNodeAggregatorTest.getRootNode(expectedNode2, schemaContext));
1420 doReturn(MemberName.forName(memberName)).when(mockActorContext).getCurrentMemberName();
1422 doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(mockActorContext).getClientDispatcher();
1424 TransactionProxy transactionProxy = new TransactionProxy(mockComponentFactory, READ_ONLY);
1426 Optional<NormalizedNode<?, ?>> readOptional = transactionProxy.read(
1427 YangInstanceIdentifier.EMPTY).get(5, TimeUnit.SECONDS);
1429 assertEquals("NormalizedNode isPresent", true, readOptional.isPresent());
1431 NormalizedNode<?, ?> normalizedNode = readOptional.get();
1433 assertTrue("Expect value to be a Collection", normalizedNode.getValue() instanceof Collection);
1435 @SuppressWarnings("unchecked")
1436 Collection<NormalizedNode<?,?>> collection = (Collection<NormalizedNode<?,?>>) normalizedNode.getValue();
1438 for (NormalizedNode<?,?> node : collection) {
1439 assertTrue("Expected " + node + " to be a ContainerNode", node instanceof ContainerNode);
1442 assertTrue("Child with QName = " + TestModel.TEST_QNAME + " not found",
1443 NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME) != null);
1445 assertEquals(expectedNode1, NormalizedNodeAggregatorTest.findChildWithQName(collection, TestModel.TEST_QNAME));
1447 assertTrue("Child with QName = " + CarsModel.BASE_QNAME + " not found",
1448 NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME) != null);
1450 assertEquals(expectedNode2, NormalizedNodeAggregatorTest.findChildWithQName(collection, CarsModel.BASE_QNAME));
1454 private void setUpReadData(final String shardName, final NormalizedNode<?, ?> expectedNode) {
1455 ActorSystem actorSystem = getSystem();
1456 ActorRef shardActorRef = getSystem().actorOf(Props.create(DoNothingActor.class));
1458 doReturn(getSystem().actorSelection(shardActorRef.path())).when(mockActorContext)
1459 .actorSelection(shardActorRef.path().toString());
1461 doReturn(primaryShardInfoReply(getSystem(), shardActorRef)).when(mockActorContext)
1462 .findPrimaryShardAsync(eq(shardName));
1464 ActorRef txActorRef = actorSystem.actorOf(Props.create(DoNothingActor.class));
1466 doReturn(actorSystem.actorSelection(txActorRef.path())).when(mockActorContext)
1467 .actorSelection(txActorRef.path().toString());
1469 doReturn(Futures.successful(createTransactionReply(txActorRef, DataStoreVersions.CURRENT_VERSION)))
1470 .when(mockActorContext).executeOperationAsync(eq(actorSystem.actorSelection(shardActorRef.path())),
1471 eqCreateTransaction(memberName, TransactionType.READ_ONLY), any(Timeout.class));
1473 doReturn(readDataReply(expectedNode)).when(mockActorContext).executeOperationAsync(
1474 eq(actorSelection(txActorRef)), eqReadData(YangInstanceIdentifier.EMPTY), any(Timeout.class));