X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FRemoteTransactionContextSupport.java;h=c1162c2d9385f0b40e644adbf09cfba1a83c07a2;hb=5dbaf1259ead1904536db204bbc742a3359c1eb1;hp=4f41d8902e029dc97b99b9cb385e47682e076f7b;hpb=c796596b5c46b5203c30b143e6282662e66c5642;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java
index 4f41d8902e..c1162c2d93 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RemoteTransactionContextSupport.java
@@ -21,7 +21,6 @@ import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
-import org.opendaylight.controller.cluster.datastore.utils.TransactionIdentifierUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Future;
@@ -31,7 +30,7 @@ import scala.concurrent.duration.FiniteDuration;
  * Handles creation of TransactionContext instances for remote transactions. This class creates
  * remote transactions, if necessary, by sending CreateTransaction messages with retries, up to a limit,
  * if the shard doesn't have a leader yet. This is done by scheduling a retry task after a short delay.
- * <p>
+ * <p/>
  * The end result from a completed CreateTransaction message is a TransactionContext that is
  * used to perform transaction operations. Transaction operations that occur before the
  * CreateTransaction completes are cache via a TransactionContextWrapper and executed once the
@@ -60,8 +59,8 @@ final class RemoteTransactionContextSupport {
 
     private final TransactionContextWrapper transactionContextWrapper;
 
-    RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextWrapper, final TransactionProxy parent,
-            final String shardName) {
+    RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextWrapper,
+            final TransactionProxy parent, final String shardName) {
         this.parent = Preconditions.checkNotNull(parent);
         this.shardName = shardName;
         this.transactionContextWrapper = transactionContextWrapper;
@@ -98,12 +97,12 @@ final class RemoteTransactionContextSupport {
     /**
      * Sets the target primary shard and initiates a CreateTransaction try.
      */
-    void setPrimaryShard(PrimaryShardInfo primaryShardInfo) {
-        this.primaryShardInfo = primaryShardInfo;
+    void setPrimaryShard(final PrimaryShardInfo newPrimaryShardInfo) {
+        this.primaryShardInfo = newPrimaryShardInfo;
 
-        if (getTransactionType() == TransactionType.WRITE_ONLY  &&
-                getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
-            ActorSelection primaryShard = primaryShardInfo.getPrimaryShardActor();
+        if (getTransactionType() == TransactionType.WRITE_ONLY
+                && getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
+            ActorSelection primaryShard = newPrimaryShardInfo.getPrimaryShardActor();
 
             LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
                 getIdentifier(), primaryShard);
@@ -111,32 +110,20 @@ final class RemoteTransactionContextSupport {
             // For write-only Tx's we prepare the transaction modifications directly on the shard actor
             // to avoid the overhead of creating a separate transaction actor.
             transactionContextWrapper.executePriorTransactionOperations(createValidTransactionContext(
-                    primaryShard, String.valueOf(primaryShard.path()), primaryShardInfo.getPrimaryShardVersion()));
+                    primaryShard, String.valueOf(primaryShard.path()), newPrimaryShardInfo.getPrimaryShardVersion()));
         } else {
             tryCreateTransaction();
         }
     }
 
     /**
-     * @deprecated Temporary utility for extracting transaction chain ID from a {@link TransactionIdentifier}
-     */
-    @Deprecated
-    static String compatTransactionChainId(final TransactionIdentifier txId) {
-        final long historyId = txId.getHistoryId().getHistoryId();
-        return historyId == 0 ? "" : Long.toUnsignedString(historyId);
-    }
-
-    /**
-     * Performs a CreateTransaction try async.
+      Performs a CreateTransaction try async.
      */
     private void tryCreateTransaction() {
-        if(LOG.isDebugEnabled()) {
-            LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(),
-                    primaryShardInfo.getPrimaryShardActor());
-        }
+        LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(),
+                primaryShardInfo.getPrimaryShardActor());
 
-        Object serializedCreateMessage = new CreateTransaction(TransactionIdentifierUtils.actorNameFor(getIdentifier()),
-                getTransactionType().ordinal(), compatTransactionChainId(getIdentifier()),
+        Object serializedCreateMessage = new CreateTransaction(getIdentifier(), getTransactionType().ordinal(),
                     primaryShardInfo.getPrimaryShardVersion()).toSerializable();
 
         Future<Object> createTxFuture = getActorContext().executeOperationAsync(
@@ -144,7 +131,7 @@ final class RemoteTransactionContextSupport {
 
         createTxFuture.onComplete(new OnComplete<Object>() {
             @Override
-            public void onComplete(Throwable failure, Object response) {
+            public void onComplete(final Throwable failure, final Object response) {
                 onCreateTransactionComplete(failure, response);
             }
         }, getActorContext().getClientDispatcher());
@@ -157,15 +144,15 @@ final class RemoteTransactionContextSupport {
         Future<PrimaryShardInfo> findPrimaryFuture = getActorContext().findPrimaryShardAsync(shardName);
         findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
             @Override
-            public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) {
-                onFindPrimaryShardComplete(failure, primaryShardInfo);
+            public void onComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
+                onFindPrimaryShardComplete(failure, newPrimaryShardInfo);
             }
         }, getActorContext().getClientDispatcher());
     }
 
-    private void onFindPrimaryShardComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) {
+    private void onFindPrimaryShardComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
         if (failure == null) {
-            this.primaryShardInfo = primaryShardInfo;
+            this.primaryShardInfo = newPrimaryShardInfo;
             tryCreateTransaction();
         } else {
             LOG.debug("Tx {}: Find primary for shard {} failed", getIdentifier(), shardName, failure);
@@ -174,46 +161,40 @@ final class RemoteTransactionContextSupport {
         }
     }
 
-    private void onCreateTransactionComplete(Throwable failure, Object response) {
+    private void onCreateTransactionComplete(final Throwable failure, final Object response) {
         // An AskTimeoutException will occur if the local shard forwards to an unavailable remote leader or
         // the cached remote leader actor is no longer available.
-        boolean retryCreateTransaction = primaryShardInfo != null &&
-                (failure instanceof NoShardLeaderException || failure instanceof AskTimeoutException);
-        if(retryCreateTransaction) {
-            // Schedule a retry unless we're out of retries. Note: totalCreateTxTimeout is volatile as it may
-            // be written by different threads however not concurrently, therefore decrementing it
-            // non-atomically here is ok.
-            if(totalCreateTxTimeout > 0) {
-                long scheduleInterval = CREATE_TX_TRY_INTERVAL_IN_MS;
-                if(failure instanceof AskTimeoutException) {
-                    // Since we use the createTxMessageTimeout for the CreateTransaction request and it timed
-                    // out, subtract it from the total timeout. Also since the createTxMessageTimeout period
-                    // has already elapsed, we can immediately schedule the retry (10 ms is virtually immediate).
-                    totalCreateTxTimeout -= createTxMessageTimeout.duration().toMillis();
-                    scheduleInterval = 10;
-                }
-
-                totalCreateTxTimeout -= scheduleInterval;
-
-                LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms",
-                        getIdentifier(), shardName, failure, scheduleInterval);
-
-                getActorContext().getActorSystem().scheduler().scheduleOnce(
-                        FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS),
-                        new Runnable() {
-                            @Override
-                            public void run() {
-                                tryFindPrimaryShard();
-                            }
-                        }, getActorContext().getClientDispatcher());
-                return;
+        boolean retryCreateTransaction = primaryShardInfo != null
+                && (failure instanceof NoShardLeaderException || failure instanceof AskTimeoutException);
+
+        // Schedule a retry unless we're out of retries. Note: totalCreateTxTimeout is volatile as it may
+        // be written by different threads however not concurrently, therefore decrementing it
+        // non-atomically here is ok.
+        if (retryCreateTransaction && totalCreateTxTimeout > 0) {
+            long scheduleInterval = CREATE_TX_TRY_INTERVAL_IN_MS;
+            if (failure instanceof AskTimeoutException) {
+                // Since we use the createTxMessageTimeout for the CreateTransaction request and it timed
+                // out, subtract it from the total timeout. Also since the createTxMessageTimeout period
+                // has already elapsed, we can immediately schedule the retry (10 ms is virtually immediate).
+                totalCreateTxTimeout -= createTxMessageTimeout.duration().toMillis();
+                scheduleInterval = 10;
             }
+
+            totalCreateTxTimeout -= scheduleInterval;
+
+            LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms",
+                    getIdentifier(), shardName, failure, scheduleInterval);
+
+            getActorContext().getActorSystem().scheduler().scheduleOnce(
+                    FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS),
+                    this::tryFindPrimaryShard, getActorContext().getClientDispatcher());
+            return;
         }
 
         createTransactionContext(failure, response);
     }
 
-    private void createTransactionContext(Throwable failure, Object response) {
+    private void createTransactionContext(final Throwable failure, final Object response) {
         // Create the TransactionContext from the response or failure. Store the new
         // TransactionContext locally until we've completed invoking the
         // TransactionOperations. This avoids thread timing issues which could cause
@@ -223,15 +204,15 @@ final class RemoteTransactionContextSupport {
         // TransactionOperations. So to avoid thus timing, we don't publish the
         // TransactionContext until after we've executed all cached TransactionOperations.
         TransactionContext localTransactionContext;
-        if(failure != null) {
+        if (failure != null) {
             LOG.debug("Tx {} Creating NoOpTransaction because of error", getIdentifier(), failure);
 
             Throwable resultingEx = failure;
-            if(failure instanceof AskTimeoutException) {
+            if (failure instanceof AskTimeoutException) {
                 resultingEx = new ShardLeaderNotRespondingException(String.format(
                         "Could not create a %s transaction on shard %s. The shard leader isn't responding.",
                         parent.getType(), shardName), failure);
-            } else if(!(failure instanceof NoShardLeaderException)) {
+            } else if (!(failure instanceof NoShardLeaderException)) {
                 resultingEx = new Exception(String.format(
                     "Error creating %s transaction on shard %s", parent.getType(), shardName), failure);
             }
@@ -250,19 +231,19 @@ final class RemoteTransactionContextSupport {
         transactionContextWrapper.executePriorTransactionOperations(localTransactionContext);
     }
 
-    private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
+    private TransactionContext createValidTransactionContext(final CreateTransactionReply reply) {
         LOG.debug("Tx {} Received {}", getIdentifier(), reply);
 
         return createValidTransactionContext(getActorContext().actorSelection(reply.getTransactionPath()),
                 reply.getTransactionPath(), primaryShardInfo.getPrimaryShardVersion());
     }
 
-    private TransactionContext createValidTransactionContext(ActorSelection transactionActor, String transactionPath,
-            short remoteTransactionVersion) {
+    private TransactionContext createValidTransactionContext(final ActorSelection transactionActor,
+            final String transactionPath, final short remoteTransactionVersion) {
         final TransactionContext ret = new RemoteTransactionContext(transactionContextWrapper.getIdentifier(),
                 transactionActor, getActorContext(), remoteTransactionVersion, transactionContextWrapper.getLimiter());
 
-        if(parent.getType() == TransactionType.READ_ONLY) {
+        if (parent.getType() == TransactionType.READ_ONLY) {
             TransactionContextCleanup.track(parent, ret);
         }