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.

Return synchronous USE write failures as failed futures so pool accounting is restored. 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.

Add focused coverage for defunct connections, pool accounting, permanent validation errors, transient server errors, retry-policy accounting, and moving from a single-connection host pool to the next host after internal keyspace setup failure.

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;
@@ -900,7 +901,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 +926,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(
@@ -698,8 +714,14 @@ private void dequeue(final Connection connection) {
         pendingBorrowCount.decrementAndGet();
         // Ensure that the keyspace set on the connection is the one set on the pool state, in the
         // general case it will be.
-        ListenableFuture<Connection> setKeyspaceFuture =
-            connection.setKeyspaceAsync(manager.poolsState.keyspace);
+        ListenableFuture<Connection> setKeyspaceFuture;
+        try {
+          setKeyspaceFuture = connection.setKeyspaceAsync(manager.poolsState.keyspace);
+        } catch (ConnectionException | BusyConnectionException e) {
+          pendingBorrow.setException(e);
+          connection.inFlight.decrementAndGet();
+          continue;
+        }
         // Slight optimization, if the keyspace was already correct the future will be complete, so
         // simply complete it here.
         if (setKeyspaceFuture.isDone()) {

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

@@ -31,6 +31,7 @@
 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,6 +469,10 @@ public void onSuccess(Connection connection) {
 
             @Override
             public void onFailure(Throwable t) {
+              if (t instanceof QueryValidationException) {
+                setFinalException(null, (QueryValidationException) t);
+                return;
+              }
               if (t instanceof BusyPoolException) {
                 logError(host.getEndPoint(), t);
               } else {