ConcurrentDOMDataBroker extends AbstractDOMBroker { @Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); } ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor);  ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor, final DurationStatisticsTracker commitStatsTracker); DurationStatisticsTracker getCommitStatsTracker(); @Override String toString();  }
@Test public void testSubmitWithNegativeCanCommitResponse() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).abort(); doReturn(Futures.immediateFuture(Boolean.FALSE)).when(mockCohort2).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(Futures.immediateFuture(Boolean.FALSE)).when(mockCohort3).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort3).abort(); ListenableFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3)); assertFailure(future, null, mockCohort1, mockCohort2, mockCohort3); }
@Test public void testSubmitWithCanCommitException() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).abort(); IllegalStateException cause = new IllegalStateException("mock"); doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); FluentFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2)); assertFailure(future, cause, mockCohort1, mockCohort2); }
@Test public void testSubmitWithCanCommitDataStoreUnavailableException() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).abort(); NoShardLeaderException rootCause = new NoShardLeaderException("mock"); DataStoreUnavailableException cause = new DataStoreUnavailableException(rootCause.getMessage(), rootCause); doReturn(Futures.immediateFailedFuture(rootCause)).when(mockCohort2).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); FluentFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2)); assertFailure(future, cause, mockCohort1, mockCohort2); }
@Test public void testSubmitWithPreCommitException() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).preCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).abort(); doReturn(immediateTrueFluentFuture()).when(mockCohort2).canCommit(); IllegalStateException cause = new IllegalStateException("mock"); doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).preCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(immediateTrueFluentFuture()).when(mockCohort3).canCommit(); doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2"))) .when(mockCohort3).preCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort3).abort(); FluentFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3)); assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3); }
@Test public void testSubmitWithCommitException() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).preCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).commit(); doReturn(immediateNullFluentFuture()).when(mockCohort1).abort(); doReturn(immediateTrueFluentFuture()).when(mockCohort2).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).preCommit(); IllegalStateException cause = new IllegalStateException("mock"); doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).commit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(immediateTrueFluentFuture()).when(mockCohort3).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort3).preCommit(); doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2"))) .when(mockCohort3).commit(); doReturn(immediateNullFluentFuture()).when(mockCohort3).abort(); FluentFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3)); assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3); }
@Test public void testSubmitWithAbortException() throws Exception { doReturn(immediateTrueFluentFuture()).when(mockCohort1).canCommit(); doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock abort error"))) .when(mockCohort1).abort(); IllegalStateException cause = new IllegalStateException("mock canCommit error"); doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort2).abort(); FluentFuture<? extends CommitInfo> future = coordinator.commit( transaction, Arrays.asList(mockCohort1, mockCohort2)); assertFailure(future, cause, mockCohort1, mockCohort2); }
@Test public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException { DOMStore configDomStore = mock(DOMStore.class); DOMStore operationalDomStore = mock(DOMStore.class); DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class); DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction(); doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready(); doReturn(immediateFalseFluentFuture()).when(mockCohort).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohort).abort(); final CountDownLatch latch = new CountDownLatch(1); final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>(); try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of( LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor) { @Override public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx, Collection<DOMStoreThreePhaseCommitCohort> cohorts) { commitCohorts.addAll(cohorts); latch.countDown(); return super.commit(writeTx, cohorts); } }) { DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction(); domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty()); domDataReadWriteTransaction.commit(); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertTrue(commitCohorts.size() == 1); } }
@Test public void testSubmitWithOnlyTwoSubTransactions() throws InterruptedException { DOMStore configDomStore = mock(DOMStore.class); DOMStore operationalDomStore = mock(DOMStore.class); DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class); DOMStoreReadWriteTransaction configTransaction = mock(DOMStoreReadWriteTransaction.class); DOMStoreThreePhaseCommitCohort mockCohortOperational = mock(DOMStoreThreePhaseCommitCohort.class); DOMStoreThreePhaseCommitCohort mockCohortConfig = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(operationalTransaction).when(operationalDomStore).newReadWriteTransaction(); doReturn(configTransaction).when(configDomStore).newReadWriteTransaction(); doReturn(mockCohortOperational).when(operationalTransaction).ready(); doReturn(immediateFalseFluentFuture()).when(mockCohortOperational).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohortOperational).abort(); doReturn(mockCohortConfig).when(configTransaction).ready(); doReturn(immediateFalseFluentFuture()).when(mockCohortConfig).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohortConfig).abort(); final CountDownLatch latch = new CountDownLatch(1); final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>(); try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of( LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor) { @Override @SuppressWarnings("checkstyle:hiddenField") public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx, Collection<DOMStoreThreePhaseCommitCohort> cohorts) { commitCohorts.addAll(cohorts); latch.countDown(); return super.commit(writeTx, cohorts); } }) { DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction(); domDataReadWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class)); domDataReadWriteTransaction.merge(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.empty(), mock(NormalizedNode.class)); domDataReadWriteTransaction.commit(); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertTrue(commitCohorts.size() == 2); } }
@Test public void testEmptyTransactionSubmitSucceeds() throws ExecutionException, InterruptedException { DOMStore domStore = mock(DOMStore.class); try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of( LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore), futureExecutor)) { FluentFuture<? extends CommitInfo> submit1 = dataBroker.newWriteOnlyTransaction().commit(); assertNotNull(submit1); submit1.get(); FluentFuture<? extends CommitInfo> submit2 = dataBroker.newReadWriteTransaction().commit(); assertNotNull(submit2); submit2.get(); } }