3.x: fail fast on keyspace setup validation failures

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

@@ -34,6 +34,7 @@
 import com.datastax.driver.core.exceptions.DriverInternalError;
 import com.datastax.driver.core.exceptions.FrameTooLongException;
 import com.datastax.driver.core.exceptions.OperationTimedOutException;
+import com.datastax.driver.core.exceptions.QueryValidationException;
 import com.datastax.driver.core.exceptions.TransportException;
 import com.datastax.driver.core.exceptions.UnsupportedProtocolVersionException;
 import com.datastax.driver.core.utils.MoreFutures;
@@ -845,17 +846,15 @@ void setKeyspace(String keyspace) throws ConnectionException {
 
     try {
       Uninterruptibles.getUninterruptibly(setKeyspaceAsync(keyspace));
-    } catch (ConnectionException e) {
-      throw defunct(e);
-    } catch (BusyConnectionException e) {
-      logger.warn(
-          "Tried to set the keyspace on busy {}. "
-              + "This should not happen but is not critical (it will be retried)",
-          this);
-      throw new ConnectionException(endPoint, "Tried to set the keyspace on busy connection");
     } catch (ExecutionException e) {
       Throwable cause = e.getCause();
-      if (cause instanceof OperationTimedOutException) {
+      if (cause instanceof BusyConnectionException) {
+        throw keyspaceBusyException();
+      } else if (cause instanceof QueryValidationException) {
+        throw (QueryValidationException) cause;
+      } else if (cause instanceof ConnectionException) {
+        throw defunct((ConnectionException) cause);
+      } else if (cause instanceof OperationTimedOutException) {
         // Rethrow so that the caller doesn't try to use the connection, but do not defunct as we
         // don't want to mark down
         logger.warn(
@@ -869,8 +868,15 @@ void setKeyspace(String keyspace) throws ConnectionException {
     }
   }
 
-  ListenableFuture<Connection> setKeyspaceAsync(final String keyspace)
-      throws ConnectionException, BusyConnectionException {
+  private ConnectionException keyspaceBusyException() {
+    logger.warn(
+        "Tried to set the keyspace on busy {}. "
+            + "This should not happen but is not critical (it will be retried)",
+        this);
+    return new ConnectionException(endPoint, "Tried to set the keyspace on busy connection");
+  }
+
+  ListenableFuture<Connection> setKeyspaceAsync(final String keyspace) {
     SetKeyspaceAttempt existingAttempt = targetKeyspace.get();
     if (MoreObjects.equal(existingAttempt.keyspace, keyspace)) return existingAttempt.future;
 
@@ -900,7 +906,18 @@ ListenableFuture<Connection> setKeyspaceAsync(final String keyspace)
         logger.debug("{} Setting keyspace {}", this, keyspace);
         // Note: we quote the keyspace below, because the name is the one coming from Cassandra, so
         // it's in the right case already
-        Future future = write(new Requests.Query("USE \"" + keyspace + '"'));
+        Future future;
+        try {
+          future = write(new Requests.Query("USE \"" + keyspace + '"'));
+        } catch (ConnectionException | BusyConnectionException e) {
+          targetKeyspace.compareAndSet(attempt, defaultKeyspaceAttempt);
+          ksFuture.setException(e);
+          return ksFuture;
+        } catch (RuntimeException e) {
+          targetKeyspace.compareAndSet(attempt, defaultKeyspaceAttempt);
+          ksFuture.setException(e);
+          return ksFuture;
+        }
         Futures.addCallback(
             future,
             new FutureCallback<Message.Response>() {
@@ -915,7 +932,12 @@ public void onSuccess(Message.Response response) {
                   targetKeyspace.compareAndSet(attempt, defaultKeyspaceAttempt);
                   if (response.type == ERROR) {
                     Responses.Error error = (Responses.Error) response;
-                    ksFuture.setException(defunct(error.asException(endPoint)));
+                    DriverException exception = error.asException(endPoint);
+                    if (exception instanceof QueryValidationException) {
+                      ksFuture.setException(exception);
+                    } else {
+                      ksFuture.setException(defunct(exception));
+                    }
                   } else {
                     ksFuture.setException(
                         defunct(

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

@@ -27,6 +27,7 @@
 import static com.datastax.driver.core.Connection.State.TRASHED;
 
 import com.datastax.driver.core.exceptions.AuthenticationException;
+import com.datastax.driver.core.exceptions.BusyConnectionException;
 import com.datastax.driver.core.exceptions.BusyPoolException;
 import com.datastax.driver.core.exceptions.ConnectionException;
 import com.datastax.driver.core.exceptions.UnsupportedProtocolVersionException;
@@ -611,7 +612,22 @@ ListenableFuture<Connection> borrowConnection(
       if (totalInFlightCount > currentCapacity) maybeSpawnNewConnection(shardId);
     }
 
-    return leastBusy.setKeyspaceAsync(manager.poolsState.keyspace);
+    final Connection borrowedConnection = leastBusy;
+    ListenableFuture<Connection> setKeyspaceFuture =
+        borrowedConnection.setKeyspaceAsync(manager.poolsState.keyspace);
+    Futures.addCallback(
+        setKeyspaceFuture,
+        new FutureCallback<Connection>() {
+          @Override
+          public void onSuccess(Connection connection) {}
+
+          @Override
+          public void onFailure(Throwable t) {
+            borrowedConnection.release(t instanceof BusyConnectionException);
+          }
+        },
+        MoreExecutors.directExecutor());
+    return setKeyspaceFuture;
   }
 
   private ListenableFuture<Connection> enqueue(

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

@@ -24,13 +24,13 @@
 import com.codahale.metrics.Timer;
 import com.datastax.driver.core.exceptions.BootstrappingException;
 import com.datastax.driver.core.exceptions.BusyConnectionException;
-import com.datastax.driver.core.exceptions.BusyPoolException;
 import com.datastax.driver.core.exceptions.ConnectionException;
 import com.datastax.driver.core.exceptions.DriverException;
 import com.datastax.driver.core.exceptions.DriverInternalError;
 import com.datastax.driver.core.exceptions.NoHostAvailableException;
 import com.datastax.driver.core.exceptions.OperationTimedOutException;
 import com.datastax.driver.core.exceptions.OverloadedException;
+import com.datastax.driver.core.exceptions.QueryValidationException;
 import com.datastax.driver.core.exceptions.ReadFailureException;
 import com.datastax.driver.core.exceptions.ReadTimeoutException;
 import com.datastax.driver.core.exceptions.ServerError;
@@ -468,15 +468,11 @@ public void onSuccess(Connection connection) {
 
             @Override
             public void onFailure(Throwable t) {
-              if (t instanceof BusyPoolException) {
-                logError(host.getEndPoint(), t);
-              } else {
-                logger.warn(
-                    "Unexpected error while querying {} - [{}]. Find next host to query.",
-                    host.getEndPoint(),
-                    t.toString());
-                logError(host.getEndPoint(), t);
+              if (t instanceof QueryValidationException) {
+                setFinalException(null, (QueryValidationException) t);
+                return;
               }
+              logError(host.getEndPoint(), t);
               findNextHostAndQuery();
             }
           },

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

@@ -34,8 +34,8 @@
  *   <li>On a write timeout, retries once on the same host if we timeout while writing the
  *       distributed log used by batch statements.
  *   <li>On an unavailable exception, retries once on the next host.
- *   <li>On a request error, such as a client timeout, retries once on the next host. Do not retry
- *       on read or write failures.
+ *   <li>On a request error, such as a client timeout, the query is retried on the next host. Do not
+ *       retry on read or write failures.
  * </ul>
  *
  * <p>This retry policy is conservative in that it will never retry with a different consistency
@@ -136,27 +136,16 @@ public RetryDecision onUnavailable(
     return (nbRetry == 0) ? RetryDecision.tryNextHost(null) : RetryDecision.rethrow();
   }
 
-  /**
-   * {@inheritDoc}
-   *
-   * <p>This implementation triggers a maximum of one retry on the next host in the query plan. The
-   * rationale is that the first coordinator might have been network-isolated or overloaded, and
-   * moving to the next host might resolve the issue. If the retry also fails, the exception is
-   * rethrown.
-   *
-   * <p>Read and write failures are never retried, as they generally indicate a data problem that is
-   * unlikely to be resolved by a retry.
-   *
-   * @return {@code RetryDecision.tryNextHost(cl)} if no retry attempt has yet been tried and the
-   *     error is not a read/write failure, {@code RetryDecision.rethrow()} otherwise.
-   */
+  /** {@inheritDoc} */
   @Override
   public RetryDecision onRequestError(
       Statement statement, ConsistencyLevel cl, DriverException e, int nbRetry) {
+    // do not retry these by default as they generally indicate a data problem or
+    // other issue that is unlikely to be resolved by a retry.
     if (e instanceof WriteFailureException || e instanceof ReadFailureException) {
       return RetryDecision.rethrow();
     }
-    return (nbRetry == 0) ? RetryDecision.tryNextHost(cl) : RetryDecision.rethrow();
+    return RetryDecision.tryNextHost(cl);
   }
 
   @Override

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

@@ -208,24 +208,21 @@ public RetryDecision onUnavailable(
   /**
    * {@inheritDoc}
    *
-   * <p>This implementation triggers a maximum of one retry on the next host in the query plan. The
-   * rationale is that the first coordinator might have been network-isolated or overloaded, and
-   * moving to the next host might resolve the issue. If the retry also fails, the exception is
-   * rethrown.
-   *
-   * <p>Read and write failures are never retried, as they generally indicate a data problem that is
-   * unlikely to be resolved by a retry.
-   *
-   * @return {@code RetryDecision.tryNextHost(cl)} if no retry attempt has yet been tried and the
-   *     error is not a read/write failure, {@code RetryDecision.rethrow()} otherwise.
+   * <p>For historical reasons, this implementation triggers a retry on the next host in the query
+   * plan with the same consistency level, regardless of the statement's idempotence. Note that this
+   * breaks the general rule stated in {@link RetryPolicy#onRequestError(Statement,
+   * ConsistencyLevel, DriverException, int)}: "a retry should only be attempted if the request is
+   * known to be idempotent".`
    */
   @Override
   public RetryDecision onRequestError(
       Statement statement, ConsistencyLevel cl, DriverException e, int nbRetry) {
+    // do not retry these by default as they generally indicate a data problem or
+    // other issue that is unlikely to be resolved by a retry.
     if (e instanceof WriteFailureException || e instanceof ReadFailureException) {
       return RetryDecision.rethrow();
     }
-    return (nbRetry == 0) ? RetryDecision.tryNextHost(cl) : RetryDecision.rethrow();
+    return RetryDecision.tryNextHost(cl);
   }
 
   @Override