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