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