opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxyTest.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.fail;
  14 import static org.mockito.Mockito.doReturn;
  15 import static org.opendaylight.controller.cluster.datastore.DataStoreVersions.CURRENT_VERSION;
  16 import akka.actor.ActorSelection;
  17 import akka.actor.Props;
  18 import akka.actor.UntypedActor;
  19 import akka.dispatch.Dispatchers;
  20 import akka.dispatch.Futures;
  21 import akka.testkit.TestActorRef;
  22 import com.codahale.metrics.Snapshot;
  23 import com.codahale.metrics.Timer;
  24 import com.google.common.base.Preconditions;
  25 import com.google.common.base.Supplier;
  26 import com.google.common.util.concurrent.ListenableFuture;
  27 import java.util.ArrayList;
  28 import java.util.Arrays;
  29 import java.util.Collections;
  30 import java.util.List;
  31 import java.util.concurrent.ExecutionException;
  32 import java.util.concurrent.TimeUnit;
  33 import java.util.concurrent.atomic.AtomicInteger;
  34 import org.junit.Before;
  35 import org.junit.Test;
  36 import org.mockito.Mock;
  37 import org.mockito.MockitoAnnotations;
  38 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  39 import org.opendaylight.controller.cluster.datastore.ThreePhaseCommitCohortProxy.CohortInfo;
  40 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  41 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  42 import org.opendaylight.controller.cluster.datastore.messages.AbstractThreePhaseCommitMessage;
  43 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  44 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  45 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  46 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  47 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  48 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
  49 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
  50 import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
  51 import org.opendaylight.controller.cluster.raft.TestActorFactory;
  52 import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
  53
  54 public class ThreePhaseCommitCohortProxyTest extends AbstractActorTest {
  55
  56     @SuppressWarnings("serial")
  57     static class TestException extends RuntimeException {
  58     }
  59
  60     private ActorContext actorContext;
  61
  62     @Mock
  63     private Timer commitTimer;
  64
  65     @Mock
  66     private Timer.Context commitTimerContext;
  67
  68     @Mock
  69     private Snapshot commitSnapshot;
  70
  71     private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
  72     private final List<TestActorRef<CohortActor>> cohortActors = new ArrayList<>();
  73     private final TransactionIdentifier tx = nextTransactionId();
  74
  75
  76     @Before
  77     public void setUp() {
  78         MockitoAnnotations.initMocks(this);
  79
  80         actorContext = new ActorContext(getSystem(), actorFactory.createActor(Props.create(DoNothingActor.class)),
  81                 new MockClusterWrapper(), new MockConfiguration(),
  82                 DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache()) {
  83                     @Override
  84                     public Timer getOperationTimer(String operationName) {
  85                         return commitTimer;
  86                     }
  87
  88                     @Override
  89                     public double getTxCreationLimit() {
  90                         return 10.0;
  91                     }
  92                 };
  93
  94         doReturn(commitTimerContext).when(commitTimer).time();
  95         doReturn(commitSnapshot).when(commitTimer).getSnapshot();
  96         for(int i=1;i<11;i++){
  97             // Keep on increasing the amount of time it takes to complete transaction for each tenth of a
  98             // percentile. Essentially this would be 1ms for the 10th percentile, 2ms for 20th percentile and so on.
  99             doReturn(TimeUnit.MILLISECONDS.toNanos(i) * 1D).when(commitSnapshot).getValue(i * 0.1);
 100         }
 101     }
 102
 103     @Test
 104     public void testCanCommitYesWithOneCohort() throws Exception {
 105         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 106                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 107                         CanCommitTransactionReply.yes(CURRENT_VERSION)))), tx);
 108
 109         verifyCanCommit(proxy.canCommit(), true);
 110         verifyCohortActors();
 111     }
 112
 113     @Test
 114     public void testCanCommitNoWithOneCohort() throws Exception {
 115         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 116                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 117                         CanCommitTransactionReply.no(CURRENT_VERSION)))), tx);
 118
 119         verifyCanCommit(proxy.canCommit(), false);
 120         verifyCohortActors();
 121     }
 122
 123     @Test
 124     public void testCanCommitYesWithTwoCohorts() throws Exception {
 125         List<CohortInfo> cohorts = Arrays.asList(
 126                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 127                         CanCommitTransactionReply.yes(CURRENT_VERSION))),
 128                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 129                         CanCommitTransactionReply.yes(CURRENT_VERSION))));
 130         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 131
 132         verifyCanCommit(proxy.canCommit(), true);
 133         verifyCohortActors();
 134     }
 135
 136     @Test
 137     public void testCanCommitNoWithThreeCohorts() throws Exception {
 138         List<CohortInfo> cohorts = Arrays.asList(
 139                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 140                         CanCommitTransactionReply.yes(CURRENT_VERSION))),
 141                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
 142                         CanCommitTransactionReply.no(CURRENT_VERSION))),
 143                 newCohortInfo(new CohortActor.Builder(tx)));
 144         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 145
 146         verifyCanCommit(proxy.canCommit(), false);
 147         verifyCohortActors();
 148     }
 149
 150     @Test(expected = TestException.class)
 151     public void testCanCommitWithExceptionFailure() throws Throwable {
 152         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 153                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(new TestException()))), tx);
 154
 155         propagateExecutionExceptionCause(proxy.canCommit());
 156     }
 157
 158     @Test(expected = IllegalArgumentException.class)
 159     public void testCanCommitWithInvalidResponseType() throws Throwable {
 160         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 161                 newCohortInfo(new CohortActor.Builder(tx).expectCanCommit("invalid"))), tx);
 162
 163         propagateExecutionExceptionCause(proxy.canCommit());
 164     }
 165
 166     @Test(expected = TestException.class)
 167     public void testCanCommitWithFailedCohortFuture() throws Throwable {
 168         List<CohortInfo> cohorts = Arrays.asList(
 169                 newCohortInfo(new CohortActor.Builder(tx)),
 170                 newCohortInfoWithFailedFuture(new TestException()),
 171                 newCohortInfo(new CohortActor.Builder(tx)));
 172         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 173
 174         propagateExecutionExceptionCause(proxy.canCommit());
 175     }
 176
 177     @Test
 178     public void testAllThreePhasesSuccessful() throws Exception {
 179         List<CohortInfo> cohorts = Arrays.asList(
 180                 newCohortInfo(new CohortActor.Builder(tx).
 181                         expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).
 182                         expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))),
 183                 newCohortInfo(new CohortActor.Builder(tx).
 184                         expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).
 185                         expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))));
 186         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 187
 188         verifyCanCommit(proxy.canCommit(), true);
 189         verifySuccessfulFuture(proxy.preCommit());
 190         verifySuccessfulFuture(proxy.commit());
 191         verifyCohortActors();
 192     }
 193
 194     @Test(expected = TestException.class)
 195     public void testCommitWithExceptionFailure() throws Throwable {
 196         List<CohortInfo> cohorts = Arrays.asList(
 197                 newCohortInfo(new CohortActor.Builder(tx).
 198                         expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).
 199                         expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))),
 200                 newCohortInfo(new CohortActor.Builder(tx).
 201                         expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).
 202                         expectCommit(new TestException())));
 203         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 204
 205         verifyCanCommit(proxy.canCommit(), true);
 206         verifySuccessfulFuture(proxy.preCommit());
 207         propagateExecutionExceptionCause(proxy.commit());
 208     }
 209
 210     @Test(expected = IllegalArgumentException.class)
 211     public void testCommitWithInvalidResponseType() throws Throwable {
 212         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 213                 newCohortInfo(new CohortActor.Builder(tx).
 214                         expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).
 215                         expectCommit("invalid"))), tx);
 216
 217         verifyCanCommit(proxy.canCommit(), true);
 218         verifySuccessfulFuture(proxy.preCommit());
 219         propagateExecutionExceptionCause(proxy.commit());
 220     }
 221
 222     @Test
 223     public void testAbort() throws Exception {
 224         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 225                 newCohortInfo(new CohortActor.Builder(tx).expectAbort(
 226                         AbortTransactionReply.instance(CURRENT_VERSION)))), tx);
 227
 228         verifySuccessfulFuture(proxy.abort());
 229         verifyCohortActors();
 230     }
 231
 232     @Test
 233     public void testAbortWithFailure() throws Exception {
 234         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
 235                 newCohortInfo(new CohortActor.Builder(tx).expectAbort(new RuntimeException("mock")))), tx);
 236
 237         // The exception should not get propagated.
 238         verifySuccessfulFuture(proxy.abort());
 239         verifyCohortActors();
 240     }
 241
 242     @Test
 243     public void testAbortWithFailedCohortFuture() throws Throwable {
 244         List<CohortInfo> cohorts = Arrays.asList(
 245                 newCohortInfoWithFailedFuture(new TestException()), newCohortInfo(new CohortActor.Builder(tx)));
 246         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
 247
 248         verifySuccessfulFuture(proxy.abort());
 249         verifyCohortActors();
 250     }
 251
 252     @Test
 253     public void testWithNoCohorts() throws Exception {
 254         ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext,
 255                 Collections.<CohortInfo>emptyList(), tx);
 256
 257         verifyCanCommit(proxy.canCommit(), true);
 258         verifySuccessfulFuture(proxy.preCommit());
 259         verifySuccessfulFuture(proxy.commit());
 260         verifyCohortActors();
 261     }
 262
 263     private void propagateExecutionExceptionCause(ListenableFuture<?> future) throws Throwable {
 264
 265         try {
 266             future.get(5, TimeUnit.SECONDS);
 267             fail("Expected ExecutionException");
 268         } catch(ExecutionException e) {
 269             verifyCohortActors();
 270             throw e.getCause();
 271         }
 272     }
 273
 274     private CohortInfo newCohortInfo(CohortActor.Builder builder, final short version) {
 275         TestActorRef<CohortActor> actor = actorFactory.createTestActor(builder.props().
 276                 withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("cohort"));
 277         cohortActors.add(actor);
 278         return new CohortInfo(Futures.successful(getSystem().actorSelection(actor.path())), new Supplier<Short>() {
 279             @Override
 280             public Short get() {
 281                 return version;
 282             }
 283         });
 284     }
 285
 286     private static CohortInfo newCohortInfoWithFailedFuture(Exception failure) {
 287         return new CohortInfo(Futures.<ActorSelection>failed(failure), new Supplier<Short>() {
 288             @Override
 289             public Short get() {
 290                 return CURRENT_VERSION;
 291             }
 292         });
 293     }
 294
 295     private CohortInfo newCohortInfo(CohortActor.Builder builder) {
 296         return newCohortInfo(builder, CURRENT_VERSION);
 297     }
 298
 299     private void verifyCohortActors() {
 300         for(TestActorRef<CohortActor> actor: cohortActors) {
 301             actor.underlyingActor().verify();
 302         }
 303     }
 304
 305     private <T> T verifySuccessfulFuture(ListenableFuture<T> future) throws Exception {
 306         try {
 307             return future.get(5, TimeUnit.SECONDS);
 308         } catch(Exception e) {
 309             verifyCohortActors();
 310             throw e;
 311         }
 312     }
 313
 314     private void verifyCanCommit(ListenableFuture<Boolean> future, boolean expected) throws Exception {
 315         Boolean actual = verifySuccessfulFuture(future);
 316         assertEquals("canCommit", expected, actual);
 317     }
 318
 319     private static class CohortActor extends UntypedActor {
 320         private final Builder builder;
 321         private final AtomicInteger canCommitCount = new AtomicInteger();
 322         private final AtomicInteger commitCount = new AtomicInteger();
 323         private final AtomicInteger abortCount = new AtomicInteger();
 324         private volatile AssertionError assertionError;
 325
 326         private CohortActor(Builder builder) {
 327             this.builder = builder;
 328         }
 329
 330         @Override
 331         public void onReceive(Object message) {
 332             if(CanCommitTransaction.isSerializedType(message)) {
 333                 canCommitCount.incrementAndGet();
 334                 onMessage("CanCommitTransaction", message, CanCommitTransaction.fromSerializable(message),
 335                         builder.expCanCommitType, builder.canCommitReply);
 336             } else if(CommitTransaction.isSerializedType(message)) {
 337                 commitCount.incrementAndGet();
 338                 onMessage("CommitTransaction", message, CommitTransaction.fromSerializable(message),
 339                         builder.expCommitType, builder.commitReply);
 340             } else if(AbortTransaction.isSerializedType(message)) {
 341                 abortCount.incrementAndGet();
 342                 onMessage("AbortTransaction", message, AbortTransaction.fromSerializable(message),
 343                         builder.expAbortType, builder.abortReply);
 344             } else {
 345                 assertionError = new AssertionError("Unexpected message " + message);
 346             }
 347         }
 348
 349         private void onMessage(String name, Object rawMessage, AbstractThreePhaseCommitMessage actualMessage,
 350                 Class<?> expType, Object reply) {
 351             try {
 352                 assertNotNull("Unexpected " + name, expType);
 353                 assertEquals(name + " type", expType, rawMessage.getClass());
 354                 assertEquals(name + " transactionId", builder.transactionId, actualMessage.getTransactionID());
 355
 356                 if(reply instanceof Throwable) {
 357                     getSender().tell(new akka.actor.Status.Failure((Throwable)reply), self());
 358                 } else {
 359                     getSender().tell(reply, self());
 360                 }
 361             } catch(AssertionError e) {
 362                 assertionError = e;
 363             }
 364         }
 365
 366         void verify() {
 367             if(assertionError != null) {
 368                 throw assertionError;
 369             }
 370
 371             if(builder.expCanCommitType != null) {
 372                 assertEquals("CanCommitTransaction count", 1, canCommitCount.get());
 373             }
 374
 375             if(builder.expCommitType != null) {
 376                 assertEquals("CommitTransaction count", 1, commitCount.get());
 377             }
 378
 379             if(builder.expAbortType != null) {
 380                 assertEquals("AbortTransaction count", 1, abortCount.get());
 381             }
 382         }
 383
 384         static class Builder {
 385             private Class<?> expCanCommitType;
 386             private Class<?> expCommitType;
 387             private Class<?> expAbortType;
 388             private Object canCommitReply;
 389             private Object commitReply;
 390             private Object abortReply;
 391             private final TransactionIdentifier transactionId;
 392
 393             Builder(TransactionIdentifier transactionId) {
 394                 this.transactionId = Preconditions.checkNotNull(transactionId);
 395             }
 396
 397             Builder expectCanCommit(Class<?> expCanCommitType, Object canCommitReply) {
 398                 this.expCanCommitType = expCanCommitType;
 399                 this.canCommitReply = canCommitReply;
 400                 return this;
 401             }
 402
 403             Builder expectCanCommit(Object canCommitReply) {
 404                 return expectCanCommit(CanCommitTransaction.class, canCommitReply);
 405             }
 406
 407             Builder expectCommit(Class<?> expCommitType, Object commitReply) {
 408                 this.expCommitType = expCommitType;
 409                 this.commitReply = commitReply;
 410                 return this;
 411             }
 412
 413             Builder expectCommit(Object commitReply) {
 414                 return expectCommit(CommitTransaction.class, commitReply);
 415             }
 416
 417             Builder expectAbort(Class<?> expAbortType, Object abortReply) {
 418                 this.expAbortType = expAbortType;
 419                 this.abortReply = abortReply;
 420                 return this;
 421             }
 422
 423             Builder expectAbort(Object abortReply) {
 424                 return expectAbort(AbortTransaction.class, abortReply);
 425             }
 426
 427             Props props() {
 428                 return Props.create(CohortActor.class, this);
 429             }
 430         }
 431     }
 432 }