fix: 4.x can't connect to the cluster when first node is non responsive

.github/workflows/tests@v1.yml

@@ -179,7 +179,7 @@ jobs:
 
  - name: Setup environment
  run: |
- pip3 install https://github.com/scylladb/scylla-ccm/archive/81076bce792a0fb3f2050e4c209a93e4a62ab55f.zip
+ pip3 install https://github.com/scylladb/scylla-ccm/archive/master.zip
 
  - name: Get cassandra version
  id: cassandra-version

core/src/main/java/com/datastax/oss/driver/api/core/auth/PlainTextAuthProviderBase.java

@@ -21,6 +21,7 @@
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.api.core.session.Session;
 import com.datastax.oss.driver.shaded.guava.common.base.Charsets;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.net.InetSocketAddress;
@@ -29,6 +30,7 @@
 import java.nio.CharBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
@@ -171,6 +173,12 @@ public SocketAddress resolve() {
  return new InetSocketAddress("127.0.0.1", 9042);
  }
 
+ @NonNull
+ @Override
+ public List<EndPoint> resolveAll() {
+ return ImmutableList.of(this);
+ }
+
  @NonNull
  @Override
  public String asMetricPrefix() {

core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java

@@ -20,6 +20,7 @@
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.List;
 
 /**
  * Encapsulates the information needed to open connections to a node.
@@ -40,6 +41,13 @@ public interface EndPoint {
  @NonNull
  SocketAddress resolve();
 
+ /**
+ * Resolves this instance to a list of {@link EndPoint}.
+ *
+ * <p>This is called occasionally to resolve unresolved endpoints to their resolved counterparts.
+ */
+ @NonNull
+ List<EndPoint> resolveAll();
  /**
  * Returns an alternate string representation for use in node-level metric names.
  *

core/src/main/java/com/datastax/oss/driver/internal/core/ContactPoints.java

@@ -19,6 +19,7 @@
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
 import com.datastax.oss.driver.internal.core.metadata.DefaultEndPoint;
+import com.datastax.oss.driver.internal.core.metadata.UnresolvedEndPoint;
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet;
 import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
 import java.net.InetAddress;
@@ -41,18 +42,17 @@ public static Set<EndPoint> merge(
 
  Set<EndPoint> result = Sets.newHashSet(programmaticContactPoints);
  for (String spec : configContactPoints) {
- for (InetSocketAddress address : extract(spec, resolve)) {
- DefaultEndPoint endPoint = new DefaultEndPoint(address);
+ for (EndPoint endPoint : extract(spec, resolve)) {
  boolean wasNew = result.add(endPoint);
  if (!wasNew) {
- LOG.warn("Duplicate contact point {}", address);
+ LOG.warn("Duplicate contact point {}", endPoint);
  }
  }
  }
  return ImmutableSet.copyOf(result);
  }
 
- private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
+ private static Set<EndPoint> extract(String spec, boolean resolve) {
  int separator = spec.lastIndexOf(':');
  if (separator < 0) {
  LOG.warn("Ignoring invalid contact point {} (expecting host:port)", spec);
@@ -69,7 +69,7 @@ private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
  return Collections.emptySet();
  }
  if (!resolve) {
- return ImmutableSet.of(InetSocketAddress.createUnresolved(host, port));
+ return ImmutableSet.of(new UnresolvedEndPoint(host, port));
  } else {
  try {
  InetAddress[] inetAddresses = InetAddress.getAllByName(host);
@@ -79,9 +79,9 @@ private static Set<InetSocketAddress> extract(String spec, boolean resolve) {
  spec,
  Arrays.deepToString(inetAddresses));
  }
- Set<InetSocketAddress> result = new HashSet<>();
+ Set<EndPoint> result = new HashSet<>();
  for (InetAddress inetAddress : inetAddresses) {
- result.add(new InetSocketAddress(inetAddress, port));
+ result.add(new DefaultEndPoint(new InetSocketAddress(inetAddress, port)));
  }
  return result;
  } catch (UnknownHostException e) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java

@@ -18,9 +18,11 @@
 package com.datastax.oss.driver.internal.core.metadata;
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.Serializable;
 import java.net.InetSocketAddress;
+import java.util.List;
 import java.util.Objects;
 
 public class DefaultEndPoint implements EndPoint, Serializable {
@@ -41,6 +43,12 @@ public InetSocketAddress resolve() {
  return address;
  }
 
+ @NonNull
+ @Override
+ public List<EndPoint> resolveAll() {
+ return ImmutableList.of(this);
+ }
+
  @Override
  public boolean equals(Object other) {
  if (other == this) {

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/InitialNodeListRefresh.java

@@ -72,20 +72,26 @@ public Result compute(
  + "keeping only the first one",
  logPrefix,
  hostId);
+ continue;
+ }
+ EndPoint endPoint = nodeInfo.getEndPoint();
+ DefaultNode node = findIn(contactPoints, endPoint);
+ if (node == null) {
+ node = new DefaultNode(endPoint, context);
+ LOG.debug("[{}] Adding new node {}", logPrefix, node);
  } else {
- EndPoint endPoint = nodeInfo.getEndPoint();
-  DefaultNode node = findIn(contactPoints, endPoint);
-  if (node == null) {
-  node = new DefaultNode(endPoint, context);
-  LOG.debug("[{}] Adding new node {}", logPrefix, node);
- } else {
-  LOG.debug("[{}] Copying contact point {}", logPrefix, node);
-  }
- if (tokenMapEnabled && tokenFactory == null && nodeInfo.getPartitioner() != null) {
+ LOG.debug("[{}] Copying contact point {}", logPrefix, node);
+ }
+ copyInfos(nodeInfo, node, context);
+ newNodes.put(hostId, node);
+ }
+
+ if (tokenMapEnabled) {
+ for (NodeInfo nodeInfo : nodeInfos) {
+ if (nodeInfo.getPartitioner() != null) {
  tokenFactory = tokenFactoryRegistry.tokenFactoryFor(nodeInfo.getPartitioner());
+ break;
  }
- copyInfos(nodeInfo, node, context);
- newNodes.put(hostId, node);
  }
  }
 

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java

@@ -55,6 +55,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArraySet;
 import net.jcip.annotations.ThreadSafe;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -80,7 +81,8 @@ public class MetadataManager implements AsyncAutoCloseable {
  private volatile KeyspaceFilter keyspaceFilter;
  private volatile Boolean schemaEnabledProgrammatically;
  private volatile boolean tokenMapEnabled;
- private volatile Set<DefaultNode> contactPoints;
+ private volatile Set<EndPoint> contactPoints;
+ private volatile Set<DefaultNode> resolvedContactPoints;
  private volatile boolean wasImplicitContactPoint;
  private volatile TypeCodec<TupleValue> tabletPayloadCodec = null;
 
@@ -102,7 +104,7 @@ protected MetadataManager(InternalDriverContext context, DefaultMetadata initial
  DefaultDriverOption.METADATA_SCHEMA_REFRESHED_KEYSPACES, Collections.emptyList());
  this.keyspaceFilter = KeyspaceFilter.newInstance(logPrefix, refreshedKeyspaces);
  this.tokenMapEnabled = config.getBoolean(DefaultDriverOption.METADATA_TOKEN_MAP_ENABLED);
-
+ this.resolvedContactPoints = new CopyOnWriteArraySet<>();
  context.getEventBus().register(ConfigChangeEvent.class, this::onConfigChanged);
  }
 
@@ -145,18 +147,19 @@ public void addContactPoints(Set<EndPoint> providedContactPoints) {
  // Convert the EndPoints to Nodes, but we can't put them into the Metadata yet, because we
  // don't know their host_id. So store them in a volatile field instead, they will get copied
  // during the first node refresh.
- ImmutableSet.Builder<DefaultNode> contactPointsBuilder = ImmutableSet.builder();
+ ImmutableSet.Builder<EndPoint> contactPointsBuilder = ImmutableSet.builder();
  if (providedContactPoints == null || providedContactPoints.isEmpty()) {
  LOG.info(
  "[{}] No contact points provided, defaulting to {}", logPrefix, DEFAULT_CONTACT_POINT);
  this.wasImplicitContactPoint = true;
- contactPointsBuilder.add(new DefaultNode(DEFAULT_CONTACT_POINT, context));
+ contactPointsBuilder.add(DEFAULT_CONTACT_POINT);
  } else {
  for (EndPoint endPoint : providedContactPoints) {
- contactPointsBuilder.add(new DefaultNode(endPoint, context));
+ contactPointsBuilder.add(endPoint);
  }
  }
  this.contactPoints = contactPointsBuilder.build();
+ this.resolveContactPoints();
  LOG.debug("[{}] Adding initial contact points {}", logPrefix, contactPoints);
  }
 
@@ -167,7 +170,30 @@ public void addContactPoints(Set<EndPoint> providedContactPoints) {
  * @see #wasImplicitContactPoint()
  */
  public Set<DefaultNode> getContactPoints() {
- return contactPoints;
+ return resolvedContactPoints;
+ }
+
+ public synchronized void resolveContactPoints() {
+ ImmutableSet.Builder<EndPoint> resultBuilder = ImmutableSet.builder();
+ for (EndPoint endPoint : contactPoints) {
+ List<EndPoint> resolveEndpoints = endPoint.resolveAll();
+ if (resolveEndpoints.isEmpty()) {
+ LOG.error("failed to resolve contact endpoint {}", endPoint);
+ } else {
+ resultBuilder.addAll(resolveEndpoints);
+ }
+ }
+
+ Set<EndPoint> result = resultBuilder.build();
+ for (EndPoint endPoint : result) {
+ if (resolvedContactPoints.stream()
+ .anyMatch(resolved -> resolved.getEndPoint().equals(endPoint))) {
+ continue;
+ }
+ this.resolvedContactPoints.add(new DefaultNode(endPoint, context));
+ }
+
+ this.resolvedContactPoints.removeIf(endPoint -> !result.contains(endPoint.getEndPoint()));
  }
 
  /** Whether the default contact point was used (because none were provided explicitly). */
@@ -337,10 +363,13 @@ private SingleThreaded(InternalDriverContext context, DriverExecutionProfile con
  }
 
  private Void refreshNodes(Iterable<NodeInfo> nodeInfos) {
+ if (!didFirstNodeListRefresh) {
+ resolveContactPoints();
+ }
  MetadataRefresh refresh =
  didFirstNodeListRefresh
  ? new FullNodeListRefresh(nodeInfos)
- : new InitialNodeListRefresh(nodeInfos, contactPoints);
+ : new InitialNodeListRefresh(nodeInfos, resolvedContactPoints);
  didFirstNodeListRefresh = true;
  return apply(refresh);
  }

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java

@@ -18,13 +18,15 @@
 package com.datastax.oss.driver.internal.core.metadata;
 
 import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.driver.shaded.guava.common.primitives.UnsignedBytes;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -72,6 +74,12 @@ public InetSocketAddress resolve() {
  }
  }
 
+ @NonNull
+ @Override
+ public List<EndPoint> resolveAll() {
+ return ImmutableList.of(this);
+ }
+
  @Override
  public boolean equals(Object other) {
  if (other == this) {