3.x: fail fast on permanent keyspace setup failures

Keyspace setup during connection borrow used to treat every failed internal USE as a host or connection failure. That is correct for driver-side failures and transient server errors, but permanent validation errors such as a missing keyspace should fail the user request directly.

Report keyspace setup write failures through setKeyspaceAsync failed futures consistently, removing stale synchronous throws and caller catches. Reset failed keyspace attempts so later attempts are not blocked by stale in-flight state, and restore pool accounting when borrow-time setup fails.

Preserve QueryValidationException from internal USE without defuncting the connection, and have RequestHandler fail fast for those permanent errors. Keep internal driver failures and transient server-side errors on the normal next-host path before the user query is written, logging those pre-write host-attempt failures only at debug level.

Preserve the synchronous setKeyspace busy-connection behavior after async failures are reported through futures: BusyConnectionException is still converted to a non-defuncting ConnectionException, while ConnectionException keeps the previous defunct behavior.

Add focused coverage for defunct connections, pool accounting, permanent validation errors, transient server errors, retry-policy accounting, single-connection-pool next-host failover, and synchronous busy keyspace setup. Release test ByteBufs used to synthesize server errors.

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,13 @@ 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 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 +866,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 +904,17 @@ 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);
+          throw e;
+        }
         Futures.addCallback(
             future,
             new FutureCallback<Message.Response>() {
@@ -915,7 +929,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();
             }
           },