2 * Copyright (c) 2014 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.databroker;
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 static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateFalseFluentFuture;
23 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateNullFluentFuture;
24 import static org.opendaylight.yangtools.util.concurrent.FluentFutures.immediateTrueFluentFuture;
26 import com.google.common.base.Throwables;
27 import com.google.common.collect.ClassToInstanceMap;
28 import com.google.common.collect.ImmutableMap;
29 import com.google.common.util.concurrent.FluentFuture;
30 import com.google.common.util.concurrent.FutureCallback;
31 import com.google.common.util.concurrent.Futures;
32 import com.google.common.util.concurrent.ListenableFuture;
33 import com.google.common.util.concurrent.MoreExecutors;
34 import com.google.common.util.concurrent.SettableFuture;
35 import com.google.common.util.concurrent.Uninterruptibles;
36 import java.util.ArrayList;
37 import java.util.List;
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.mdsal.common.api.CommitInfo;
52 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
53 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
54 import org.opendaylight.mdsal.dom.api.DOMDataBrokerExtension;
55 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
56 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
57 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistry;
58 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
59 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
60 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
61 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
62 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
63 import org.opendaylight.mdsal.dom.api.DOMTransactionChainListener;
64 import org.opendaylight.mdsal.dom.broker.TransactionCommitFailedExceptionMapper;
65 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
66 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
67 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
68 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
69 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
70 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
71 import org.opendaylight.mdsal.dom.store.inmemory.InMemoryDOMDataStore;
72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
73 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
76 * Unit tests for DOMConcurrentDataCommitCoordinator.
78 * @author Thomas Pantelis
80 public class ConcurrentDOMDataBrokerTest {
82 private final DOMDataTreeWriteTransaction transaction = mock(DOMDataTreeWriteTransaction.class);
83 private final DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class);
84 private final ThreadPoolExecutor futureExecutor =
85 new ThreadPoolExecutor(0, 1, 5, TimeUnit.SECONDS, new SynchronousQueue<>());
86 private ConcurrentDOMDataBroker coordinator;
90 doReturn("tx").when(transaction).getIdentifier();
92 DOMStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.newDirectExecutorService());
94 coordinator = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL, store),
99 public void tearDown() {
100 futureExecutor.shutdownNow();
104 public void testSuccessfulSubmitAsync() throws Exception {
105 testSuccessfulSubmit(true);
109 public void testSuccessfulSubmitSync() throws Exception {
110 testSuccessfulSubmit(false);
113 private void testSuccessfulSubmit(final boolean doAsync) throws InterruptedException {
114 final CountDownLatch asyncCanCommitContinue = new CountDownLatch(1);
115 Answer<ListenableFuture<Boolean>> asyncCanCommit = invocation -> {
116 final SettableFuture<Boolean> future = SettableFuture.create();
119 Uninterruptibles.awaitUninterruptibly(asyncCanCommitContinue, 10, TimeUnit.SECONDS);
120 future.set(Boolean.TRUE);
123 future.set(Boolean.TRUE);
129 doAnswer(asyncCanCommit).when(mockCohort).canCommit();
130 doReturn(immediateNullFluentFuture()).when(mockCohort).preCommit();
131 doReturn(immediateNullFluentFuture()).when(mockCohort).commit();
133 ListenableFuture<? extends CommitInfo> future = coordinator.commit(transaction, mockCohort);
135 final CountDownLatch doneLatch = new CountDownLatch(1);
136 final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
137 Futures.addCallback(future, new FutureCallback<CommitInfo>() {
139 public void onSuccess(final CommitInfo result) {
140 doneLatch.countDown();
144 public void onFailure(final Throwable failure) {
145 caughtEx.set(failure);
146 doneLatch.countDown();
148 }, MoreExecutors.directExecutor());
150 asyncCanCommitContinue.countDown();
152 assertTrue("Submit complete", doneLatch.await(5, TimeUnit.SECONDS));
154 if (caughtEx.get() != null) {
155 Throwables.throwIfUnchecked(caughtEx.get());
156 throw new RuntimeException(caughtEx.get());
159 assertEquals("Task count", doAsync ? 1 : 0, futureExecutor.getTaskCount());
161 InOrder inOrder = inOrder(mockCohort);
162 inOrder.verify(mockCohort, times(1)).canCommit();
163 inOrder.verify(mockCohort, times(1)).preCommit();
164 inOrder.verify(mockCohort, times(1)).commit();
168 public void testSubmitWithNegativeCanCommitResponse() throws Exception {
169 doReturn(Futures.immediateFuture(Boolean.FALSE)).when(mockCohort).canCommit();
170 doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
172 assertFailure(coordinator.commit(transaction, mockCohort), null, mockCohort);
175 private static void assertFailure(final ListenableFuture<?> future, final Exception expCause,
176 final DOMStoreThreePhaseCommitCohort mockCohort) throws Exception {
178 future.get(5, TimeUnit.SECONDS);
179 fail("Expected TransactionCommitFailedException");
180 } catch (ExecutionException e) {
181 TransactionCommitFailedException tcf = TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER.apply(e);
182 if (expCause != null) {
183 assertSame("Expected cause", expCause.getClass(), tcf.getCause().getClass());
185 verify(mockCohort, times(1)).abort();
186 } catch (TimeoutException e) {
192 public void testSubmitWithCanCommitException() throws Exception {
193 final Exception cause = new IllegalStateException("mock");
194 doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).canCommit();
195 doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
197 assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
201 public void testSubmitWithPreCommitException() throws Exception {
202 doReturn(immediateTrueFluentFuture()).when(mockCohort).canCommit();
203 final IllegalStateException cause = new IllegalStateException("mock");
204 doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).preCommit();
205 doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
207 assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
211 public void testSubmitWithCommitException() throws Exception {
212 doReturn(immediateTrueFluentFuture()).when(mockCohort).canCommit();
213 doReturn(immediateNullFluentFuture()).when(mockCohort).preCommit();
214 final IllegalStateException cause = new IllegalStateException("mock");
215 doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort).commit();
216 doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
218 assertFailure(coordinator.commit(transaction, mockCohort), cause, mockCohort);
222 public void testSubmitWithAbortException() throws Exception {
223 final Exception canCommitCause = new IllegalStateException("canCommit error");
224 doReturn(Futures.immediateFailedFuture(canCommitCause)).when(mockCohort).canCommit();
225 final Exception abortCause = new IllegalStateException("abort error");
226 doReturn(Futures.immediateFailedFuture(abortCause)).when(mockCohort).abort();
228 assertFailure(coordinator.commit(transaction, mockCohort), canCommitCause, mockCohort);
232 public void testCreateReadWriteTransaction() {
233 DOMStore domStore = mock(DOMStore.class);
234 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
235 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
237 dataBroker.newReadWriteTransaction();
239 verify(domStore, never()).newReadWriteTransaction();
244 public void testCreateWriteOnlyTransaction() {
245 DOMStore domStore = mock(DOMStore.class);
246 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
247 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
249 dataBroker.newWriteOnlyTransaction();
251 verify(domStore, never()).newWriteOnlyTransaction();
256 public void testCreateReadOnlyTransaction() {
257 DOMStore domStore = mock(DOMStore.class);
258 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
259 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
261 dataBroker.newReadOnlyTransaction();
263 verify(domStore, never()).newReadOnlyTransaction();
268 public void testLazySubTransactionCreationForReadWriteTransactions() {
269 DOMStore configDomStore = mock(DOMStore.class);
270 DOMStore operationalDomStore = mock(DOMStore.class);
271 DOMStoreReadWriteTransaction storeTxn = mock(DOMStoreReadWriteTransaction.class);
273 doReturn(storeTxn).when(operationalDomStore).newReadWriteTransaction();
274 doReturn(storeTxn).when(configDomStore).newReadWriteTransaction();
276 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
277 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
278 configDomStore), futureExecutor)) {
279 DOMDataTreeReadWriteTransaction dataTxn = dataBroker.newReadWriteTransaction();
281 dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
282 dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
283 dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
285 verify(configDomStore, never()).newReadWriteTransaction();
286 verify(operationalDomStore, times(1)).newReadWriteTransaction();
292 public void testLazySubTransactionCreationForWriteOnlyTransactions() {
293 DOMStore configDomStore = mock(DOMStore.class);
294 DOMStore operationalDomStore = mock(DOMStore.class);
295 DOMStoreWriteTransaction storeTxn = mock(DOMStoreWriteTransaction.class);
297 doReturn(storeTxn).when(operationalDomStore).newWriteOnlyTransaction();
298 doReturn(storeTxn).when(configDomStore).newWriteOnlyTransaction();
300 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
301 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
302 configDomStore), futureExecutor)) {
303 DOMDataTreeWriteTransaction dataTxn = dataBroker.newWriteOnlyTransaction();
305 dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
306 dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class));
308 verify(configDomStore, never()).newWriteOnlyTransaction();
309 verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
314 public void testLazySubTransactionCreationForReadOnlyTransactions() {
315 DOMStore configDomStore = mock(DOMStore.class);
316 DOMStore operationalDomStore = mock(DOMStore.class);
317 DOMStoreReadTransaction storeTxn = mock(DOMStoreReadTransaction.class);
319 doReturn(storeTxn).when(operationalDomStore).newReadOnlyTransaction();
320 doReturn(storeTxn).when(configDomStore).newReadOnlyTransaction();
322 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
323 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
324 configDomStore), futureExecutor)) {
325 DOMDataTreeReadTransaction dataTxn = dataBroker.newReadOnlyTransaction();
327 dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
328 dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
330 verify(configDomStore, never()).newReadOnlyTransaction();
331 verify(operationalDomStore, times(1)).newReadOnlyTransaction();
336 public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException {
337 DOMStore configDomStore = mock(DOMStore.class);
338 DOMStore operationalDomStore = mock(DOMStore.class);
339 DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class);
341 doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction();
342 doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready();
343 doReturn(immediateFalseFluentFuture()).when(mockCohort).canCommit();
344 doReturn(immediateNullFluentFuture()).when(mockCohort).abort();
346 final CountDownLatch latch = new CountDownLatch(1);
347 final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
349 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
350 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
351 configDomStore), futureExecutor) {
353 public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx,
354 DOMStoreThreePhaseCommitCohort cohort) {
355 commitCohorts.add(cohort);
357 return super.commit(writeTx, cohort);
360 DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
362 domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty());
364 domDataReadWriteTransaction.commit();
366 assertTrue(latch.await(10, TimeUnit.SECONDS));
368 assertTrue(commitCohorts.size() == 1);
373 public void testCreateTransactionChain() {
374 DOMStore domStore = mock(DOMStore.class);
375 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
376 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
379 dataBroker.createTransactionChain(mock(DOMTransactionChainListener.class));
381 verify(domStore, times(2)).createTransactionChain();
387 public void testCreateTransactionOnChain() {
388 DOMStore domStore = mock(DOMStore.class);
389 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
390 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
393 DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
394 DOMStoreTransactionChain mockChain = mock(DOMStoreTransactionChain.class);
396 doReturn(mockChain).when(domStore).createTransactionChain();
397 doReturn(operationalTransaction).when(mockChain).newWriteOnlyTransaction();
399 DOMTransactionChain transactionChain = dataBroker.createTransactionChain(
400 mock(DOMTransactionChainListener.class));
402 DOMDataTreeWriteTransaction domDataWriteTransaction = transactionChain.newWriteOnlyTransaction();
404 verify(mockChain, never()).newWriteOnlyTransaction();
406 domDataWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(),
407 mock(NormalizedNode.class));
412 public void testEmptyTransactionSubmitSucceeds() throws ExecutionException, InterruptedException {
413 DOMStore domStore = mock(DOMStore.class);
414 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
415 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
418 FluentFuture<? extends CommitInfo> submit1 = dataBroker.newWriteOnlyTransaction().commit();
420 assertNotNull(submit1);
424 FluentFuture<? extends CommitInfo> submit2 = dataBroker.newReadWriteTransaction().commit();
426 assertNotNull(submit2);
433 public void testExtensions() {
434 DistributedDataStore mockConfigStore = mock(DistributedDataStore.class);
435 DistributedDataStore mockOperStore = mock(DistributedDataStore.class);
436 try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
437 LogicalDatastoreType.OPERATIONAL, mockOperStore,
438 LogicalDatastoreType.CONFIGURATION, mockConfigStore), futureExecutor)) {
440 ClassToInstanceMap<DOMDataBrokerExtension> supportedExtensions = dataBroker.getExtensions();
441 assertNotNull(supportedExtensions.getInstance(DOMDataTreeChangeService.class));
443 DOMDataTreeCommitCohortRegistry cohortRegistry = supportedExtensions.getInstance(
444 DOMDataTreeCommitCohortRegistry.class);
445 assertNotNull(cohortRegistry);
447 DOMDataTreeCommitCohort cohort = mock(DOMDataTreeCommitCohort.class);
448 DOMDataTreeIdentifier path = new DOMDataTreeIdentifier(
449 org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION,
450 YangInstanceIdentifier.empty());
451 cohortRegistry.registerCommitCohort(path, cohort);
453 verify(mockConfigStore).registerCommitCohort(path, cohort);