Adjust for zero token nodes

Adds ZeroTokenNodesIT that checks the behaviour of the driver when zero-token
nodes are involved.

Adds two options that control if peers that do not own any tokens are
considered valid and if peers that do not own any tokens are considered for
query planning.
Query planning setting does not by default override LBP behavior.
DCAwareRoundRobinPolicy, RackAwareRoundRobinPolicy and RoundRobinPolicy were
adjusted to follow this setting.

Driver behavior with those options disabled, which is the default option,
should remain unchanged.

driver-core/src/main/java/com/datastax/driver/core/ControlConnection.java

@@ -681,6 +681,10 @@ private static void updateInfo(
     host.setHostId(row.getUUID("host_id"));
     host.setSchemaVersion(row.getUUID("schema_version"));
 
+    if (row.getColumnDefinitions().contains("tokens")) {
+      host.setZeroToken(row.isNull("tokens"));
+    }
+
     EndPoint endPoint = cluster.configuration.getPolicies().getEndPointFactory().create(row);
     if (endPoint != null) {
       host.setEndPoint(endPoint);
@@ -950,8 +954,12 @@ private void refreshNodeListAndTokenMap(
         peerHost.setSchemaVersion(schemaVersions.get(i));
       }
 
-      if (metadataEnabled && factory != null && allTokens.get(i) != null)
+      if (metadataEnabled && factory != null && allTokens.get(i) != null) {
         tokenMap.put(peerHost, allTokens.get(i));
+      }
+      if (allTokens.get(i) == null) {
+        peerHost.setZeroToken(true);
+      }
 
       if (!isNew && isInitialConnection) {
         // If we're at init and the node already existed, it means it was a contact point, so we
@@ -1008,7 +1016,7 @@ private boolean isValidPeer(Row peerRow, boolean logIfInvalid) {
               && peerRow.getColumnDefinitions().contains("rack")
               && !peerRow.isNull("rack")
               && peerRow.getColumnDefinitions().contains("tokens")
-              && !peerRow.isNull("tokens");
+              && (!peerRow.isNull("tokens") || cluster.configuration.getQueryOptions().shouldConsiderZeroTokenNodesValidPeers());
     }
     if (!isValid && logIfInvalid)
       logger.warn(

driver-core/src/main/java/com/datastax/driver/core/Host.java

@@ -73,6 +73,9 @@ public class Host {
   // Whether host supports TABLETS_ROUTING_V1
   private volatile TabletInfo tabletInfo = null;
 
+  // Whether at the time of the last refresh system.peers/local row had column 'tokens' with 'null' value.
+  private volatile boolean zeroToken = false;
+
   enum State {
     ADDED,
     DOWN,
@@ -461,6 +464,16 @@ public void setTabletInfo(TabletInfo tabletInfo) {
     this.tabletInfo = tabletInfo;
   }
 
+  public boolean setZeroToken(boolean newValue) {
+    boolean previousValue = this.zeroToken;
+    this.zeroToken = newValue;
+    return previousValue;
+  }
+
+  public boolean isZeroToken() {
+    return this.zeroToken;
+  }
+
   /**
    * Returns whether the host is considered up by the driver.
    *

driver-core/src/main/java/com/datastax/driver/core/QueryOptions.java

@@ -74,6 +74,10 @@ public class QueryOptions {
 
   private volatile boolean addOriginalContactsToReconnectionPlan = false;
 
+  private volatile boolean excludeZeroTokenNodesFromQueryPlan = false;
+
+  private volatile boolean considerZeroTokenNodesValidPeers = false;
+
   /**
    * Creates a new {@link QueryOptions} instance using the {@link #DEFAULT_CONSISTENCY_LEVEL},
    * {@link #DEFAULT_SERIAL_CONSISTENCY_LEVEL} and {@link #DEFAULT_FETCH_SIZE}.
@@ -521,6 +525,34 @@ public boolean shouldAddOriginalContactsToReconnectionPlan() {
     return this.addOriginalContactsToReconnectionPlan;
   }
 
+  /**
+   * Whether the load balancing policies should consider zero-token nodes as valid targets for regular queries.
+   * Decision ultimately lies in LBP implementation. Config should be generally honored by current implementations.
+   */
+  public QueryOptions setExcludeZeroTokenNodesFromQueryPlan(boolean enabled) {
+    this.excludeZeroTokenNodesFromQueryPlan = enabled;
+    return this;
+  }
+
+  public boolean shouldExcludeZeroTokenNodesFromQueryPlan() {
+    return this.excludeZeroTokenNodesFromQueryPlan;
+  }
+
+  /**
+   * Scylla zero-token nodes have null value in "tokens" column in their system.peers rows.
+   * By default extended peer check considers such rows as invalid.
+   * Enabling this option will exclude this field from the check, and allow such rows
+   * from system.peers queries to be used when refreshing metadata.
+   */
+  public QueryOptions setConsiderZeroTokenNodesValidPeers(boolean enabled) {
+    this.considerZeroTokenNodesValidPeers = enabled;
+    return this;
+  }
+
+  public boolean shouldConsiderZeroTokenNodesValidPeers() {
+    return this.considerZeroTokenNodesValidPeers;
+  }
+
   @Override
   public boolean equals(Object that) {
     if (that == null || !(that instanceof QueryOptions)) {

driver-core/src/main/java/com/datastax/driver/core/policies/DCAwareRoundRobinPolicy.java

@@ -102,6 +102,9 @@ public void init(Cluster cluster, Collection<Host> hosts) {
     ArrayList<String> notInLocalDC = new ArrayList<String>();
 
     for (Host host : hosts) {
+      if (host.isZeroToken() && cluster.getConfiguration().getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+        continue;
+      }
       String dc = dc(host);
 
       // If the localDC was in "auto-discover" mode and it's the first host for which we have a DC,
@@ -156,6 +159,9 @@ private static CopyOnWriteArrayList<Host> cloneList(CopyOnWriteArrayList<Host> l
    */
   @Override
   public HostDistance distance(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return HostDistance.IGNORED;
+    }
     String dc = dc(host);
     if (dc == UNSET || dc.equals(localDc)) return HostDistance.LOCAL;
 
@@ -252,6 +258,9 @@ protected Host computeNext() {
 
   @Override
   public void onUp(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return;
+    }
     String dc = dc(host);
 
     // If the localDC was in "auto-discover" mode and it's the first host for which we have a DC,

driver-core/src/main/java/com/datastax/driver/core/policies/RackAwareRoundRobinPolicy.java

@@ -119,6 +119,9 @@ public void init(Cluster cluster, Collection<Host> hosts) {
     ArrayList<String> notInLocalRack = new ArrayList<String>();
 
     for (Host host : hosts) {
+      if (host.isZeroToken() && cluster.getConfiguration().getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+        continue;
+      }
       String dc = dc(host);
       String rack = rack(host);
 
@@ -204,6 +207,9 @@ private static CopyOnWriteArrayList<Host> cloneList(CopyOnWriteArrayList<Host> l
    */
   @Override
   public HostDistance distance(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return HostDistance.IGNORED;
+    }
     String dc = dc(host);
     if (dc == UNSET || dc.equals(localDc)) return HostDistance.LOCAL;
 
@@ -313,6 +319,9 @@ protected Host computeNext() {
 
   @Override
   public void onUp(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return;
+    }
     String dc = dc(host);
     String rack = rack(host);
 

driver-core/src/main/java/com/datastax/driver/core/policies/RoundRobinPolicy.java

@@ -59,8 +59,13 @@ public RoundRobinPolicy() {}
 
   @Override
   public void init(Cluster cluster, Collection<Host> hosts) {
-    this.liveHosts.addAll(hosts);
     this.configuration = cluster.getConfiguration();
+    for (Host host : hosts) {
+      if (host.isZeroToken() && cluster.getConfiguration().getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+        continue;
+      }
+      this.liveHosts.add(host);
+    }
     this.index.set(new Random().nextInt(Math.max(hosts.size(), 1)));
   }
 
@@ -76,6 +81,9 @@ public void init(Cluster cluster, Collection<Host> hosts) {
    */
   @Override
   public HostDistance distance(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return HostDistance.IGNORED;
+    }
     return HostDistance.LOCAL;
   }
 
@@ -139,6 +147,9 @@ protected Host computeNext() {
 
   @Override
   public void onUp(Host host) {
+    if (host.isZeroToken() && configuration.getQueryOptions().shouldExcludeZeroTokenNodesFromQueryPlan()) {
+      return;
+    }
     liveHosts.addIfAbsent(host);
   }