X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FAbstractShardBackendResolver.java;h=faffe0a3764985f5439e13867928a1ee2a1f2511;hp=6b221da76681911edf9155af13b1425f2b3d5017;hb=a6af137c30470b86d4bc624d4c48cb686495a182;hpb=bc2b83e97bc73930badd4a3063c65b849f82c664

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractShardBackendResolver.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractShardBackendResolver.java
index 6b221da766..faffe0a376 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractShardBackendResolver.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractShardBackendResolver.java
@@ -24,6 +24,7 @@ import org.opendaylight.controller.cluster.access.ABIVersion;
 import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
 import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
 import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
+import org.opendaylight.controller.cluster.access.commands.NotLeaderException;
 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
 import org.opendaylight.controller.cluster.common.actor.ExplicitAsk;
@@ -64,9 +65,9 @@ abstract class AbstractShardBackendResolver extends BackendInfoResolver<ShardBac
             return result;
         }
 
-        private synchronized void onStageResolved(final ShardBackendInfo result, final Throwable failure) {
+        private synchronized void onStageResolved(final ShardBackendInfo info, final Throwable failure) {
             if (failure == null) {
-                this.result = Preconditions.checkNotNull(result);
+                this.result = Preconditions.checkNotNull(info);
             } else {
                 LOG.warn("Failed to resolve shard", failure);
             }
@@ -135,26 +136,31 @@ abstract class AbstractShardBackendResolver extends BackendInfoResolver<ShardBac
         LOG.debug("Shard {} resolved to {}, attempting to connect", shardName, info);
 
         FutureConverters.toJava(ExplicitAsk.ask(info.getPrimaryShardActor(), connectFunction, CONNECT_TIMEOUT))
-            .whenComplete((response, failure) -> {
-                if (failure != null) {
-                    LOG.debug("Connect attempt to {} failed", shardName, failure);
-                    future.completeExceptionally(failure);
-                    return;
-                }
-                if (response instanceof RequestFailure) {
-                    final Throwable cause = ((RequestFailure<?, ?>) response).getCause().unwrap();
-                    LOG.debug("Connect attempt to {} failed to process", shardName, cause);
-                    future.completeExceptionally(cause);
-                    return;
-                }
-
-                LOG.debug("Resolved backend information to {}", response);
-                Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response {}",
-                    response);
-                final ConnectClientSuccess success = (ConnectClientSuccess) response;
-                future.complete(new ShardBackendInfo(success.getBackend(), nextSessionId.getAndIncrement(),
-                    success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie), success.getDataTree(),
-                    success.getMaxMessages()));
-            });
+            .whenComplete((response, failure) -> onConnectResponse(shardName, cookie, future, response, failure));
+    }
+
+    private void onConnectResponse(final String shardName, final long cookie,
+            final CompletableFuture<ShardBackendInfo> future, final Object response, final Throwable failure) {
+        if (failure != null) {
+            LOG.debug("Connect attempt to {} failed, will retry", shardName, failure);
+            future.completeExceptionally(wrap("Connection attempt failed", failure));
+            return;
+        }
+        if (response instanceof RequestFailure) {
+            final Throwable cause = ((RequestFailure<?, ?>) response).getCause().unwrap();
+            LOG.debug("Connect attempt to {} failed to process", shardName, cause);
+            final Throwable result = cause instanceof NotLeaderException
+                    ? wrap("Leader moved during establishment", cause) : cause;
+            future.completeExceptionally(result);
+            return;
+        }
+
+        LOG.debug("Resolved backend information to {}", response);
+        Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response %s",
+            response);
+        final ConnectClientSuccess success = (ConnectClientSuccess) response;
+        future.complete(new ShardBackendInfo(success.getBackend(), nextSessionId.getAndIncrement(),
+            success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie), success.getDataTree(),
+            success.getMaxMessages()));
     }
 }