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