JCBC-1642: ClusterManager must use resolved SeedNodes

Previously the cluster manager would use the seed nodes directly
from the connection string, but this is a problem because if
DNS SRV is used it would try to connect to the dns host instead
of the resolved hosts.

This changeset updates the code so that the potentially resolved
seed hosts are used instead of the original one from the connection
string.

Note that this is still not 100% accurate since the cluster topology
might drift over time, but it fixes the issue at hand. Further
improvements can be maded alongside if neeeded in a later
changeset.

Change-Id: Ie96186d72b01c783172544433a4c456be15ced95
Reviewed-on: http://review.couchbase.org/c/couchbase-java-client/+/129906
Reviewed-by: Graham Pople <grahampople@gmail.com>
Tested-by: Michael Nitschinger <michael.nitschinger@couchbase.com>

Author: daschl

src/main/java/com/couchbase/client/java/CouchbaseAsyncCluster.java

@@ -137,9 +137,9 @@ public class CouchbaseAsyncCluster implements AsyncCluster {
 
     private final ClusterFacade core;
     private final CouchbaseEnvironment environment;
-    private final ConnectionString connectionString;
     private final Map<String, AsyncBucket> bucketCache;
     private final boolean sharedEnvironment;
+    private final List<String> seedNodes;
     private Authenticator authenticator;
 
     /**
@@ -271,14 +271,12 @@ public static CouchbaseAsyncCluster fromConnectionString(final CouchbaseEnvironm
         if(connectionString.username() != null && !connectionString.username().equals("")) {
             this.authenticator = new PasswordAuthenticator(connectionString.username(), "");
         }
-        core = new CouchbaseCore(environment);
-        SeedNodesRequest request = new SeedNodesRequest(
-            assembleSeedNodes(connectionString, environment)
-        );
-        core.send(request).toBlocking().single();
+        this.core = new CouchbaseCore(environment);
+        this.seedNodes = assembleSeedNodes(connectionString, environment);
         this.environment = environment;
-        this.connectionString = connectionString;
-        this.bucketCache = new ConcurrentHashMap<String, AsyncBucket>();
+        this.bucketCache = new ConcurrentHashMap<>();
+
+        core.send(new SeedNodesRequest(seedNodes)).toBlocking().single();
     }
 
     /**
@@ -294,7 +292,7 @@ public static CouchbaseAsyncCluster fromConnectionString(final CouchbaseEnvironm
      */
     private static List<String> assembleSeedNodes(ConnectionString connectionString,
         CouchbaseEnvironment environment) {
-        List<String> seedNodes = new ArrayList<String>();
+        List<String> seedNodes = new ArrayList<>();
 
         if (environment.dnsSrvEnabled()) {
             seedNodesViaDnsSrv(connectionString, environment, seedNodes);
@@ -481,9 +479,16 @@ public void call(Boolean aBoolean) {
     @Override
     public Observable<AsyncClusterManager> clusterManager(final String username,
         final String password) {
+
+        // To not change too much code, just turn the resolved hosts into a new
+        // connection string and resolve the hosts from there.
+        final List<ConnectionString.UnresolvedSocket> seedNodes = ConnectionString
+          .fromHostnames(this.seedNodes)
+          .hosts();
+
         return Observable.just(
             (AsyncClusterManager) DefaultAsyncClusterManager.create(
-                username, password, connectionString, environment, core
+                username, password, seedNodes, environment, core
             )
         );
     }

src/main/java/com/couchbase/client/java/CouchbaseCluster.java

@@ -244,7 +244,7 @@ public static CouchbaseCluster fromConnectionString(final CouchbaseEnvironment e
         );
         this.environment = environment;
         this.connectionString = connectionString;
-        this.bucketCache = new ConcurrentHashMap<String, Bucket>();
+        this.bucketCache = new ConcurrentHashMap<>();
     }
 
     @Override
@@ -388,8 +388,7 @@ public ClusterManager clusterManager(final String username, final String passwor
             .map(new Func1<AsyncClusterManager, ClusterManager>() {
                 @Override
                 public ClusterManager call(AsyncClusterManager asyncClusterManager) {
-                    return DefaultClusterManager.create(username, password, connectionString,
-                        environment, core());
+                    return new DefaultClusterManager(asyncClusterManager, username, password, environment, core());
                 }
             })
             .toBlocking()
@@ -404,8 +403,7 @@ public ClusterManager clusterManager() {
                 .map(new Func1<AsyncClusterManager, ClusterManager>() {
                     @Override
                     public ClusterManager call(AsyncClusterManager asyncClusterManager) {
-                        return DefaultClusterManager.create(cred.login(), cred.password(), connectionString,
-                                environment, core());
+                        return new DefaultClusterManager(asyncClusterManager, cred.login(), cred.password(), environment, core());
                     }
                 })
                 .toBlocking()

src/main/java/com/couchbase/client/java/cluster/DefaultAsyncClusterManager.java

@@ -77,20 +77,26 @@ public class DefaultAsyncClusterManager implements AsyncClusterManager {
     final String username;
     final String password;
     final CouchbaseEnvironment environment;
-    private final ConnectionString connectionString;
+    private final List<ConnectionString.UnresolvedSocket> seedNodes;
 
-    DefaultAsyncClusterManager(final String username, final String password, final ConnectionString connectionString,
+    DefaultAsyncClusterManager(final String username, final String password, final List<ConnectionString.UnresolvedSocket> seedNodes,
                                final CouchbaseEnvironment environment, final ClusterFacade core) {
         this.username = username;
         this.password = password;
         this.core = core;
         this.environment = environment;
-        this.connectionString = connectionString;
+        this.seedNodes = seedNodes;
     }
 
+    @Deprecated
     public static DefaultAsyncClusterManager create(final String username, final String password,
                                                     final ConnectionString connectionString, final CouchbaseEnvironment environment, final ClusterFacade core) {
-        return new DefaultAsyncClusterManager(username, password, connectionString, environment, core);
+        return create(username, password, connectionString.hosts(), environment, core);
+    }
+
+    public static DefaultAsyncClusterManager create(final String username, final String password,
+                                                    final List<ConnectionString.UnresolvedSocket> seedNodes, final CouchbaseEnvironment environment, final ClusterFacade core) {
+        return new DefaultAsyncClusterManager(username, password, seedNodes, environment, core);
     }
 
     @Override
@@ -695,17 +701,17 @@ public Boolean call(AddServiceResponse addServiceResponse) {
     }
 
     private Observable<Boolean> ensureServiceEnabled() {
-        if (connectionString.hosts().isEmpty()) {
-            return Observable.error(new IllegalStateException("No host found in the connection string! " + connectionString.toString()));
+        if (seedNodes.isEmpty()) {
+            return Observable.error(new IllegalStateException("No host found in the seed nodes!"));
         }
 
         final AtomicInteger integer = new AtomicInteger(0);
-        return Observable.just(connectionString.hosts())
+        return Observable.just(seedNodes)
                 .flatMap(new Func1<List<ConnectionString.UnresolvedSocket>, Observable<Boolean>>() {
                     @Override
                     public Observable<Boolean> call(List<ConnectionString.UnresolvedSocket> inetSocketAddresses) {
                         int hostIndex = integer.getAndIncrement();
-                        if (hostIndex >= connectionString.hosts().size()) {
+                        if (hostIndex >= seedNodes.size()) {
                             integer.set(0);
                             return Observable.error(new CouchbaseException("Could not enable ClusterManager service to function properly."));
                         }

src/main/java/com/couchbase/client/java/cluster/DefaultClusterManager.java

@@ -18,7 +18,6 @@
 import com.couchbase.client.core.ClusterFacade;
 import com.couchbase.client.core.annotations.InterfaceStability;
 import com.couchbase.client.core.utils.Blocking;
-import com.couchbase.client.java.cluster.api.AsyncClusterApiClient;
 import com.couchbase.client.core.utils.ConnectionString;
 import com.couchbase.client.java.cluster.api.ClusterApiClient;
 import com.couchbase.client.java.env.CouchbaseEnvironment;
@@ -29,19 +28,40 @@
 public class DefaultClusterManager implements ClusterManager {
 
     private static final TimeUnit TIMEOUT_UNIT = TimeUnit.MILLISECONDS;
-    private final DefaultAsyncClusterManager asyncClusterManager;
+    private final AsyncClusterManager asyncClusterManager;
+    private final String username;
+    private final String password;
+    private final ClusterFacade core;
     private final long timeout;
 
-    DefaultClusterManager(final String username, final String password, final ConnectionString connectionString,
+    public DefaultClusterManager(final AsyncClusterManager asyncClusterManager, final String username, final String password,
                           final CouchbaseEnvironment environment, final ClusterFacade core) {
-        asyncClusterManager = DefaultAsyncClusterManager.create(username, password, connectionString, environment,
+        this.asyncClusterManager = asyncClusterManager;
+        this.timeout = environment.managementTimeout();
+        this.username = username;
+        this.password = password;
+        this.core = core;
+    }
+
+    DefaultClusterManager(final String username, final String password, final List<ConnectionString.UnresolvedSocket> seedNodes,
+                          final CouchbaseEnvironment environment, final ClusterFacade core) {
+        asyncClusterManager = DefaultAsyncClusterManager.create(username, password, seedNodes, environment,
             core);
         this.timeout = environment.managementTimeout();
+        this.username = username;
+        this.password = password;
+        this.core = core;
     }
 
+    @Deprecated
     public static DefaultClusterManager create(final String username, final String password,
                                                final ConnectionString connectionString, final CouchbaseEnvironment environment, final ClusterFacade core) {
-        return new DefaultClusterManager(username, password, connectionString, environment, core);
+        return create(username, password, connectionString.hosts(), environment, core);
+    }
+
+    public static DefaultClusterManager create(final String username, final String password,
+                                               final List<ConnectionString.UnresolvedSocket> seedNodes, final CouchbaseEnvironment environment, final ClusterFacade core) {
+        return new DefaultClusterManager(username, password, seedNodes, environment, core);
     }
 
     @Override
@@ -162,7 +182,7 @@ public User getUser(AuthDomain domain, String userid, long timeout, TimeUnit tim
     @Override
     @InterfaceStability.Experimental
     public ClusterApiClient apiClient() {
-        return new ClusterApiClient(asyncClusterManager.username, asyncClusterManager.password, asyncClusterManager.core,
-                this.timeout, TIMEOUT_UNIT); //uses the management timeout as default for API calls as well
+        // uses the management timeout as default for API calls as well
+        return new ClusterApiClient(username, password, core, this.timeout, TIMEOUT_UNIT);
     }
 }