BUG 8422: Change tx handlers hard timeout

[controller.git] / opendaylight / md-sal / samples / clustering-test-app / provider / src / main / java / org / opendaylight / controller / clustering / it / provider / impl / WriteTransactionsHandler.java

diff --git a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java
index fd47b7176b2411b2bb629d65c56c5a1bac7474af..c4abb391e64b94e59a5058d018d887af3e55d680 100644 (file)
--- a/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java
+++ b/opendaylight/md-sal/samples/clustering-test-app/provider/src/main/java/org/opendaylight/controller/clustering/it/provider/impl/WriteTransactionsHandler.java
@@ -9,7 +9,6 @@
 package org.opendaylight.controller.clustering.it.provider.impl;
 
 import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
@@ -19,13 +18,15 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.SplittableRandom;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
-import javax.annotation.Nullable;
+import java.util.concurrent.TimeoutException;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.controller.md.sal.common.api.data.OptimisticLockFailedException;
 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
@@ -40,10 +41,13 @@ import org.opendaylight.yangtools.yang.common.RpcError;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,6 +60,8 @@ public class WriteTransactionsHandler implements Runnable {
 
     private static final QName ID_INTS =
             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-ints");
+    private static final QName ID_INT =
+            QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id-int");
     private static final QName ID =
             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "id");
     private static final QName ITEM =
@@ -64,6 +70,7 @@ public class WriteTransactionsHandler implements Runnable {
             QName.create("tag:opendaylight.org,2017:controller:yang:lowlevel:target", "2017-02-15", "number");
 
     public static final YangInstanceIdentifier ID_INTS_YID = YangInstanceIdentifier.of(ID_INTS);
+    public static final YangInstanceIdentifier ID_INT_YID = ID_INTS_YID.node(ID_INT);
 
     private final DOMDataBroker domDataBroker;
     private final Long timeToTake;
@@ -127,17 +134,34 @@ public class WriteTransactionsHandler implements Runnable {
 
     private boolean ensureListExists(final SettableFuture<RpcResult<WriteTransactionsOutput>> settableFuture) {
 
-        final MapEntryNode entry = ImmutableNodes.mapEntryBuilder(ID_INTS, ID, id)
+        final ContainerNode containerNode = ImmutableContainerNodeBuilder.create()
+                .withNodeIdentifier(new NodeIdentifier(ID_INTS))
+                .withChild(ImmutableNodes.mapNodeBuilder(ID_INT).build())
+                .build();
+
+        DOMDataWriteTransaction tx = txProvider.createTransaction();
+        // write only the top list
+        tx.merge(LogicalDatastoreType.CONFIGURATION, ID_INTS_YID, containerNode);
+        try {
+            tx.submit().checkedGet();
+        } catch (final OptimisticLockFailedException e) {
+            // when multiple write-transactions are executed concurrently we need to ignore this.
+            // If we get optimistic lock here it means id-ints already exists and we can continue.
+            LOG.debug("Got an optimistic lock when writing initial top level list element.", e);
+        } catch (final TransactionCommitFailedException e) {
+            LOG.warn("Unable to ensure IdInts list for id: {} exists.", id, e);
+            settableFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
+                    .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
+            return false;
+        }
+
+        final MapEntryNode entry = ImmutableNodes.mapEntryBuilder(ID_INT, ID, id)
                 .withChild(ImmutableNodes.mapNodeBuilder(ITEM).build())
                 .build();
-        final MapNode mapNode =
-                ImmutableNodes.mapNodeBuilder(ID_INTS)
-                        .withChild(entry)
-                        .build();
 
-        final DOMDataWriteTransaction tx = txProvider.createTransaction();
-        idListWithKey = ID_INTS_YID.node(entry.getIdentifier());
-        tx.merge(LogicalDatastoreType.CONFIGURATION, ID_INTS_YID, mapNode);
+        idListWithKey = ID_INT_YID.node(entry.getIdentifier());
+        tx = txProvider.createTransaction();
+        tx.merge(LogicalDatastoreType.CONFIGURATION, idListWithKey, entry);
 
         try {
             tx.submit().checkedGet();
@@ -165,8 +189,8 @@ public class WriteTransactionsHandler implements Runnable {
         tx.put(LogicalDatastoreType.CONFIGURATION, itemListId, mapBuilder.build());
 
         try {
-            tx.submit().checkedGet();
-        } catch (final TransactionCommitFailedException e) {
+            tx.submit().checkedGet(125, TimeUnit.SECONDS);
+        } catch (final TransactionCommitFailedException | TimeoutException e) {
             LOG.warn("Unable to fill the initial item list.", e);
             settableFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
                     .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", e).build());
@@ -209,32 +233,29 @@ public class WriteTransactionsHandler implements Runnable {
 
             final ListenableFuture<List<Void>> allFutures = Futures.allAsList(futures);
 
-            Futures.addCallback(allFutures, new FutureCallback<List<Void>>() {
-                @Override
-                public void onSuccess(@Nullable final List<Void> result) {
-                    LOG.debug("All futures completed successfully.");
+            try {
+                // Timeout from cds should be 2 minutes so leave some leeway.
+                allFutures.get(125, TimeUnit.SECONDS);
 
-                    final WriteTransactionsOutput output = new WriteTransactionsOutputBuilder()
-                            .setAllTx(allTx)
-                            .setInsertTx(insertTx)
-                            .setDeleteTx(deleteTx)
-                            .build();
+                LOG.debug("All futures completed successfully.");
 
-                    completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>success()
-                            .withResult(output).build());
+                final WriteTransactionsOutput output = new WriteTransactionsOutputBuilder()
+                        .setAllTx(allTx)
+                        .setInsertTx(insertTx)
+                        .setDeleteTx(deleteTx)
+                        .build();
 
-                    executor.shutdown();
-                }
+                completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>success()
+                        .withResult(output).build());
 
-                @Override
-                public void onFailure(final Throwable t) {
-                    LOG.error("Write transactions failed.", t);
-                    completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
-                            .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", t).build());
+                executor.shutdown();
+            } catch (InterruptedException | ExecutionException | TimeoutException exception) {
+                LOG.error("Write transactions failed.", exception);
+                completionFuture.set(RpcResultBuilder.<WriteTransactionsOutput>failed()
+                        .withError(RpcError.ErrorType.APPLICATION, "Unexpected-exception", exception).build());
 
-                    executor.shutdown();
-                }
-            });
+                executor.shutdown();
+            }
         }
     }