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.

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;
@@ -848,14 +849,14 @@ void setKeyspace(String keyspace) throws ConnectionException {
     } 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");
+      throw keyspaceBusyException();
     } 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,6 +870,14 @@ void setKeyspace(String keyspace) throws ConnectionException {
     }
   }
 
+  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)
       throws ConnectionException, BusyConnectionException {
     SetKeyspaceAttempt existingAttempt = targetKeyspace.get();
@@ -900,7 +909,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 +934,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 {