opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientTransactionCommitCohortTest.java

   1 /*
   2  * Copyright (c) 2017 Pantheon Technologies s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.databroker.actors.dds;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.CLIENT_ID;
  12 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.HISTORY_ID;
  13 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.TRANSACTION_ID;
  14 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.assertOperationThrowsException;
  15 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.getWithTimeout;
  16
  17 import akka.actor.ActorSystem;
  18 import akka.testkit.TestProbe;
  19 import akka.testkit.javadsl.TestKit;
  20 import com.google.common.primitives.UnsignedLong;
  21 import com.google.common.util.concurrent.ListenableFuture;
  22 import java.util.ArrayList;
  23 import java.util.Collection;
  24 import java.util.List;
  25 import java.util.Optional;
  26 import java.util.concurrent.ExecutionException;
  27 import java.util.function.Consumer;
  28 import java.util.function.Function;
  29 import java.util.stream.Collectors;
  30 import org.junit.After;
  31 import org.junit.Before;
  32 import org.junit.Test;
  33 import org.junit.runner.RunWith;
  34 import org.mockito.Mock;
  35 import org.mockito.junit.MockitoJUnitRunner;
  36 import org.opendaylight.controller.cluster.access.ABIVersion;
  37 import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
  38 import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
  39 import org.opendaylight.controller.cluster.access.client.ClientActorContext;
  40 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
  41 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
  42 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
  43 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
  44 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
  45 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
  46 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
  47 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
  48 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
  49 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
  50 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
  51
  52 @RunWith(MockitoJUnitRunner.StrictStubs.class)
  53 public class ClientTransactionCommitCohortTest {
  54
  55     private static final String PERSISTENCE_ID = "per-1";
  56     private static final int TRANSACTIONS = 3;
  57
  58     @Mock
  59     private AbstractClientHistory history;
  60     private ActorSystem system;
  61     private List<TransactionTester<RemoteProxyTransaction>> transactions;
  62     private ClientTransactionCommitCohort cohort;
  63
  64     @Before
  65     public void setUp() {
  66         system = ActorSystem.apply();
  67         final TestProbe clientContextProbe = new TestProbe(system, "clientContext");
  68         final ClientActorContext context =
  69                 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
  70         transactions = new ArrayList<>();
  71         for (int i = 0; i < TRANSACTIONS; i++) {
  72             transactions.add(createTransactionTester(new TestProbe(system, "backend" + i), context, history));
  73         }
  74         final Collection<AbstractProxyTransaction> proxies = transactions.stream()
  75                 .map(TransactionTester::getTransaction)
  76                 .collect(Collectors.toList());
  77         proxies.forEach(AbstractProxyTransaction::seal);
  78         cohort = new ClientTransactionCommitCohort(history, TRANSACTION_ID, proxies);
  79     }
  80
  81     @After
  82     public void tearDown() {
  83         TestKit.shutdownActorSystem(system);
  84     }
  85
  86     @Test
  87     public void testCanCommit() throws Exception {
  88         testOpSuccess(ClientTransactionCommitCohort::canCommit, this::expectCanCommit,
  89                 this::replyCanCommitSuccess, Boolean.TRUE);
  90     }
  91
  92     @Test
  93     public void testCanCommitFail() throws Exception {
  94         testOpFail(ClientTransactionCommitCohort::canCommit, this::expectCanCommit, this::replyCanCommitSuccess);
  95     }
  96
  97     @Test
  98     public void testPreCommit() throws Exception {
  99         testOpSuccess(ClientTransactionCommitCohort::preCommit, this::expectPreCommit, this::replyPreCommitSuccess,
 100                 null);
 101     }
 102
 103     @Test
 104     public void testPreCommitFail() throws Exception {
 105         testOpFail(ClientTransactionCommitCohort::preCommit, this::expectPreCommit, this::replyPreCommitSuccess);
 106     }
 107
 108     @Test
 109     public void testCommit() throws Exception {
 110         testOpSuccess(ClientTransactionCommitCohort::commit, this::expectCommit, this::replyCommitSuccess, null);
 111     }
 112
 113     @Test
 114     public void testCommitFail() throws Exception {
 115         testOpFail(ClientTransactionCommitCohort::commit, this::expectCommit, this::replyCommitSuccess);
 116     }
 117
 118     @Test
 119     public void testAbort() throws Exception {
 120         testOpSuccess(ClientTransactionCommitCohort::abort, this::expectAbort, this::replyAbortSuccess, null);
 121     }
 122
 123     @Test
 124     public void testAbortFail() throws Exception {
 125         testOpFail(ClientTransactionCommitCohort::abort, this::expectAbort, this::replyAbortSuccess);
 126     }
 127
 128     private void expectCanCommit(final TransactionTester<RemoteProxyTransaction> tester) {
 129         final ModifyTransactionRequest request = tester.expectTransactionRequest(ModifyTransactionRequest.class);
 130         assertEquals(Optional.of(PersistenceProtocol.THREE_PHASE), request.getPersistenceProtocol());
 131     }
 132
 133     void expectPreCommit(final TransactionTester<?> tester) {
 134         tester.expectTransactionRequest(TransactionPreCommitRequest.class);
 135     }
 136
 137     void expectCommit(final TransactionTester<?> tester) {
 138         tester.expectTransactionRequest(TransactionDoCommitRequest.class);
 139     }
 140
 141     void expectAbort(final TransactionTester<?> tester) {
 142         tester.expectTransactionRequest(TransactionAbortRequest.class);
 143     }
 144
 145     void replyCanCommitSuccess(final TransactionTester<?> tester) {
 146         final RequestSuccess<?, ?> success = new TransactionCanCommitSuccess(tester.getTransaction().getIdentifier(),
 147                 tester.getLastReceivedMessage().getSequence());
 148         tester.replySuccess(success);
 149     }
 150
 151     void replyPreCommitSuccess(final TransactionTester<?> tester) {
 152         final RequestSuccess<?, ?> success = new TransactionPreCommitSuccess(tester.getTransaction().getIdentifier(),
 153                 tester.getLastReceivedMessage().getSequence());
 154         tester.replySuccess(success);
 155     }
 156
 157     void replyCommitSuccess(final TransactionTester<?> tester) {
 158         final RequestSuccess<?, ?> success = new TransactionCommitSuccess(tester.getTransaction().getIdentifier(),
 159                 tester.getLastReceivedMessage().getSequence());
 160         tester.replySuccess(success);
 161     }
 162
 163     void replyAbortSuccess(final TransactionTester<?> tester) {
 164         final RequestSuccess<?, ?> success = new TransactionAbortSuccess(tester.getTransaction().getIdentifier(),
 165                 tester.getLastReceivedMessage().getSequence());
 166         tester.replySuccess(success);
 167     }
 168
 169     private static TransactionTester<RemoteProxyTransaction> createTransactionTester(final TestProbe backendProbe,
 170                                                              final ClientActorContext context,
 171                                                              final AbstractClientHistory history) {
 172         final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
 173                 "default", UnsignedLong.ZERO, Optional.empty(), 3);
 174         final AbstractClientConnection<ShardBackendInfo> connection =
 175                 AccessClientUtil.createConnectedConnection(context, 0L, backend);
 176         final ProxyHistory proxyHistory = ProxyHistory.createClient(history, connection, HISTORY_ID);
 177         final RemoteProxyTransaction transaction =
 178                 new RemoteProxyTransaction(proxyHistory, TRANSACTION_ID, false, false, false);
 179         return new TransactionTester<>(transaction, connection, backendProbe);
 180     }
 181
 182     private static <T extends TransactionTester<?>> void replySuccess(final Collection<T> transactions,
 183                               final Consumer<T> expect, final Consumer<T> reply) {
 184         for (final T transaction : transactions) {
 185             expect.accept(transaction);
 186             reply.accept(transaction);
 187         }
 188     }
 189
 190     /**
 191      * Test operation success. Invokes given operation, which initiates message to the backend.
 192      * Received message is checked by expectFunction. Then replyFunction is invoked. Expected result is compared
 193      * to the operation future result.
 194      *
 195      * @param operation      operation
 196      * @param expectFunction expected message check
 197      * @param replyFunction  response function
 198      * @param expectedResult expected operation result
 199      * @param <T>            type
 200      * @throws Exception unexpected exception
 201      */
 202     private <T> void testOpSuccess(final Function<ClientTransactionCommitCohort, ListenableFuture<T>> operation,
 203                                    final Consumer<TransactionTester<RemoteProxyTransaction>> expectFunction,
 204                                    final Consumer<TransactionTester<RemoteProxyTransaction>> replyFunction,
 205                                    final T expectedResult) throws Exception {
 206         final ListenableFuture<T> result = operation.apply(cohort);
 207         replySuccess(transactions, expectFunction, replyFunction);
 208         assertEquals(expectedResult, getWithTimeout(result));
 209     }
 210
 211     /**
 212      * Test operation failure. Invokes given operation, which initiates message to the backend.
 213      * Received message is checked by expectFunction. Then replyFunction is invoked. One of the transactions in
 214      * cohort receives failure response.
 215      *
 216      * @param operation      operation
 217      * @param expectFunction expected message check
 218      * @param replyFunction  response function
 219      * @param <T>            type
 220      * @throws Exception unexpected exception
 221      */
 222     private <T> void testOpFail(final Function<ClientTransactionCommitCohort, ListenableFuture<T>> operation,
 223             final Consumer<TransactionTester<RemoteProxyTransaction>> expectFunction,
 224             final Consumer<TransactionTester<RemoteProxyTransaction>> replyFunction) throws Exception {
 225         final ListenableFuture<T> canCommit = operation.apply(cohort);
 226         //reply success to all except last transaction
 227         replySuccess(transactions.subList(0, transactions.size() - 1), expectFunction, replyFunction);
 228         //reply fail to last transaction
 229         final TransactionTester<RemoteProxyTransaction> last = transactions.get(transactions.size() - 1);
 230         expectFunction.accept(last);
 231         final RuntimeException e = new RuntimeException();
 232         final RuntimeRequestException cause = new RuntimeRequestException("fail", e);
 233         last.replyFailure(cause);
 234         //check future fail
 235         final ExecutionException exception =
 236                 assertOperationThrowsException(() -> getWithTimeout(canCommit), ExecutionException.class);
 237         assertEquals(e, exception.getCause());
 238     }
 239
 240 }