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

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import akka.actor.ActorPath;
   4 import akka.actor.ActorSelection;
   5 import akka.actor.Props;
   6 import akka.dispatch.Futures;
   7 import com.google.common.collect.Lists;
   8 import com.google.common.util.concurrent.ListenableFuture;
   9 import org.junit.Before;
  10 import org.junit.Test;
  11 import org.mockito.Mock;
  12 import org.mockito.MockitoAnnotations;
  13 import org.mockito.stubbing.Stubber;
  14 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
  15 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
  16 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
  17 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
  18 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
  19 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  20 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
  21 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
  22 import org.opendaylight.controller.cluster.datastore.messages.SerializableMessage;
  23 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  24 import org.opendaylight.controller.cluster.datastore.utils.DoNothingActor;
  25 import scala.concurrent.Future;
  26
  27 import java.util.List;
  28 import java.util.concurrent.ExecutionException;
  29 import java.util.concurrent.TimeUnit;
  30
  31 import static org.junit.Assert.assertEquals;
  32 import static org.junit.Assert.fail;
  33 import static org.mockito.Mockito.any;
  34 import static org.mockito.Mockito.doReturn;
  35 import static org.mockito.Mockito.isA;
  36 import static org.mockito.Mockito.times;
  37 import static org.mockito.Mockito.verify;
  38
  39 public class ThreePhaseCommitCohortProxyTest extends AbstractActorTest {
  40
  41     @SuppressWarnings("serial")
  42     static class TestException extends RuntimeException {
  43     }
  44
  45     @Mock
  46     private ActorContext actorContext;
  47
  48     @Before
  49     public void setUp() {
  50         MockitoAnnotations.initMocks(this);
  51
  52         doReturn(getSystem()).when(actorContext).getActorSystem();
  53     }
  54
  55     private Future<ActorSelection> newCohort() {
  56         ActorPath path = getSystem().actorOf(Props.create(DoNothingActor.class)).path();
  57         ActorSelection actorSelection = getSystem().actorSelection(path);
  58         return Futures.successful(actorSelection);
  59     }
  60
  61     private final ThreePhaseCommitCohortProxy setupProxy(int nCohorts) throws Exception {
  62         List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
  63         for(int i = 1; i <= nCohorts; i++) {
  64             cohortFutures.add(newCohort());
  65         }
  66
  67         return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures, "txn-1");
  68     }
  69
  70     private ThreePhaseCommitCohortProxy setupProxyWithFailedCohortPath()
  71             throws Exception {
  72         List<Future<ActorSelection>> cohortFutures = Lists.newArrayList();
  73         cohortFutures.add(newCohort());
  74         cohortFutures.add(Futures.<ActorSelection>failed(new TestException()));
  75
  76         return new ThreePhaseCommitCohortProxy(actorContext, cohortFutures, "txn-1");
  77     }
  78
  79     private void setupMockActorContext(Class<?> requestType, Object... responses) {
  80         Stubber stubber = doReturn(responses[0] instanceof Throwable ? Futures
  81                 .failed((Throwable) responses[0]) : Futures
  82                 .successful(((SerializableMessage) responses[0]).toSerializable()));
  83
  84         for(int i = 1; i < responses.length; i++) {
  85             stubber = stubber.doReturn(responses[i] instanceof Throwable ? Futures
  86                     .failed((Throwable) responses[i]) : Futures
  87                     .successful(((SerializableMessage) responses[i]).toSerializable()));
  88         }
  89
  90         stubber.when(actorContext).executeOperationAsync(any(ActorSelection.class),
  91                 isA(requestType));
  92     }
  93
  94     private void verifyCohortInvocations(int nCohorts, Class<?> requestType) {
  95         verify(actorContext, times(nCohorts)).executeOperationAsync(
  96                 any(ActorSelection.class), isA(requestType));
  97     }
  98
  99     private void propagateExecutionExceptionCause(ListenableFuture<?> future) throws Throwable {
 100
 101         try {
 102             future.get(5, TimeUnit.SECONDS);
 103             fail("Expected ExecutionException");
 104         } catch(ExecutionException e) {
 105             throw e.getCause();
 106         }
 107     }
 108
 109     @Test
 110     public void testCanCommitWithOneCohort() throws Exception {
 111
 112         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 113
 114         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 115                 new CanCommitTransactionReply(true));
 116
 117         ListenableFuture<Boolean> future = proxy.canCommit();
 118
 119         assertEquals("canCommit", true, future.get(5, TimeUnit.SECONDS));
 120
 121         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 122                 new CanCommitTransactionReply(false));
 123
 124         future = proxy.canCommit();
 125
 126         assertEquals("canCommit", false, future.get(5, TimeUnit.SECONDS));
 127
 128         verifyCohortInvocations(2, CanCommitTransaction.SERIALIZABLE_CLASS);
 129     }
 130
 131     @Test
 132     public void testCanCommitWithMultipleCohorts() throws Exception {
 133
 134         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 135
 136         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 137                 new CanCommitTransactionReply(true), new CanCommitTransactionReply(true));
 138
 139         ListenableFuture<Boolean> future = proxy.canCommit();
 140
 141         assertEquals("canCommit", true, future.get(5, TimeUnit.SECONDS));
 142
 143         verifyCohortInvocations(2, CanCommitTransaction.SERIALIZABLE_CLASS);
 144     }
 145
 146     @Test
 147     public void testCanCommitWithMultipleCohortsAndOneFailure() throws Exception {
 148
 149         ThreePhaseCommitCohortProxy proxy = setupProxy(3);
 150
 151         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 152                 new CanCommitTransactionReply(true), new CanCommitTransactionReply(false),
 153                 new CanCommitTransactionReply(true));
 154
 155         ListenableFuture<Boolean> future = proxy.canCommit();
 156
 157         assertEquals("canCommit", false, future.get(5, TimeUnit.SECONDS));
 158
 159         verifyCohortInvocations(3, CanCommitTransaction.SERIALIZABLE_CLASS);
 160     }
 161
 162     @Test(expected = TestException.class)
 163     public void testCanCommitWithExceptionFailure() throws Throwable {
 164
 165         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 166
 167         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS, new TestException());
 168
 169         propagateExecutionExceptionCause(proxy.canCommit());
 170     }
 171
 172     @Test(expected = ExecutionException.class)
 173     public void testCanCommitWithInvalidResponseType() throws Exception {
 174
 175         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 176
 177         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 178                 new PreCommitTransactionReply());
 179
 180         proxy.canCommit().get(5, TimeUnit.SECONDS);
 181     }
 182
 183     @Test(expected = TestException.class)
 184     public void testCanCommitWithFailedCohortPath() throws Throwable {
 185
 186         ThreePhaseCommitCohortProxy proxy = setupProxyWithFailedCohortPath();
 187
 188         try {
 189             propagateExecutionExceptionCause(proxy.canCommit());
 190         } finally {
 191             verifyCohortInvocations(0, CanCommitTransaction.SERIALIZABLE_CLASS);
 192         }
 193     }
 194
 195     @Test
 196     public void testPreCommit() throws Exception {
 197         // Precommit is currently a no-op
 198         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 199
 200         setupMockActorContext(PreCommitTransaction.SERIALIZABLE_CLASS,
 201                 new PreCommitTransactionReply());
 202
 203         proxy.preCommit().get(5, TimeUnit.SECONDS);
 204     }
 205
 206     @Test
 207     public void testAbort() throws Exception {
 208         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 209
 210         setupMockActorContext(AbortTransaction.SERIALIZABLE_CLASS, new AbortTransactionReply());
 211
 212         proxy.abort().get(5, TimeUnit.SECONDS);
 213
 214         verifyCohortInvocations(1, AbortTransaction.SERIALIZABLE_CLASS);
 215     }
 216
 217     @Test
 218     public void testAbortWithFailure() throws Exception {
 219         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 220
 221         setupMockActorContext(AbortTransaction.SERIALIZABLE_CLASS, new RuntimeException("mock"));
 222
 223         // The exception should not get propagated.
 224         proxy.abort().get(5, TimeUnit.SECONDS);
 225
 226         verifyCohortInvocations(1, AbortTransaction.SERIALIZABLE_CLASS);
 227     }
 228
 229     @Test
 230     public void testAbortWithFailedCohortPath() throws Throwable {
 231
 232         ThreePhaseCommitCohortProxy proxy = setupProxyWithFailedCohortPath();
 233
 234         // The exception should not get propagated.
 235         proxy.abort().get(5, TimeUnit.SECONDS);
 236
 237         verifyCohortInvocations(0, AbortTransaction.SERIALIZABLE_CLASS);
 238     }
 239
 240     @Test
 241     public void testCommit() throws Exception {
 242
 243         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 244
 245         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new CommitTransactionReply(),
 246                 new CommitTransactionReply());
 247
 248         proxy.commit().get(5, TimeUnit.SECONDS);
 249
 250         verifyCohortInvocations(2, CommitTransaction.SERIALIZABLE_CLASS);
 251     }
 252
 253     @Test(expected = TestException.class)
 254     public void testCommitWithFailure() throws Throwable {
 255
 256         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 257
 258         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new CommitTransactionReply(),
 259                 new TestException());
 260
 261         propagateExecutionExceptionCause(proxy.commit());
 262     }
 263
 264     @Test(expected = ExecutionException.class)
 265     public void testCommitWithInvalidResponseType() throws Exception {
 266
 267         ThreePhaseCommitCohortProxy proxy = setupProxy(1);
 268
 269         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS, new PreCommitTransactionReply());
 270
 271         proxy.commit().get(5, TimeUnit.SECONDS);
 272     }
 273
 274     @Test(expected = TestException.class)
 275     public void testCommitWithFailedCohortPath() throws Throwable {
 276
 277         ThreePhaseCommitCohortProxy proxy = setupProxyWithFailedCohortPath();
 278
 279         try {
 280             propagateExecutionExceptionCause(proxy.commit());
 281         } finally {
 282             verifyCohortInvocations(0, CommitTransaction.SERIALIZABLE_CLASS);
 283         }
 284     }
 285
 286     @Test
 287     public void testAllThreePhasesSuccessful() throws Exception {
 288
 289         ThreePhaseCommitCohortProxy proxy = setupProxy(2);
 290
 291         setupMockActorContext(CanCommitTransaction.SERIALIZABLE_CLASS,
 292                 new CanCommitTransactionReply(true), new CanCommitTransactionReply(true));
 293
 294         setupMockActorContext(PreCommitTransaction.SERIALIZABLE_CLASS,
 295                 new PreCommitTransactionReply(), new PreCommitTransactionReply());
 296
 297         setupMockActorContext(CommitTransaction.SERIALIZABLE_CLASS,
 298                 new CommitTransactionReply(), new CommitTransactionReply());
 299
 300         proxy.canCommit().get(5, TimeUnit.SECONDS);
 301         proxy.preCommit().get(5, TimeUnit.SECONDS);
 302         proxy.commit().get(5, TimeUnit.SECONDS);
 303
 304         verifyCohortInvocations(2, CanCommitTransaction.SERIALIZABLE_CLASS);
 305         verifyCohortInvocations(2, CommitTransaction.SERIALIZABLE_CLASS);
 306     }
 307 }