+++ /dev/null
-/*
- * Copyright (c) 2017 Pantheon Technologies, s.r.o. and others. All rights reserved.
- *
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-package org.opendaylight.controller.cluster.datastore;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-
-import akka.actor.ActorRef;
-import akka.actor.Status.Failure;
-import akka.dispatch.ExecutionContexts;
-import akka.dispatch.OnComplete;
-import akka.testkit.javadsl.TestKit;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.SettableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mockito;
-import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
-import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
-import org.opendaylight.controller.cluster.access.concepts.FrontendType;
-import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
-import org.opendaylight.controller.cluster.access.concepts.MemberName;
-import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.config.Configuration;
-import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
-import org.opendaylight.controller.cluster.datastore.messages.DataExists;
-import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.duration.FiniteDuration;
-
-/**
- * Test whether RmoteTransactionContext operates correctly.
- */
-public class RemoteTransactionContextTest extends AbstractActorTest {
- private static final TransactionIdentifier TX_ID = new TransactionIdentifier(new LocalHistoryIdentifier(
- ClientIdentifier.create(FrontendIdentifier.create(MemberName.forName("test"), FrontendType.forName("test")), 0),
- 0), 0);
-
- private OperationLimiter limiter;
- private RemoteTransactionContext txContext;
- private ActorUtils actorUtils;
- private TestKit kit;
-
- @Before
- public void before() {
- kit = new TestKit(getSystem());
- actorUtils = Mockito.spy(new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
- mock(Configuration.class)));
- limiter = new OperationLimiter(TX_ID, 4, 0);
- txContext = new RemoteTransactionContext(TX_ID, actorUtils.actorSelection(kit.getRef().path()), actorUtils,
- DataStoreVersions.CURRENT_VERSION, limiter);
- txContext.operationHandOffComplete();
- }
-
- /**
- * OperationLimiter should be correctly released when a failure, like AskTimeoutException occurs. Future reads
- * need to complete immediately with the failure and modifications should not be throttled and thrown away
- * immediately.
- */
- @Test
- public void testLimiterOnFailure() throws TimeoutException, InterruptedException {
- txContext.executeDelete(null, null);
- txContext.executeDelete(null, null);
- assertEquals(2, limiter.availablePermits());
-
- final Future<Object> sendFuture = txContext.sendBatchedModifications();
- assertEquals(2, limiter.availablePermits());
-
- BatchedModifications msg = kit.expectMsgClass(BatchedModifications.class);
- assertEquals(2, msg.getModifications().size());
- assertEquals(1, msg.getTotalMessagesSent());
- sendReply(new Failure(new NullPointerException()));
- assertFuture(sendFuture, new OnComplete<>() {
- @Override
- public void onComplete(final Throwable failure, final Object success) {
- assertTrue(failure instanceof NullPointerException);
- assertEquals(4, limiter.availablePermits());
-
- // The transaction has failed, no throttling should occur
- txContext.executeDelete(null, null);
- assertEquals(4, limiter.availablePermits());
-
- // Executing a read should result in immediate failure
- final SettableFuture<Boolean> readFuture = SettableFuture.create();
- txContext.executeRead(new DataExists(), readFuture, null);
- assertTrue(readFuture.isDone());
- try {
- readFuture.get();
- fail("Read future did not fail");
- } catch (ExecutionException | InterruptedException e) {
- assertTrue(e.getCause() instanceof NullPointerException);
- }
- }
- });
-
- final Future<Object> commitFuture = txContext.directCommit(null);
-
- msg = kit.expectMsgClass(BatchedModifications.class);
- // Modification should have been thrown away by the dropped transmit induced by executeRead()
- assertEquals(0, msg.getModifications().size());
- assertTrue(msg.isDoCommitOnReady());
- assertTrue(msg.isReady());
- assertEquals(2, msg.getTotalMessagesSent());
- sendReply(new Failure(new IllegalStateException()));
- assertFuture(commitFuture, new OnComplete<>() {
- @Override
- public void onComplete(final Throwable failure, final Object success) {
- assertTrue(failure instanceof IllegalStateException);
- }
- });
-
- kit.expectNoMessage();
- }
-
- /**
- * OperationLimiter gives up throttling at some point -- {@link RemoteTransactionContext} needs to deal with that
- * case, too.
- */
- @Test
- public void testLimiterOnOverflowFailure() throws TimeoutException, InterruptedException {
- txContext.executeDelete(null, null);
- txContext.executeDelete(null, null);
- txContext.executeDelete(null, null);
- txContext.executeDelete(null, null);
- assertEquals(0, limiter.availablePermits());
- txContext.executeDelete(null, null);
- // Last acquire should have failed ...
- assertEquals(0, limiter.availablePermits());
-
- final Future<Object> future = txContext.sendBatchedModifications();
- assertEquals(0, limiter.availablePermits());
-
- BatchedModifications msg = kit.expectMsgClass(BatchedModifications.class);
- // ... so we are sending 5 modifications ...
- assertEquals(5, msg.getModifications().size());
- assertEquals(1, msg.getTotalMessagesSent());
- sendReply(new Failure(new NullPointerException()));
-
- assertFuture(future, new OnComplete<>() {
- @Override
- public void onComplete(final Throwable failure, final Object success) {
- assertTrue(failure instanceof NullPointerException);
- // ... but they account for only 4 permits.
- assertEquals(4, limiter.availablePermits());
- }
- });
-
- kit.expectNoMessage();
- }
-
- private void sendReply(final Object message) {
- final ActorRef askActor = kit.getLastSender();
- kit.watch(askActor);
- kit.reply(new Failure(new IllegalStateException()));
- kit.expectTerminated(askActor);
- }
-
- private static void assertFuture(final Future<Object> future, final OnComplete<Object> complete)
- throws TimeoutException, InterruptedException {
- Await.ready(future, FiniteDuration.apply(3, TimeUnit.SECONDS));
- future.onComplete(complete, ExecutionContexts.fromExecutor(MoreExecutors.directExecutor()));
- }
-}
Code Review / controller.git / blobdiff