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

   1 /*
   2  * Copyright (c) 2018 Inocybe Technologies and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertTrue;
  12 import static org.mockito.Matchers.any;
  13 import static org.mockito.Mockito.doReturn;
  14 import static org.mockito.Mockito.mock;
  15 import static org.mockito.Mockito.reset;
  16 import static org.mockito.Mockito.verify;
  17
  18 import akka.actor.ActorRef;
  19 import akka.pattern.Patterns;
  20 import akka.util.Timeout;
  21 import com.google.common.util.concurrent.Futures;
  22 import com.google.common.util.concurrent.JdkFutureAdapters;
  23 import com.google.common.util.concurrent.ListenableFuture;
  24 import com.google.common.util.concurrent.MoreExecutors;
  25 import com.google.common.util.concurrent.Uninterruptibles;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.concurrent.ExecutorService;
  29 import java.util.concurrent.Executors;
  30 import java.util.concurrent.TimeUnit;
  31 import org.junit.After;
  32 import org.junit.Before;
  33 import org.junit.Test;
  34 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  35 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Abort;
  36 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit;
  37 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Commit;
  38 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CommitProtocolCommand;
  39 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.PreCommit;
  40 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
  41 import org.opendaylight.controller.cluster.raft.TestActorFactory;
  42 import org.opendaylight.mdsal.common.api.PostCanCommitStep;
  43 import org.opendaylight.mdsal.common.api.PostPreCommitStep;
  44 import org.opendaylight.mdsal.common.api.ThreePhaseCommitStep;
  45 import org.opendaylight.mdsal.dom.api.DOMDataTreeCandidate;
  46 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
  47 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
  49 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  50 import scala.concurrent.Await;
  51
  52 /**
  53  * Unit tests for DataTreeCohortActor.
  54  *
  55  * @author Thomas Pantelis
  56  */
  57 public class DataTreeCohortActorTest extends AbstractActorTest {
  58     private static final Collection<DOMDataTreeCandidate> CANDIDATES = new ArrayList<>();
  59     private static final SchemaContext MOCK_SCHEMA = mock(SchemaContext.class);
  60     private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
  61     private final DOMDataTreeCommitCohort mockCohort = mock(DOMDataTreeCommitCohort.class);
  62     private final PostCanCommitStep mockPostCanCommit = mock(PostCanCommitStep.class);
  63     private final PostPreCommitStep mockPostPreCommit = mock(PostPreCommitStep.class);
  64
  65     @Before
  66     public void setup() {
  67         resetMockCohort();
  68     }
  69
  70     @After
  71     public void tearDown() {
  72         actorFactory.close();
  73     }
  74
  75     @Test
  76     public void testSuccessfulThreePhaseCommit() throws Exception {
  77         ActorRef cohortActor = newCohortActor("testSuccessfulThreePhaseCommit");
  78
  79         TransactionIdentifier txId = nextTransactionId();
  80         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
  81         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
  82
  83         askAndAwait(cohortActor, new PreCommit(txId));
  84         verify(mockPostCanCommit).preCommit();
  85
  86         askAndAwait(cohortActor, new Commit(txId));
  87         verify(mockPostPreCommit).commit();
  88
  89         resetMockCohort();
  90         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
  91         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
  92     }
  93
  94     @Test
  95     public void testMultipleThreePhaseCommits() throws Exception {
  96         ActorRef cohortActor = newCohortActor("testMultipleThreePhaseCommits");
  97
  98         TransactionIdentifier txId1 = nextTransactionId();
  99         TransactionIdentifier txId2 = nextTransactionId();
 100
 101         askAndAwait(cohortActor, new CanCommit(txId1, CANDIDATES, MOCK_SCHEMA, cohortActor));
 102         askAndAwait(cohortActor, new CanCommit(txId2, CANDIDATES, MOCK_SCHEMA, cohortActor));
 103
 104         askAndAwait(cohortActor, new PreCommit(txId1));
 105         askAndAwait(cohortActor, new PreCommit(txId2));
 106
 107         askAndAwait(cohortActor, new Commit(txId1));
 108         askAndAwait(cohortActor, new Commit(txId2));
 109     }
 110
 111     @SuppressWarnings("unchecked")
 112     @Test
 113     public void testAsyncCohort() throws Exception {
 114         ExecutorService executor = Executors.newSingleThreadExecutor();
 115
 116         doReturn(Futures.makeChecked(executeWithDelay(executor, mockPostCanCommit),
 117             ex -> new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock")))
 118                 .when(mockCohort).canCommit(any(Object.class), any(Collection.class), any(SchemaContext.class));
 119
 120         doReturn(JdkFutureAdapters.listenInPoolThread(executor.submit(() ->
 121             mockPostPreCommit), MoreExecutors.directExecutor())).when(mockPostCanCommit).preCommit();
 122
 123         doReturn(JdkFutureAdapters.listenInPoolThread(executor.submit(() ->
 124             null), MoreExecutors.directExecutor())).when(mockPostPreCommit).commit();
 125
 126         ActorRef cohortActor = newCohortActor("testAsyncCohort");
 127
 128         TransactionIdentifier txId = nextTransactionId();
 129         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 130         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
 131
 132         askAndAwait(cohortActor, new PreCommit(txId));
 133         verify(mockPostCanCommit).preCommit();
 134
 135         askAndAwait(cohortActor, new Commit(txId));
 136         verify(mockPostPreCommit).commit();
 137
 138         executor.shutdownNow();
 139     }
 140
 141     @SuppressWarnings("unchecked")
 142     @Test
 143     public void testFailureOnCanCommit() throws Exception {
 144         DataValidationFailedException failure = new DataValidationFailedException(YangInstanceIdentifier.EMPTY, "mock");
 145         doReturn(Futures.immediateFailedCheckedFuture(failure)).when(mockCohort).canCommit(any(Object.class),
 146                 any(Collection.class), any(SchemaContext.class));
 147
 148         ActorRef cohortActor = newCohortActor("testFailureOnCanCommit");
 149
 150         TransactionIdentifier txId = nextTransactionId();
 151         try {
 152             askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 153         } catch (DataValidationFailedException e) {
 154             assertEquals("DataValidationFailedException", failure, e);
 155         }
 156
 157         resetMockCohort();
 158         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 159         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
 160     }
 161
 162     @Test
 163     public void testAbortAfterCanCommit() throws Exception {
 164         ActorRef cohortActor = newCohortActor("testAbortAfterCanCommit");
 165
 166         TransactionIdentifier txId = nextTransactionId();
 167         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 168         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
 169
 170         askAndAwait(cohortActor, new Abort(txId));
 171         verify(mockPostCanCommit).abort();
 172
 173         resetMockCohort();
 174         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 175         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
 176     }
 177
 178     @Test
 179     public void testAbortAfterPreCommit() throws Exception {
 180         ActorRef cohortActor = newCohortActor("testAbortAfterPreCommit");
 181
 182         TransactionIdentifier txId = nextTransactionId();
 183         askAndAwait(cohortActor, new CanCommit(txId, CANDIDATES, MOCK_SCHEMA, cohortActor));
 184         verify(mockCohort).canCommit(txId, CANDIDATES, MOCK_SCHEMA);
 185
 186         askAndAwait(cohortActor, new PreCommit(txId));
 187         verify(mockPostCanCommit).preCommit();
 188
 189         askAndAwait(cohortActor, new Abort(txId));
 190         verify(mockPostPreCommit).abort();
 191     }
 192
 193     private <T> ListenableFuture<T> executeWithDelay(ExecutorService executor, T result) {
 194         return JdkFutureAdapters.listenInPoolThread(executor.submit(() -> {
 195             Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
 196             return result;
 197         }), MoreExecutors.directExecutor());
 198     }
 199
 200     private ActorRef newCohortActor(String name) {
 201         return actorFactory.createActor(DataTreeCohortActor.props(mockCohort, YangInstanceIdentifier.EMPTY), name);
 202     }
 203
 204     @SuppressWarnings("unchecked")
 205     private void resetMockCohort() {
 206         reset(mockCohort);
 207         doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostCanCommit).abort();
 208         doReturn(Futures.immediateFuture(mockPostPreCommit)).when(mockPostCanCommit).preCommit();
 209         doReturn(Futures.immediateCheckedFuture(mockPostCanCommit)).when(mockCohort).canCommit(any(Object.class),
 210                 any(Collection.class), any(SchemaContext.class));
 211
 212         doReturn(ThreePhaseCommitStep.NOOP_ABORT_FUTURE).when(mockPostPreCommit).abort();
 213         doReturn(Futures.immediateFuture(null)).when(mockPostPreCommit).commit();
 214     }
 215
 216     private static void askAndAwait(ActorRef actor, CommitProtocolCommand<?> message) throws Exception {
 217         Timeout timeout = new Timeout(5, TimeUnit.SECONDS);
 218         Object result = Await.result(Patterns.ask(actor, message, timeout), timeout.duration());
 219         assertTrue("Expected Success but was " + result, result instanceof Success);
 220         assertEquals("Success", message.getTxId(), ((Success)result).getTxId());
 221     }
 222 }