opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/databroker/ConcurrentDOMDataBrokerTest.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.cluster.databroker;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12 import static org.junit.Assert.assertSame;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.junit.Assert.fail;
  15 import static org.mockito.Mockito.doAnswer;
  16 import static org.mockito.Mockito.doReturn;
  17 import static org.mockito.Mockito.inOrder;
  18 import static org.mockito.Mockito.mock;
  19 import static org.mockito.Mockito.never;
  20 import static org.mockito.Mockito.times;
  21 import static org.mockito.Mockito.verify;
  22 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateFalseFluentFuture;
  23 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateNullFluentFuture;
  24 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateTrueFluentFuture;
  25
  26 import com.google.common.base.Throwables;
  27 import com.google.common.collect.ClassToInstanceMap;
  28 import com.google.common.collect.ImmutableMap;
  29 import com.google.common.util.concurrent.FluentFuture;
  30 import com.google.common.util.concurrent.FutureCallback;
  31 import com.google.common.util.concurrent.Futures;
  32 import com.google.common.util.concurrent.ListenableFuture;
  33 import com.google.common.util.concurrent.MoreExecutors;
  34 import com.google.common.util.concurrent.SettableFuture;
  35 import com.google.common.util.concurrent.Uninterruptibles;
  36 import java.util.ArrayList;
  37 import java.util.List;
  38 import java.util.concurrent.CountDownLatch;
  39 import java.util.concurrent.ExecutionException;
  40 import java.util.concurrent.SynchronousQueue;
  41 import java.util.concurrent.ThreadPoolExecutor;
  42 import java.util.concurrent.TimeUnit;
  43 import java.util.concurrent.TimeoutException;
  44 import java.util.concurrent.atomic.AtomicReference;
  45 import org.junit.After;
  46 import org.junit.Before;
  47 import org.junit.Test;
  48 import org.mockito.InOrder;
  49 import org.mockito.stubbing.Answer;
  50 import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
  51 import org.opendaylight.mdsal.common.api.CommitInfo;
  52 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  53 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  54 import org.opendaylight.mdsal.dom.api.DOMDataBrokerExtension;
  55 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
  56 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
  57 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistry;
  58 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  59 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
  60 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
  61 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  62 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
  63 import org.opendaylight.mdsal.dom.api.DOMTransactionChainListener;
  64 import org.opendaylight.mdsal.dom.broker.TransactionCommitFailedExceptionMapper;
  65 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
  66 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
  67 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
  68 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  69 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
  70 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
  71 import org.opendaylight.mdsal.dom.store.inmemory.InMemoryDOMDataStore;
  72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  73 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  74
  75 /**
  76  * Unit tests for DOMConcurrentDataCommitCoordinator.
  77  *
  78  * @author Thomas Pantelis
  79  */
  80 public class ConcurrentDOMDataBrokerTest {
  81
  82     private final DOMDataTreeWriteTransaction transaction = mock(DOMDataTreeWriteTransaction.class);
  83     private final DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class);
  84     private final ThreadPoolExecutor futureExecutor =
  85             new ThreadPoolExecutor(0, 1, 5, TimeUnit.SECONDS, new SynchronousQueue<>());
  86     private ConcurrentDOMDataBroker coordinator;
  87
  88     @Before
  89     public void setup() {
  90         doReturn("tx").when(transaction).getIdentifier();
  91
  92         DOMStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.newDirectExecutorService());
  93
  94         coordinator = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL, store),
  95                 futureExecutor);
  96     }
  97
  98     @After
  99     public void tearDown() {
 100         futureExecutor.shutdownNow();
 101     }
 102
 103     @Test
 104     public void testSuccessfulSubmitAsync() throws Exception {
 105         testSuccessfulSubmit(true);
 106     }
 107
 108     @Test
 109     public void testSuccessfulSubmitSync() throws Exception {
 110         testSuccessfulSubmit(false);
 111     }
 112
 113     private void testSuccessfulSubmit(final boolean doAsync) throws InterruptedException {
 114         final CountDownLatch asyncCanCommitContinue = new CountDownLatch(1);
 115         Answer<ListenableFuture<Boolean>> asyncCanCommit = invocation -> {
 116             final SettableFuture<Boolean> future = SettableFuture.create();
 117             if (doAsync) {
 118                 new Thread(() -> {
 119                     Uninterruptibles.awaitUninterruptibly(asyncCanCommitContinue, 10, TimeUnit.SECONDS);
 120                     future.set(Boolean.TRUE);
 121                 }).start();
 122             } else {
 123                 future.set(Boolean.TRUE);
 124             }
 125
 126             return future;
 127         };
 128
 129         doAnswer(asyncCanCommit).when(mockCohort).canCommit();
 130         doReturn(immediateNullFluentFuture()).when(mockCohort).preCommit();
 131         doReturn(immediateNullFluentFuture()).when(mockCohort).commit();
 132
 133         ListenableFuture<? extends CommitInfo> future = coordinator.commit(transaction, mockCohort);
 134
 135         final CountDownLatch doneLatch = new CountDownLatch(1);
 136         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 137         Futures.addCallback(future, new FutureCallback<CommitInfo>() {
 138             @Override
 139             public void onSuccess(final CommitInfo result) {
 140                 doneLatch.countDown();
 141             }
 142
 143             @Override
 144             public void onFailure(final Throwable failure) {
 145                 caughtEx.set(failure);
 146                 doneLatch.countDown();
 147             }
 148         }, MoreExecutors.directExecutor());
 149
 150         asyncCanCommitContinue.countDown();
 151
 152         assertTrue("Submit complete", doneLatch.await(5, TimeUnit.SECONDS));
 153
 154         if (caughtEx.get() != null) {
 155             Throwables.throwIfUnchecked(caughtEx.get());
 156             throw new RuntimeException(caughtEx.get());
 157         }
 158
 159         assertEquals("Task count", doAsync ? 1 : 0, futureExecutor.getTaskCount());
 160
 161         InOrder inOrder = inOrder(mockCohort);
 162         inOrder.verify(mockCohort, times(1)).canCommit();
 163         inOrder.verify(mockCohort, times(1)).preCommit();
 164         inOrder.verify(mockCohort, times(1)).commit();
 165     }
 166
 167     @Test
 168     public void testSubmitWithNegativeCanCommitResponse() throws Exception {
 169         doReturn(Futures.immediateFuture(Boolean.FALSE)).when(mockCohort).canCommit();
 170         doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
 171
 172         assertFailure(coordinator.commit(transaction, mockCohort), null, mockCohort);
 173     }
 174
 175     private static void assertFailure(final ListenableFuture<?> future, final Exception expCause,
 176             final DOMStoreThreePhaseCommitCohort mockCohort) throws Exception {
 177         try {
 178             future.get(5, TimeUnit.SECONDS);
 179             fail("Expected TransactionCommitFailedException");
 180         } catch (ExecutionException e) {
 181             TransactionCommitFailedException tcf = TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER.apply(e);
 182             if (expCause != null) {
 183                 assertSame("Expected cause", expCause.getClass(), tcf.getCause().getClass());
 184             }
 185             verify(mockCohort, times(1)).abort();
 186         } catch (TimeoutException e) {
 187             throw e;
 188         }
 189     }
 190
 191     @Test
 192     public void testSubmitWithCanCommitException() throws Exception {
 193         final Exception cause = new IllegalStateException("mock");
 194         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).canCommit();
 195         doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
 196
 197         assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
 198     }
 199
 200     @Test
 201     public void testSubmitWithPreCommitException() throws Exception {
 202         doReturn(immediateTrueFluentFuture()).when(mockCohort).canCommit();
 203         final IllegalStateException cause = new IllegalStateException("mock");
 204         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).preCommit();
 205         doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
 206
 207         assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
 208     }
 209
 210     @Test
 211     public void testSubmitWithCommitException() throws Exception {
 212         doReturn(immediateTrueFluentFuture()).when(mockCohort).canCommit();
 213         doReturn(immediateNullFluentFuture()).when(mockCohort).preCommit();
 214         final IllegalStateException cause = new IllegalStateException("mock");
 215         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).commit();
 216         doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
 217
 218         assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
 219     }
 220
 221     @Test
 222     public void testSubmitWithAbortException() throws Exception {
 223         final Exception canCommitCause = new IllegalStateException("canCommit error");
 224         doReturn(Futures.immediateFailedFuture(canCommitCause)).when(mockCohort).canCommit();
 225         final Exception abortCause = new IllegalStateException("abort error");
 226         doReturn(Futures.immediateFailedFuture(abortCause)).when(mockCohort).abort();
 227
 228         assertFailure(coordinator.commit(transaction, mockCohort), canCommitCause, mockCohort);
 229     }
 230
 231     @Test
 232     public void testCreateReadWriteTransaction() {
 233         DOMStore domStore = mock(DOMStore.class);
 234         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 235                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 236                 futureExecutor)) {
 237             dataBroker.newReadWriteTransaction();
 238
 239             verify(domStore, never()).newReadWriteTransaction();
 240         }
 241     }
 242
 243     @Test
 244     public void testCreateWriteOnlyTransaction() {
 245         DOMStore domStore = mock(DOMStore.class);
 246         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 247                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 248                 futureExecutor)) {
 249             dataBroker.newWriteOnlyTransaction();
 250
 251             verify(domStore, never()).newWriteOnlyTransaction();
 252         }
 253     }
 254
 255     @Test
 256     public void testCreateReadOnlyTransaction() {
 257         DOMStore domStore = mock(DOMStore.class);
 258         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 259                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 260                 futureExecutor)) {
 261             dataBroker.newReadOnlyTransaction();
 262
 263             verify(domStore, never()).newReadOnlyTransaction();
 264         }
 265     }
 266
 267     @Test
 268     public void testLazySubTransactionCreationForReadWriteTransactions() {
 269         DOMStore configDomStore = mock(DOMStore.class);
 270         DOMStore operationalDomStore = mock(DOMStore.class);
 271         DOMStoreReadWriteTransaction storeTxn = mock(DOMStoreReadWriteTransaction.class);
 272
 273         doReturn(storeTxn).when(operationalDomStore).newReadWriteTransaction();
 274         doReturn(storeTxn).when(configDomStore).newReadWriteTransaction();
 275
 276         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 277                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 278                 configDomStore), futureExecutor)) {
 279             DOMDataTreeReadWriteTransaction dataTxn = dataBroker.newReadWriteTransaction();
 280
 281             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
 282             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
 283             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
 284
 285             verify(configDomStore, never()).newReadWriteTransaction();
 286             verify(operationalDomStore, times(1)).newReadWriteTransaction();
 287         }
 288
 289     }
 290
 291     @Test
 292     public void testLazySubTransactionCreationForWriteOnlyTransactions() {
 293         DOMStore configDomStore = mock(DOMStore.class);
 294         DOMStore operationalDomStore = mock(DOMStore.class);
 295         DOMStoreWriteTransaction storeTxn = mock(DOMStoreWriteTransaction.class);
 296
 297         doReturn(storeTxn).when(operationalDomStore).newWriteOnlyTransaction();
 298         doReturn(storeTxn).when(configDomStore).newWriteOnlyTransaction();
 299
 300         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 301                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 302                 configDomStore), futureExecutor)) {
 303             DOMDataTreeWriteTransaction dataTxn = dataBroker.newWriteOnlyTransaction();
 304
 305             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
 306             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
 307
 308             verify(configDomStore, never()).newWriteOnlyTransaction();
 309             verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
 310         }
 311     }
 312
 313     @Test
 314     public void testLazySubTransactionCreationForReadOnlyTransactions() {
 315         DOMStore configDomStore = mock(DOMStore.class);
 316         DOMStore operationalDomStore = mock(DOMStore.class);
 317         DOMStoreReadTransaction storeTxn = mock(DOMStoreReadTransaction.class);
 318
 319         doReturn(storeTxn).when(operationalDomStore).newReadOnlyTransaction();
 320         doReturn(storeTxn).when(configDomStore).newReadOnlyTransaction();
 321
 322         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 323                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 324                 configDomStore), futureExecutor)) {
 325             DOMDataTreeReadTransaction dataTxn = dataBroker.newReadOnlyTransaction();
 326
 327             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
 328             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
 329
 330             verify(configDomStore, never()).newReadOnlyTransaction();
 331             verify(operationalDomStore, times(1)).newReadOnlyTransaction();
 332         }
 333     }
 334
 335     @Test
 336     public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException {
 337         DOMStore configDomStore = mock(DOMStore.class);
 338         DOMStore operationalDomStore = mock(DOMStore.class);
 339         DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class);
 340
 341         doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction();
 342         doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready();
 343         doReturn(immediateFalseFluentFuture()).when(mockCohort).canCommit();
 344         doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
 345
 346         final CountDownLatch latch = new CountDownLatch(1);
 347         final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
 348
 349         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 350                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 351                 configDomStore), futureExecutor) {
 352             @Override
 353             public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx,
 354                     DOMStoreThreePhaseCommitCohort cohort) {
 355                 commitCohorts.add(cohort);
 356                 latch.countDown();
 357                 return super.commit(writeTx, cohort);
 358             }
 359         }) {
 360             DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
 361
 362             domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
 363
 364             domDataReadWriteTransaction.commit();
 365
 366             assertTrue(latch.await(10, TimeUnit.SECONDS));
 367
 368             assertTrue(commitCohorts.size() == 1);
 369         }
 370     }
 371
 372     @Test
 373     public void testCreateTransactionChain() {
 374         DOMStore domStore = mock(DOMStore.class);
 375         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 376                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 377                 futureExecutor)) {
 378
 379             dataBroker.createTransactionChain(mock(DOMTransactionChainListener.class));
 380
 381             verify(domStore, times(2)).createTransactionChain();
 382         }
 383
 384     }
 385
 386     @Test
 387     public void testCreateTransactionOnChain() {
 388         DOMStore domStore = mock(DOMStore.class);
 389         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 390                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 391                 futureExecutor)) {
 392
 393             DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
 394             DOMStoreTransactionChain mockChain = mock(DOMStoreTransactionChain.class);
 395
 396             doReturn(mockChain).when(domStore).createTransactionChain();
 397             doReturn(operationalTransaction).when(mockChain).newWriteOnlyTransaction();
 398
 399             DOMTransactionChain transactionChain = dataBroker.createTransactionChain(
 400                     mock(DOMTransactionChainListener.class));
 401
 402             DOMDataTreeWriteTransaction domDataWriteTransaction = transactionChain.newWriteOnlyTransaction();
 403
 404             verify(mockChain, never()).newWriteOnlyTransaction();
 405
 406             domDataWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(),
 407                     mock(NormalizedNode.class));
 408         }
 409     }
 410
 411     @Test
 412     public void testEmptyTransactionSubmitSucceeds() throws ExecutionException, InterruptedException {
 413         DOMStore domStore = mock(DOMStore.class);
 414         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 415                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 416                 futureExecutor)) {
 417
 418             FluentFuture<? extends CommitInfo> submit1 = dataBroker.newWriteOnlyTransaction().commit();
 419
 420             assertNotNull(submit1);
 421
 422             submit1.get();
 423
 424             FluentFuture<? extends CommitInfo> submit2 = dataBroker.newReadWriteTransaction().commit();
 425
 426             assertNotNull(submit2);
 427
 428             submit2.get();
 429         }
 430     }
 431
 432     @Test
 433     public void testExtensions() {
 434         final var mockConfigStore = mock(AbstractDataStore.class);
 435         final var mockOperStore = mock(AbstractDataStore.class);
 436         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 437                 LogicalDatastoreType.OPERATIONAL, mockOperStore,
 438                 LogicalDatastoreType.CONFIGURATION, mockConfigStore), futureExecutor)) {
 439
 440             ClassToInstanceMap<DOMDataBrokerExtension> supportedExtensions = dataBroker.getExtensions();
 441             assertNotNull(supportedExtensions.getInstance(DOMDataTreeChangeService.class));
 442
 443             DOMDataTreeCommitCohortRegistry cohortRegistry = supportedExtensions.getInstance(
 444                     DOMDataTreeCommitCohortRegistry.class);
 445             assertNotNull(cohortRegistry);
 446
 447             DOMDataTreeCommitCohort cohort = mock(DOMDataTreeCommitCohort.class);
 448             DOMDataTreeIdentifier path = new DOMDataTreeIdentifier(
 449                     org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION,
 450                     YangInstanceIdentifier.empty());
 451             cohortRegistry.registerCommitCohort(path, cohort);
 452
 453             verify(mockConfigStore).registerCommitCohort(path, cohort);
 454         }
 455     }
 456 }