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);
  }
 }