opendaylight/md-sal/sal-dom-broker/src/test/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMConcurrentDataCommitCoordinatorTest.java

   1 /*
   2  * Copyright (c) 2014 Brocade Communications 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 package org.opendaylight.controller.md.sal.dom.broker.impl;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertSame;
  12 import static org.junit.Assert.fail;
  13 import static org.mockito.Mockito.mock;
  14 import static org.mockito.Mockito.doReturn;
  15 import static org.mockito.Mockito.doAnswer;
  16 import static org.mockito.Mockito.inOrder;
  17 import java.util.Arrays;
  18 import java.util.concurrent.CountDownLatch;
  19 import java.util.concurrent.SynchronousQueue;
  20 import java.util.concurrent.ThreadPoolExecutor;
  21 import java.util.concurrent.TimeUnit;
  22 import java.util.concurrent.TimeoutException;
  23 import java.util.concurrent.atomic.AtomicReference;
  24 import org.junit.After;
  25 import org.junit.Before;
  26 import org.junit.Test;
  27 import org.mockito.InOrder;
  28 import org.mockito.invocation.InvocationOnMock;
  29 import org.mockito.stubbing.Answer;
  30 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  31 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  33 import com.google.common.util.concurrent.CheckedFuture;
  34 import com.google.common.util.concurrent.FutureCallback;
  35 import com.google.common.util.concurrent.Futures;
  36 import com.google.common.util.concurrent.ListenableFuture;
  37 import com.google.common.util.concurrent.SettableFuture;
  38 import com.google.common.util.concurrent.Uninterruptibles;
  39
  40 /**
  41  * Unit tests for DOMConcurrentDataCommitCoordinator.
  42  *
  43  * @author Thomas Pantelis
  44  */
  45 public class DOMConcurrentDataCommitCoordinatorTest {
  46
  47     private final DOMDataWriteTransaction transaction = mock(DOMDataWriteTransaction.class);
  48     private final DOMStoreThreePhaseCommitCohort mockCohort1 = mock(DOMStoreThreePhaseCommitCohort.class);
  49     private final DOMStoreThreePhaseCommitCohort mockCohort2 = mock(DOMStoreThreePhaseCommitCohort.class);
  50     private final ThreadPoolExecutor futureExecutor =
  51             new ThreadPoolExecutor(0, 1, 5, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
  52     private final DOMConcurrentDataCommitCoordinator coordinator =
  53             new DOMConcurrentDataCommitCoordinator(futureExecutor);
  54
  55     @Before
  56     public void setup() {
  57         doReturn("tx").when(transaction).getIdentifier();
  58     }
  59
  60     @After
  61     public void tearDown() {
  62         futureExecutor.shutdownNow();
  63     }
  64
  65     @Test
  66     public void testSuccessfulSubmitAsync() throws Throwable {
  67         testSuccessfulSubmit(true);
  68     }
  69
  70     @Test
  71     public void testSuccessfulSubmitSync() throws Throwable {
  72         testSuccessfulSubmit(false);
  73     }
  74
  75     private void testSuccessfulSubmit(final boolean doAsync) throws Throwable {
  76         final CountDownLatch asyncCanCommitContinue = new CountDownLatch(1);
  77         Answer<ListenableFuture<Boolean>> asyncCanCommit = new Answer<ListenableFuture<Boolean>>() {
  78             @Override
  79             public ListenableFuture<Boolean> answer(InvocationOnMock invocation) {
  80                 final SettableFuture<Boolean> future = SettableFuture.create();
  81                 if(doAsync) {
  82                     new Thread() {
  83                         @Override
  84                         public void run() {
  85                             Uninterruptibles.awaitUninterruptibly(asyncCanCommitContinue,
  86                                     10, TimeUnit.SECONDS);
  87                             future.set(true);
  88                         }
  89                     }.start();
  90                 } else {
  91                     future.set(true);
  92                 }
  93
  94                 return future;
  95             }
  96         };
  97
  98         doAnswer(asyncCanCommit).when(mockCohort1).canCommit();
  99         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 100         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
 101
 102         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 103         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
 104         doReturn(Futures.immediateFuture(null)).when(mockCohort2).commit();
 105
 106         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 107                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 108
 109         final CountDownLatch doneLatch = new CountDownLatch(1);
 110         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 111         Futures.addCallback(future, new FutureCallback<Void>() {
 112             @Override
 113             public void onSuccess(Void result) {
 114                 doneLatch.countDown();
 115             }
 116
 117             @Override
 118             public void onFailure(Throwable t) {
 119                 caughtEx.set(t);
 120                 doneLatch.countDown();
 121             }
 122         });
 123
 124         asyncCanCommitContinue.countDown();
 125
 126         assertEquals("Submit complete", true, doneLatch.await(5, TimeUnit.SECONDS));
 127
 128         if(caughtEx.get() != null) {
 129             throw caughtEx.get();
 130         }
 131
 132         assertEquals("Task count", doAsync ? 1 : 0, futureExecutor.getTaskCount());
 133
 134         InOrder inOrder = inOrder(mockCohort1, mockCohort2);
 135         inOrder.verify(mockCohort1).canCommit();
 136         inOrder.verify(mockCohort2).canCommit();
 137         inOrder.verify(mockCohort1).preCommit();
 138         inOrder.verify(mockCohort2).preCommit();
 139         inOrder.verify(mockCohort1).commit();
 140         inOrder.verify(mockCohort2).commit();
 141     }
 142
 143     @Test
 144     public void testSubmitWithNegativeCanCommitResponse() throws Exception {
 145         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 146         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 147
 148         doReturn(Futures.immediateFuture(false)).when(mockCohort2).canCommit();
 149         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 150
 151         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 152         doReturn(Futures.immediateFuture(false)).when(mockCohort3).canCommit();
 153         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 154
 155         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 156                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 157
 158         assertFailure(future, null, mockCohort1, mockCohort2, mockCohort3);
 159     }
 160
 161     private void assertFailure(CheckedFuture<Void, TransactionCommitFailedException> future,
 162             Exception expCause, DOMStoreThreePhaseCommitCohort... mockCohorts)
 163                     throws Exception {
 164         try {
 165             future.checkedGet(5, TimeUnit.SECONDS);
 166             fail("Expected TransactionCommitFailedException");
 167         } catch (TransactionCommitFailedException e) {
 168             if(expCause != null) {
 169                 assertSame("Expected cause", expCause, e.getCause());
 170             }
 171
 172             InOrder inOrder = inOrder((Object[])mockCohorts);
 173             for(DOMStoreThreePhaseCommitCohort c: mockCohorts) {
 174                 inOrder.verify(c).abort();
 175             }
 176         } catch (TimeoutException e) {
 177             throw e;
 178         }
 179     }
 180
 181     @Test
 182     public void testSubmitWithCanCommitException() throws Exception {
 183         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 184         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 185
 186         IllegalStateException cause = new IllegalStateException("mock");
 187         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
 188         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 189
 190         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 191                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 192
 193         assertFailure(future, cause, mockCohort1, mockCohort2);
 194     }
 195
 196     @Test
 197     public void testSubmitWithPreCommitException() throws Exception {
 198         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 199         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 200         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 201
 202         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 203         IllegalStateException cause = new IllegalStateException("mock");
 204         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).preCommit();
 205         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 206
 207         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 208         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
 209         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2"))).
 210                 when(mockCohort3).preCommit();
 211         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 212
 213         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 214                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 215
 216         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
 217     }
 218
 219     @Test
 220     public void testSubmitWithCommitException() throws Exception {
 221         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 222         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 223         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
 224         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 225
 226         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 227         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
 228         IllegalStateException cause = new IllegalStateException("mock");
 229         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).commit();
 230         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 231
 232         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 233         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
 234         doReturn(Futures.immediateFuture(null)).when(mockCohort3).preCommit();
 235         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2"))).
 236                 when(mockCohort3).commit();
 237         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 238
 239         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 240                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 241
 242         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
 243     }
 244
 245     @Test
 246     public void testSubmitWithAbortException() throws Exception {
 247         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 248         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock abort error"))).
 249                 when(mockCohort1).abort();
 250
 251         IllegalStateException cause = new IllegalStateException("mock canCommit error");
 252         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
 253         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 254
 255         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 256                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 257
 258         assertFailure(future, cause, mockCohort1, mockCohort2);
 259     }
 260 }