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