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
  23 import com.google.common.base.Throwables;
  24 import com.google.common.collect.ImmutableMap;
  25 import com.google.common.util.concurrent.CheckedFuture;
  26 import com.google.common.util.concurrent.FutureCallback;
  27 import com.google.common.util.concurrent.Futures;
  28 import com.google.common.util.concurrent.ListenableFuture;
  29 import com.google.common.util.concurrent.MoreExecutors;
  30 import com.google.common.util.concurrent.SettableFuture;
  31 import com.google.common.util.concurrent.Uninterruptibles;
  32 import java.util.ArrayList;
  33 import java.util.Arrays;
  34 import java.util.Collection;
  35 import java.util.List;
  36 import java.util.Map;
  37 import java.util.concurrent.CountDownLatch;
  38 import java.util.concurrent.ExecutionException;
  39 import java.util.concurrent.SynchronousQueue;
  40 import java.util.concurrent.ThreadPoolExecutor;
  41 import java.util.concurrent.TimeUnit;
  42 import java.util.concurrent.TimeoutException;
  43 import java.util.concurrent.atomic.AtomicReference;
  44 import org.junit.After;
  45 import org.junit.Before;
  46 import org.junit.Test;
  47 import org.mockito.InOrder;
  48 import org.mockito.stubbing.Answer;
  49 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  50 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  51 import org.opendaylight.controller.md.sal.common.api.data.DataStoreUnavailableException;
  52 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  53 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  54 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  55 import org.opendaylight.controller.md.sal.dom.api.DOMDataBrokerExtension;
  56 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
  57 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadWriteTransaction;
  58 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeService;
  59 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeCommitCohortRegistry;
  60 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  61 import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain;
  62 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
  63 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  64 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  65 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  66 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  67 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  68 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  69 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
  70 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  71 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  72 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  73
  74 /**
  75  * Unit tests for DOMConcurrentDataCommitCoordinator.
  76  *
  77  * @author Thomas Pantelis
  78  */
  79 public class ConcurrentDOMDataBrokerTest {
  80
  81     private final DOMDataWriteTransaction transaction = mock(DOMDataWriteTransaction.class);
  82     private final DOMStoreThreePhaseCommitCohort mockCohort1 = mock(DOMStoreThreePhaseCommitCohort.class);
  83     private final DOMStoreThreePhaseCommitCohort mockCohort2 = 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",
  93             MoreExecutors.newDirectExecutorService());
  94
  95         coordinator = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL, store),
  96                 futureExecutor);
  97     }
  98
  99     @After
 100     public void tearDown() {
 101         futureExecutor.shutdownNow();
 102     }
 103
 104     @Test
 105     public void testSuccessfulSubmitAsync() throws Exception {
 106         testSuccessfulSubmit(true);
 107     }
 108
 109     @Test
 110     public void testSuccessfulSubmitSync() throws Exception {
 111         testSuccessfulSubmit(false);
 112     }
 113
 114     private void testSuccessfulSubmit(final boolean doAsync) throws InterruptedException {
 115         final CountDownLatch asyncCanCommitContinue = new CountDownLatch(1);
 116         Answer<ListenableFuture<Boolean>> asyncCanCommit = invocation -> {
 117             final SettableFuture<Boolean> future = SettableFuture.create();
 118             if (doAsync) {
 119                 new Thread() {
 120                     @Override
 121                     public void run() {
 122                         Uninterruptibles.awaitUninterruptibly(asyncCanCommitContinue,
 123                                 10, TimeUnit.SECONDS);
 124                         future.set(true);
 125                     }
 126                 }.start();
 127             } else {
 128                 future.set(true);
 129             }
 130
 131             return future;
 132         };
 133
 134         doAnswer(asyncCanCommit).when(mockCohort1).canCommit();
 135         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 136         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
 137
 138         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 139         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
 140         doReturn(Futures.immediateFuture(null)).when(mockCohort2).commit();
 141
 142         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 143                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 144
 145         final CountDownLatch doneLatch = new CountDownLatch(1);
 146         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
 147         Futures.addCallback(future, new FutureCallback<Void>() {
 148             @Override
 149             public void onSuccess(final Void result) {
 150                 doneLatch.countDown();
 151             }
 152
 153             @Override
 154             public void onFailure(final Throwable failure) {
 155                 caughtEx.set(failure);
 156                 doneLatch.countDown();
 157             }
 158         });
 159
 160         asyncCanCommitContinue.countDown();
 161
 162         assertEquals("Submit complete", true, doneLatch.await(5, TimeUnit.SECONDS));
 163
 164         if (caughtEx.get() != null) {
 165             Throwables.propagate(caughtEx.get());
 166         }
 167
 168         assertEquals("Task count", doAsync ? 1 : 0, futureExecutor.getTaskCount());
 169
 170         InOrder inOrder = inOrder(mockCohort1, mockCohort2);
 171         inOrder.verify(mockCohort1).canCommit();
 172         inOrder.verify(mockCohort2).canCommit();
 173         inOrder.verify(mockCohort1).preCommit();
 174         inOrder.verify(mockCohort2).preCommit();
 175         inOrder.verify(mockCohort1).commit();
 176         inOrder.verify(mockCohort2).commit();
 177     }
 178
 179     @Test
 180     public void testSubmitWithNegativeCanCommitResponse() throws Exception {
 181         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 182         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 183
 184         doReturn(Futures.immediateFuture(false)).when(mockCohort2).canCommit();
 185         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 186
 187         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 188         doReturn(Futures.immediateFuture(false)).when(mockCohort3).canCommit();
 189         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 190
 191         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 192                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 193
 194         assertFailure(future, null, mockCohort1, mockCohort2, mockCohort3);
 195     }
 196
 197     private static void assertFailure(final CheckedFuture<Void, TransactionCommitFailedException> future,
 198             final Exception expCause, final DOMStoreThreePhaseCommitCohort... mockCohorts)
 199                     throws Exception {
 200         try {
 201             future.checkedGet(5, TimeUnit.SECONDS);
 202             fail("Expected TransactionCommitFailedException");
 203         } catch (TransactionCommitFailedException e) {
 204             if (expCause != null) {
 205                 assertSame("Expected cause", expCause.getClass(), e.getCause().getClass());
 206             }
 207
 208             InOrder inOrder = inOrder((Object[])mockCohorts);
 209             for (DOMStoreThreePhaseCommitCohort c: mockCohorts) {
 210                 inOrder.verify(c).abort();
 211             }
 212         } catch (TimeoutException e) {
 213             throw e;
 214         }
 215     }
 216
 217     @Test
 218     public void testSubmitWithCanCommitException() throws Exception {
 219         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 220         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 221
 222         IllegalStateException cause = new IllegalStateException("mock");
 223         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
 224         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 225
 226         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 227                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 228
 229         assertFailure(future, cause, mockCohort1, mockCohort2);
 230     }
 231
 232     @Test
 233     public void testSubmitWithCanCommitDataStoreUnavailableException() throws Exception {
 234         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 235         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 236         NoShardLeaderException rootCause = new NoShardLeaderException("mock");
 237         DataStoreUnavailableException cause = new DataStoreUnavailableException(rootCause.getMessage(), rootCause);
 238         doReturn(Futures.immediateFailedFuture(rootCause)).when(mockCohort2).canCommit();
 239         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 240
 241         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 242             transaction, Arrays.asList(mockCohort1, mockCohort2));
 243
 244         assertFailure(future, cause, mockCohort1, mockCohort2);
 245     }
 246
 247     @Test
 248     public void testSubmitWithPreCommitException() throws Exception {
 249         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 250         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 251         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 252
 253         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 254         IllegalStateException cause = new IllegalStateException("mock");
 255         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).preCommit();
 256         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 257
 258         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 259         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
 260         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2")))
 261                 .when(mockCohort3).preCommit();
 262         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 263
 264         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 265                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 266
 267         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
 268     }
 269
 270     @Test
 271     public void testSubmitWithCommitException() throws Exception {
 272         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 273         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
 274         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
 275         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
 276
 277         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
 278         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
 279         IllegalStateException cause = new IllegalStateException("mock");
 280         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).commit();
 281         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 282
 283         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
 284         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
 285         doReturn(Futures.immediateFuture(null)).when(mockCohort3).preCommit();
 286         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2")))
 287                 .when(mockCohort3).commit();
 288         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
 289
 290         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 291                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
 292
 293         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
 294     }
 295
 296     @Test
 297     public void testSubmitWithAbortException() throws Exception {
 298         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
 299         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock abort error")))
 300                 .when(mockCohort1).abort();
 301
 302         IllegalStateException cause = new IllegalStateException("mock canCommit error");
 303         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
 304         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
 305
 306         CheckedFuture<Void, TransactionCommitFailedException> future = coordinator.submit(
 307                 transaction, Arrays.asList(mockCohort1, mockCohort2));
 308
 309         assertFailure(future, cause, mockCohort1, mockCohort2);
 310     }
 311
 312     @Test
 313     public void testCreateReadWriteTransaction() {
 314         DOMStore domStore = mock(DOMStore.class);
 315         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 316                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 317                 futureExecutor)) {
 318             dataBroker.newReadWriteTransaction();
 319
 320             verify(domStore, never()).newReadWriteTransaction();
 321         }
 322     }
 323
 324     @Test
 325     public void testCreateWriteOnlyTransaction() {
 326         DOMStore domStore = mock(DOMStore.class);
 327         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 328                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 329                 futureExecutor)) {
 330             dataBroker.newWriteOnlyTransaction();
 331
 332             verify(domStore, never()).newWriteOnlyTransaction();
 333         }
 334     }
 335
 336     @Test
 337     public void testCreateReadOnlyTransaction() {
 338         DOMStore domStore = mock(DOMStore.class);
 339         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 340                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 341                 futureExecutor)) {
 342             dataBroker.newReadOnlyTransaction();
 343
 344             verify(domStore, never()).newReadOnlyTransaction();
 345         }
 346     }
 347
 348     @Test
 349     public void testLazySubTransactionCreationForReadWriteTransactions() {
 350         DOMStore configDomStore = mock(DOMStore.class);
 351         DOMStore operationalDomStore = mock(DOMStore.class);
 352         DOMStoreReadWriteTransaction storeTxn = mock(DOMStoreReadWriteTransaction.class);
 353
 354         doReturn(storeTxn).when(operationalDomStore).newReadWriteTransaction();
 355         doReturn(storeTxn).when(configDomStore).newReadWriteTransaction();
 356
 357         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 358                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 359                 configDomStore), futureExecutor)) {
 360             DOMDataReadWriteTransaction dataTxn = dataBroker.newReadWriteTransaction();
 361
 362             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 363             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 364             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
 365
 366             verify(configDomStore, never()).newReadWriteTransaction();
 367             verify(operationalDomStore, times(1)).newReadWriteTransaction();
 368
 369             dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 370
 371             verify(configDomStore, times(1)).newReadWriteTransaction();
 372             verify(operationalDomStore, times(1)).newReadWriteTransaction();
 373         }
 374
 375     }
 376
 377     @Test
 378     public void testLazySubTransactionCreationForWriteOnlyTransactions() {
 379         DOMStore configDomStore = mock(DOMStore.class);
 380         DOMStore operationalDomStore = mock(DOMStore.class);
 381         DOMStoreWriteTransaction storeTxn = mock(DOMStoreWriteTransaction.class);
 382
 383         doReturn(storeTxn).when(operationalDomStore).newWriteOnlyTransaction();
 384         doReturn(storeTxn).when(configDomStore).newWriteOnlyTransaction();
 385
 386         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 387                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 388                 configDomStore), futureExecutor)) {
 389             DOMDataWriteTransaction dataTxn = dataBroker.newWriteOnlyTransaction();
 390
 391             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 392             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 393
 394             verify(configDomStore, never()).newWriteOnlyTransaction();
 395             verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
 396
 397             dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
 398
 399             verify(configDomStore, times(1)).newWriteOnlyTransaction();
 400             verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
 401         }
 402     }
 403
 404     @Test
 405     public void testLazySubTransactionCreationForReadOnlyTransactions() {
 406         DOMStore configDomStore = mock(DOMStore.class);
 407         DOMStore operationalDomStore = mock(DOMStore.class);
 408         DOMStoreReadTransaction storeTxn = mock(DOMStoreReadTransaction.class);
 409
 410         doReturn(storeTxn).when(operationalDomStore).newReadOnlyTransaction();
 411         doReturn(storeTxn).when(configDomStore).newReadOnlyTransaction();
 412
 413         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 414                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 415                 configDomStore), futureExecutor)) {
 416             DOMDataReadOnlyTransaction dataTxn = dataBroker.newReadOnlyTransaction();
 417
 418             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
 419             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
 420
 421             verify(configDomStore, never()).newReadOnlyTransaction();
 422             verify(operationalDomStore, times(1)).newReadOnlyTransaction();
 423
 424             dataTxn.read(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 425
 426             verify(configDomStore, times(1)).newReadOnlyTransaction();
 427             verify(operationalDomStore, times(1)).newReadOnlyTransaction();
 428         }
 429     }
 430
 431     @Test
 432     public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException {
 433         DOMStore configDomStore = mock(DOMStore.class);
 434         DOMStore operationalDomStore = mock(DOMStore.class);
 435         DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class);
 436         DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class);
 437
 438         doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction();
 439         doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready();
 440         doReturn(Futures.immediateFuture(false)).when(mockCohort).canCommit();
 441         doReturn(Futures.immediateFuture(null)).when(mockCohort).abort();
 442
 443         final CountDownLatch latch = new CountDownLatch(1);
 444         final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
 445
 446         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 447                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 448                 configDomStore), futureExecutor) {
 449             @Override
 450             public CheckedFuture<Void, TransactionCommitFailedException> submit(DOMDataWriteTransaction writeTx,
 451                     Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
 452                 commitCohorts.addAll(cohorts);
 453                 latch.countDown();
 454                 return super.submit(writeTx, cohorts);
 455             }
 456         }) {
 457             DOMDataReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
 458
 459             domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
 460
 461             domDataReadWriteTransaction.submit();
 462
 463             assertTrue(latch.await(10, TimeUnit.SECONDS));
 464
 465             assertTrue(commitCohorts.size() == 1);
 466         }
 467     }
 468
 469     @Test
 470     public void testSubmitWithOnlyTwoSubTransactions() throws InterruptedException {
 471         DOMStore configDomStore = mock(DOMStore.class);
 472         DOMStore operationalDomStore = mock(DOMStore.class);
 473         DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
 474         DOMStoreReadWriteTransaction configTransaction = mock(DOMStoreReadWriteTransaction.class);
 475         DOMStoreThreePhaseCommitCohort mockCohortOperational = mock(DOMStoreThreePhaseCommitCohort.class);
 476         DOMStoreThreePhaseCommitCohort mockCohortConfig = mock(DOMStoreThreePhaseCommitCohort.class);
 477
 478         doReturn(operationalTransaction).when(operationalDomStore).newReadWriteTransaction();
 479         doReturn(configTransaction).when(configDomStore).newReadWriteTransaction();
 480
 481         doReturn(mockCohortOperational).when(operationalTransaction).ready();
 482         doReturn(Futures.immediateFuture(false)).when(mockCohortOperational).canCommit();
 483         doReturn(Futures.immediateFuture(null)).when(mockCohortOperational).abort();
 484
 485         doReturn(mockCohortConfig).when(configTransaction).ready();
 486         doReturn(Futures.immediateFuture(false)).when(mockCohortConfig).canCommit();
 487         doReturn(Futures.immediateFuture(null)).when(mockCohortConfig).abort();
 488
 489         final CountDownLatch latch = new CountDownLatch(1);
 490         final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
 491
 492         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 493                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
 494                 configDomStore), futureExecutor) {
 495             @Override
 496             public CheckedFuture<Void, TransactionCommitFailedException> submit(DOMDataWriteTransaction transaction,
 497                     Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
 498                 commitCohorts.addAll(cohorts);
 499                 latch.countDown();
 500                 return super.submit(transaction, cohorts);
 501             }
 502         }) {
 503             DOMDataReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
 504
 505             domDataReadWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY,
 506                     mock(NormalizedNode.class));
 507             domDataReadWriteTransaction.merge(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY,
 508                     mock(NormalizedNode.class));
 509
 510             domDataReadWriteTransaction.submit();
 511
 512             assertTrue(latch.await(10, TimeUnit.SECONDS));
 513
 514             assertTrue(commitCohorts.size() == 2);
 515         }
 516     }
 517
 518     @Test
 519     public void testCreateTransactionChain() {
 520         DOMStore domStore = mock(DOMStore.class);
 521         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 522                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 523                 futureExecutor)) {
 524
 525             dataBroker.createTransactionChain(mock(TransactionChainListener.class));
 526
 527             verify(domStore, times(2)).createTransactionChain();
 528         }
 529
 530     }
 531
 532     @Test
 533     public void testCreateTransactionOnChain() {
 534         DOMStore domStore = mock(DOMStore.class);
 535         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 536                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 537                 futureExecutor)) {
 538
 539             DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
 540             DOMStoreTransactionChain mockChain = mock(DOMStoreTransactionChain.class);
 541
 542             doReturn(mockChain).when(domStore).createTransactionChain();
 543             doReturn(operationalTransaction).when(mockChain).newWriteOnlyTransaction();
 544
 545             DOMTransactionChain transactionChain = dataBroker.createTransactionChain(
 546                     mock(TransactionChainListener.class));
 547
 548             DOMDataWriteTransaction domDataWriteTransaction = transactionChain.newWriteOnlyTransaction();
 549
 550             verify(mockChain, never()).newWriteOnlyTransaction();
 551
 552             domDataWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY,
 553                     mock(NormalizedNode.class));
 554         }
 555     }
 556
 557     @Test
 558     public void testEmptyTransactionSubmitSucceeds() throws ExecutionException, InterruptedException {
 559         DOMStore domStore = mock(DOMStore.class);
 560         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 561                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
 562                 futureExecutor)) {
 563
 564             CheckedFuture<Void, TransactionCommitFailedException> submit1 =
 565                     dataBroker.newWriteOnlyTransaction().submit();
 566
 567             assertNotNull(submit1);
 568
 569             submit1.get();
 570
 571             CheckedFuture<Void, TransactionCommitFailedException> submit2 =
 572                     dataBroker.newReadWriteTransaction().submit();
 573
 574             assertNotNull(submit2);
 575
 576             submit2.get();
 577         }
 578     }
 579
 580     @Test
 581     public void testExtensions() {
 582         DistributedDataStore mockConfigStore = mock(DistributedDataStore.class);
 583         DistributedDataStore mockOperStore = mock(DistributedDataStore.class);
 584         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
 585                 LogicalDatastoreType.OPERATIONAL, mockOperStore,
 586                 LogicalDatastoreType.CONFIGURATION, mockConfigStore), futureExecutor)) {
 587
 588             Map<Class<? extends DOMDataBrokerExtension>, DOMDataBrokerExtension> supportedExtensions =
 589                     dataBroker.getSupportedExtensions();
 590             assertNotNull(supportedExtensions.get(DOMDataTreeChangeService.class));
 591
 592             DOMDataTreeCommitCohortRegistry cohortRegistry =
 593                     (DOMDataTreeCommitCohortRegistry) supportedExtensions.get(DOMDataTreeCommitCohortRegistry.class);
 594             assertNotNull(cohortRegistry);
 595
 596             DOMDataTreeCommitCohort mockCohort = mock(DOMDataTreeCommitCohort.class);
 597             DOMDataTreeIdentifier path = new DOMDataTreeIdentifier(
 598                     org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION,
 599                     YangInstanceIdentifier.EMPTY);
 600             cohortRegistry.registerCommitCohort(path, mockCohort);
 601
 602             verify(mockConfigStore).registerCommitCohort(path, mockCohort);
 603         }
 604     }
 605 }