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