3.x: fail fast on keyspace setup validation failures

Fixes #940.

Part of #939.

Jira: https://scylladb.atlassian.net/browse/DRIVER-751

Keyspace setup during connection borrow runs an internal USE before the user query is written. Treat validation failures from that USE as permanent request errors instead of connection or host failures: preserve QueryValidationException, avoid defuncting the connection, and stop the request without trying the next host or invoking retry policy.

Keep driver-side and transient server-side keyspace setup failures on the existing pre-query next-host path. Report synchronous setKeyspaceAsync write failures through failed futures, clear failed attempts, and restore pool accounting for borrow and dequeue failures.

Add regression coverage for 0x2200 INVALID responses, synchronous setKeyspace validation handling, client/server-side keyspace setup timeouts, pool accounting cleanup, retry-policy accounting, and next-host behavior.

Author: dkropachev

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