assertEquals(createInnerMapNode(1), actualInnerListNode);
txCursor.close();
- tx.submit().checkedGet();
+ tx.commit().get();
verify(commitCohort, times(2)).canCommit();
verify(commitCohort, times(2)).preCommit();
cursor.close();
LOG.warn("Got to pre submit");
- tx.submit().checkedGet();
+ tx.commit().get();
shardRegistration.close().toCompletableFuture().get();
import akka.cluster.Cluster;
import akka.testkit.javadsl.TestKit;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.FluentFuture;
+import com.google.common.util.concurrent.ListenableFuture;
import com.typesafe.config.ConfigFactory;
import java.util.ArrayList;
import java.util.Collection;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
-import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
cursor.write(test.getIdentifier(), test);
cursor.close();
- tx.submit().checkedGet();
+ tx.commit().get();
}
@Test
cursor.close();
LOG.debug("Got to pre submit");
- tx.submit().checkedGet();
+ tx.commit().get();
final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
cursor.write(new NodeIdentifier(TestModel.INNER_LIST_QNAME), innerList);
cursor.close();
- tx.submit().checkedGet();
+ tx.commit().get();
- final ArrayList<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
+ final ArrayList<ListenableFuture<?>> futures = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
final Collection<MapEntryNode> innerListMapEntries = createInnerListMapEntries(1000, "run-" + i);
for (final MapEntryNode innerListMapEntry : innerListMapEntries) {
oid1.node(new NodeIdentifier(TestModel.INNER_LIST_QNAME))));
cursor1.write(innerListMapEntry.getIdentifier(), innerListMapEntry);
cursor1.close();
- futures.add(tx1.submit());
+ futures.add(tx1.commit());
}
}
- futures.get(futures.size() - 1).checkedGet();
+ futures.get(futures.size() - 1).get();
final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
cursor.write(testNode.getIdentifier(), testNode);
cursor.close();
- transaction.submit().checkedGet();
+ transaction.commit().get();
final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
cursor.write(wholeList.getIdentifier(), wholeList);
cursor.close();
- transaction.submit().checkedGet();
+ transaction.commit().get();
leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(TEST_ID),
true, Collections.emptyList());
* This thread only removes from futures set.
*/
private final ScheduledExecutorService completingExecutor = FinalizableScheduledExecutorService.newSingleThread();
- private final Collection<ListenableFuture<Void>> futures = Collections.synchronizedSet(new HashSet<>());
+ private final Collection<ListenableFuture<?>> futures = Collections.synchronizedSet(new HashSet<>());
private final Stopwatch stopwatch = Stopwatch.createUnstarted();
private final long runtimeNanos;
private final long delayNanos;
// Not completed yet: create a transaction and hook it up
final long txId = txCounter++;
- final ListenableFuture<Void> execFuture = execWrite(txId);
+ final ListenableFuture<?> execFuture = execWrite(txId);
LOG.debug("New future #{} allocated", txId);
// Ordering is important: we need to add the future before hooking the callback
futures.add(execFuture);
- Futures.addCallback(execFuture, new FutureCallback<Void>() {
+ Futures.addCallback(execFuture, new FutureCallback<Object>() {
@Override
- public void onSuccess(final Void result) {
+ public void onSuccess(final Object result) {
txSuccess(execFuture, txId);
}
return false;
}
- final void txSuccess(final ListenableFuture<Void> execFuture, final long txId) {
+ final void txSuccess(final ListenableFuture<?> execFuture, final long txId) {
LOG.debug("Future #{} completed successfully", txId);
futures.remove(execFuture);
}
}
- final void txFailure(final ListenableFuture<Void> execFuture, final long txId, final Throwable cause) {
+ final void txFailure(final ListenableFuture<?> execFuture, final long txId, final Throwable cause) {
LOG.debug("Future #{} failed", txId, cause);
futures.remove(execFuture);
synchronized (futures) {
int offset = 0;
- for (ListenableFuture<Void> future : futures) {
+ for (ListenableFuture<?> future : futures) {
try {
future.get(0, TimeUnit.NANOSECONDS);
} catch (final TimeoutException e) {
runTimedOut(new TimeoutException("Collection did not finish in " + DEAD_TIMEOUT_SECONDS + " seconds"));
}
- abstract ListenableFuture<Void> execWrite(long txId);
+ abstract ListenableFuture<?> execWrite(long txId);
abstract void runFailed(Throwable cause);
LOG.debug("Shard[{}] created successfully.", identifier);
registrations.put(identifier, registration);
- final ListenableFuture<Void> ensureFuture = ensureListExists();
- Futures.addCallback(ensureFuture, new FutureCallback<Void>() {
+ final ListenableFuture<?> ensureFuture = ensureListExists();
+ Futures.addCallback(ensureFuture, new FutureCallback<Object>() {
@Override
- public void onSuccess(@Nullable final Void result) {
+ public void onSuccess(@Nullable final Object result) {
LOG.debug("Initial list write successful.");
future.set(RpcResultBuilder.success(new CreatePrefixShardOutputBuilder().build()).build());
}
return future;
}
- private ListenableFuture<Void> ensureListExists() {
+ private ListenableFuture<?> ensureListExists() {
final CollectionNodeBuilder<MapEntryNode, MapNode> mapBuilder = ImmutableNodes.mapNodeBuilder(ID_INT);
cursor.merge(containerNode.getIdentifier(), containerNode);
cursor.close();
- final ListenableFuture<Void> future = tx.submit();
- Futures.addCallback(future, new FutureCallback<Void>() {
+ final ListenableFuture<?> future = tx.commit();
+ Futures.addCallback(future, new FutureCallback<Object>() {
@Override
- public void onSuccess(@Nullable final Void result) {
+ public void onSuccess(@Nullable final Object result) {
try {
LOG.debug("Closing producer for initial list.");
producer.close();
package org.opendaylight.controller.clustering.it.provider.impl;
import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.FluentFuture;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.eclipse.jdt.annotation.NonNull;
+import org.opendaylight.mdsal.common.api.CommitInfo;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
cursor.close();
try {
- tx.submit().get(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ tx.commit().get(INIT_TX_TIMEOUT_SECONDS, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
LOG.warn("Unable to fill the initial item list.", e);
closeProducer(itemProducer);
}
@Override
- ListenableFuture<Void> execWrite(final long txId) {
+ FluentFuture<? extends @NonNull CommitInfo> execWrite(final long txId) {
final int i = random.nextInt(MAX_ITEM + 1);
final DOMDataTreeCursorAwareTransaction tx = itemProducer.createTransaction(false);
final DOMDataTreeWriteCursor cursor = tx.createCursor(idListItem);
cursor.close();
- return tx.submit();
+ return tx.commit();
}
@Override
Code Review / controller.git / commitdiff