From bf77fd5dff95a5e4e40490175e038ddd0081fc38 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Fri, 3 Jul 2026 08:25:36 -0400 Subject: [PATCH] 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. --- .../com/datastax/driver/core/Connection.java | 48 +- .../driver/core/HostConnectionPool.java | 18 +- .../datastax/driver/core/RequestHandler.java | 14 +- .../core/policies/DefaultRetryPolicy.java | 23 +- .../DowngradingConsistencyRetryPolicy.java | 19 +- .../driver/core/HostConnectionPoolTest.java | 706 ++++++++++++++++++ .../driver/core/ScassandraCluster.java | 22 - .../DefaultRetryPolicyIntegrationTest.java | 142 ++-- ...ConsistencyRetryPolicyIntegrationTest.java | 152 ++-- 9 files changed, 879 insertions(+), 265 deletions(-) diff --git a/driver-core/src/main/java/com/datastax/driver/core/Connection.java b/driver-core/src/main/java/com/datastax/driver/core/Connection.java index 3c7cc6acf9e..8101875239b 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/Connection.java +++ b/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 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 setKeyspaceAsync(final String keyspace) { SetKeyspaceAttempt existingAttempt = targetKeyspace.get(); if (MoreObjects.equal(existingAttempt.keyspace, keyspace)) return existingAttempt.future; @@ -900,7 +906,18 @@ ListenableFuture 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() { @@ -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( diff --git a/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java b/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java index e5a00e9c2a9..811d648bf73 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java +++ b/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 borrowConnection( if (totalInFlightCount > currentCapacity) maybeSpawnNewConnection(shardId); } - return leastBusy.setKeyspaceAsync(manager.poolsState.keyspace); + final Connection borrowedConnection = leastBusy; + ListenableFuture setKeyspaceFuture = + borrowedConnection.setKeyspaceAsync(manager.poolsState.keyspace); + Futures.addCallback( + setKeyspaceFuture, + new FutureCallback() { + @Override + public void onSuccess(Connection connection) {} + + @Override + public void onFailure(Throwable t) { + borrowedConnection.release(t instanceof BusyConnectionException); + } + }, + MoreExecutors.directExecutor()); + return setKeyspaceFuture; } private ListenableFuture enqueue( diff --git a/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java b/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java index 429ccc1535b..627ea0194fe 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java +++ b/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(); } }, diff --git a/driver-core/src/main/java/com/datastax/driver/core/policies/DefaultRetryPolicy.java b/driver-core/src/main/java/com/datastax/driver/core/policies/DefaultRetryPolicy.java index 33262030be5..dc8fed4262b 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/policies/DefaultRetryPolicy.java +++ b/driver-core/src/main/java/com/datastax/driver/core/policies/DefaultRetryPolicy.java @@ -34,8 +34,8 @@ *
  • On a write timeout, retries once on the same host if we timeout while writing the * distributed log used by batch statements. *
  • On an unavailable exception, retries once on the next host. - *
  • On a request error, such as a client timeout, retries once on the next host. Do not retry - * on read or write failures. + *
  • 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. * * *

    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} - * - *

    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. - * - *

    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 diff --git a/driver-core/src/main/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicy.java b/driver-core/src/main/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicy.java index af3a0fc4a1e..27eaf72066c 100644 --- a/driver-core/src/main/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicy.java +++ b/driver-core/src/main/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicy.java @@ -208,24 +208,21 @@ public RetryDecision onUnavailable( /** * {@inheritDoc} * - *

    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. - * - *

    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. + *

    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 diff --git a/driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java b/driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java index 88964995887..73688516c58 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java @@ -27,10 +27,12 @@ import static com.google.common.collect.Lists.newArrayList; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.mockito.Matchers.anyBoolean; import static org.mockito.Matchers.anyInt; import static org.mockito.Mockito.after; import static org.mockito.Mockito.any; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -50,8 +52,14 @@ 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.InvalidQueryException; import com.datastax.driver.core.exceptions.NoHostAvailableException; +import com.datastax.driver.core.exceptions.OperationTimedOutException; +import com.datastax.driver.core.exceptions.ReadTimeoutException; +import com.datastax.driver.core.exceptions.ServerError; import com.datastax.driver.core.policies.ConstantReconnectionPolicy; +import com.datastax.driver.core.policies.DefaultRetryPolicy; +import com.datastax.driver.core.policies.RetryPolicy; import com.google.common.base.Function; import com.google.common.base.Predicate; import com.google.common.base.Throwables; @@ -62,6 +70,9 @@ import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.Uninterruptibles; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.util.CharsetUtil; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.Collection; @@ -78,6 +89,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import org.scassandra.Scassandra; import org.scassandra.cql.PrimitiveType; import org.scassandra.http.client.PrimingRequest; import org.testng.annotations.BeforeClass; @@ -134,6 +146,48 @@ private void assertBorrowedConnection( assertBorrowedConnections(requests, Collections.singletonList(expectedConnection)); } + private static Responses.Error errorResponse(ExceptionCode code, String message) { + ByteBuf body = Unpooled.buffer(); + try { + body.writeInt(code.value); + byte[] messageBytes = message.getBytes(CharsetUtil.UTF_8); + body.writeShort(messageBytes.length); + body.writeBytes(messageBytes); + return Responses.Error.decoder.decode( + body, ProtocolVersion.V4, CodecRegistry.DEFAULT_INSTANCE, ProtocolFeatureStore.EMPTY); + } finally { + body.release(); + } + } + + private static void assertRetryPolicyNotCalled(RetryPolicy retryPolicy) { + verify(retryPolicy, never()) + .onRequestError( + any(Statement.class), + any(ConsistencyLevel.class), + any(DriverException.class), + anyInt()); + verify(retryPolicy, never()) + .onReadTimeout( + any(Statement.class), + any(ConsistencyLevel.class), + anyInt(), + anyInt(), + anyBoolean(), + anyInt()); + verify(retryPolicy, never()) + .onWriteTimeout( + any(Statement.class), + any(ConsistencyLevel.class), + any(WriteType.class), + anyInt(), + anyInt(), + anyInt()); + verify(retryPolicy, never()) + .onUnavailable( + any(Statement.class), any(ConsistencyLevel.class), anyInt(), anyInt(), anyInt()); + } + /** * Ensures that if a fixed-sized pool has filled its core connections and reached its maximum * number of enqueued requests, then borrowConnection will fail instead of creating a new @@ -393,6 +447,645 @@ public void should_adjust_connection_keyspace_on_dequeue_if_pool_state_is_differ } } + /** + * Ensures that if keyspace setup fails while borrowing a connection, the borrow future fails and + * pool accounting is restored. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_restore_pool_accounting_when_keyspace_setup_fails_on_borrow() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + + pool.manager.poolsState.setKeyspace("newkeyspace"); + ConnectionException failure = + new ConnectionException(pool.host.getEndPoint(), "Write attempt on defunct connection"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("newkeyspace"); + + MockRequest request = MockRequest.send(pool); + + try { + Uninterruptibles.getUninterruptibly(request.connectionFuture, 5, TimeUnit.SECONDS); + fail("Should have failed to borrow connection"); + } catch (ExecutionException e) { + assertThat(e.getCause()).isSameAs(failure); + } + assertThat(connection.inFlight.get()).isEqualTo(0); + assertThat(pool.totalInFlight.get()).isEqualTo(0); + } finally { + cluster.close(); + } + } + + /** + * Ensures that an unexpected synchronous failure while starting keyspace setup is reported + * through the borrow future and does not leak pool accounting. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_restore_pool_accounting_when_keyspace_setup_throws_on_borrow() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + + pool.manager.poolsState.setKeyspace("newkeyspace"); + RuntimeException failure = new RuntimeException("Unexpected keyspace setup failure"); + doThrow(failure).when(connection).write(any(Message.Request.class)); + + MockRequest request = MockRequest.send(pool); + + try { + Uninterruptibles.getUninterruptibly(request.connectionFuture, 5, TimeUnit.SECONDS); + fail("Should have failed to borrow connection"); + } catch (ExecutionException e) { + assertThat(e.getCause()).isSameAs(failure); + } + verify(connection).write(any(Message.Request.class)); + assertThat(connection.inFlight.get()).isEqualTo(0); + assertThat(pool.totalInFlight.get()).isEqualTo(0); + } finally { + cluster.close(); + } + } + + /** + * Ensures that a synchronous write failure while setting the keyspace is reported through the + * returned future and does not leave the failed keyspace attempt in-flight. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_fail_keyspace_future_and_clear_attempt_when_connection_is_defunct() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = pool.connections[0].get(0); + + connection.defunct( + new ConnectionException(pool.host.getEndPoint(), "Test defunct connection")); + + ListenableFuture firstAttempt = connection.setKeyspaceAsync("newkeyspace"); + ListenableFuture secondAttempt = connection.setKeyspaceAsync("newkeyspace"); + + assertThat(firstAttempt).isNotSameAs(secondAttempt); + try { + Uninterruptibles.getUninterruptibly(firstAttempt, 5, TimeUnit.SECONDS); + fail("Should have failed keyspace attempt"); + } catch (ExecutionException e) { + assertThat(e.getCause()) + .isInstanceOf(ConnectionException.class) + .hasMessageContaining("Write attempt on defunct connection"); + } + try { + Uninterruptibles.getUninterruptibly(secondAttempt, 5, TimeUnit.SECONDS); + fail("Should have failed keyspace attempt"); + } catch (ExecutionException e) { + assertThat(e.getCause()) + .isInstanceOf(ConnectionException.class) + .hasMessageContaining("Write attempt on defunct connection"); + } + } finally { + cluster.close(); + } + } + + /** + * Ensures that keyspace validation errors are surfaced as query errors and do not defunct the + * connection. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_not_defunct_connection_when_keyspace_setup_fails_with_validation_error() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + SettableConnectionFuture useFuture = + new SettableConnectionFuture(new Requests.Query("USE \"missingkeyspace\"")); + + doReturn(useFuture).when(connection).write(any(Message.Request.class)); + ListenableFuture keyspaceFuture = connection.setKeyspaceAsync("missingkeyspace"); + useFuture.setResponse( + errorResponse(ExceptionCode.INVALID, "Keyspace missingkeyspace does not exist")); + + try { + Uninterruptibles.getUninterruptibly(keyspaceFuture, 5, TimeUnit.SECONDS); + fail("Should have failed keyspace attempt"); + } catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(InvalidQueryException.class); + } + assertThat(connection.isDefunct()).isFalse(); + assertThat(pool.opened()).isEqualTo(1); + } finally { + cluster.close(); + } + } + + /** + * Ensures that the synchronous keyspace setup wrapper preserves validation errors and does not + * defunct the connection. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_not_defunct_connection_when_synchronous_keyspace_setup_gets_validation_error() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + InvalidQueryException failure = + new InvalidQueryException( + pool.host.getEndPoint(), "Keyspace missingkeyspace does not exist"); + + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("missingkeyspace"); + + try { + connection.setKeyspace("missingkeyspace"); + fail("Should have failed keyspace attempt"); + } catch (InvalidQueryException e) { + assertThat(e).isSameAs(failure); + } + assertThat(connection.isDefunct()).isFalse(); + assertThat(pool.opened()).isEqualTo(1); + } finally { + cluster.close(); + } + } + + /** + * Ensures that transient server errors during keyspace setup fail the attempt and defunct the + * connection. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_defunct_connection_when_keyspace_setup_fails_with_server_error() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + SettableConnectionFuture useFuture = + new SettableConnectionFuture(new Requests.Query("USE \"newkeyspace\"")); + + doReturn(useFuture).when(connection).write(any(Message.Request.class)); + ListenableFuture keyspaceFuture = connection.setKeyspaceAsync("newkeyspace"); + useFuture.setResponse(errorResponse(ExceptionCode.SERVER_ERROR, "Server Error")); + + try { + Uninterruptibles.getUninterruptibly(keyspaceFuture, 5, TimeUnit.SECONDS); + fail("Should have failed keyspace attempt"); + } catch (ExecutionException e) { + assertThat(e.getCause()).isInstanceOf(ServerError.class); + } + assertThat(connection.isDefunct()).isTrue(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that the synchronous keyspace setup wrapper preserves busy connection handling when the + * async implementation reports write failures through the returned future. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_not_defunct_connection_when_synchronous_keyspace_setup_gets_busy_connection() + throws Exception { + Cluster cluster = createClusterBuilder().build(); + try { + HostConnectionPool pool = createPool(cluster, 1, 1); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + + doThrow(new BusyConnectionException(pool.host.getEndPoint())) + .when(connection) + .write(any(Message.Request.class)); + + try { + connection.setKeyspace("newkeyspace"); + fail("Should have failed keyspace attempt"); + } catch (ConnectionException e) { + assertThat(e.getMessage()).contains("Tried to set the keyspace on busy connection"); + } + assertThat(connection.isDefunct()).isFalse(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that permanent keyspace setup failures fail the request directly instead of being + * retried across hosts as connection failures. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_fail_fast_when_keyspace_setup_fails_with_validation_error() throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + InvalidQueryException failure = + new InvalidQueryException( + pool.host.getEndPoint(), "Keyspace missingkeyspace does not exist"); + session.poolsState.setKeyspace("missingkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("missingkeyspace"); + activityClient.clearAllRecordedActivity(); + + try { + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + fail("Should have failed the request with the keyspace validation error"); + } catch (InvalidQueryException e) { + // expected + } + assertRetryPolicyNotCalled(retryPolicy); + assertThat(connection.isDefunct()).isFalse(); + assertThat(pool.opened()).isEqualTo(1); + assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that a 0x2200 INVALID response while setting the keyspace is treated as a final query + * validation error and does not retry the user query. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_fail_fast_when_keyspace_setup_receives_invalid_error_response() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + SettableConnectionFuture useFuture = + new SettableConnectionFuture(new Requests.Query("USE \"missingkeyspace\"")); + session.poolsState.setKeyspace("missingkeyspace"); + doReturn(useFuture).when(connection).write(any(Message.Request.class)); + activityClient.clearAllRecordedActivity(); + + ResultSetFuture queryFuture = + session.executeAsync(new SimpleStatement("select * from tbl").setIdempotent(true)); + useFuture.setResponse( + errorResponse(ExceptionCode.INVALID, "Keyspace missingkeyspace does not exist")); + + try { + queryFuture.getUninterruptibly(5, TimeUnit.SECONDS); + fail("Should have failed the request with the keyspace validation error"); + } catch (InvalidQueryException e) { + assertThat(e.getMessage()).contains("Keyspace missingkeyspace does not exist"); + } + assertRetryPolicyNotCalled(retryPolicy); + assertThat(connection.isDefunct()).isFalse(); + assertThat(pool.opened()).isEqualTo(1); + verify(connection, times(1)).write(any(Message.Request.class)); + assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that a 0x2200 INVALID response while setting the keyspace does not move the user query + * to the next host. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_not_try_next_host_when_keyspace_setup_receives_invalid_error_response() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + ScassandraCluster scassandras = ScassandraCluster.builder().withNodes(2).build(); + Cluster cluster = null; + try { + scassandras.init(); + scassandras + .node(2) + .primingClient() + .prime( + queryBuilder() + .withQuery("select * from tbl") + .withThen(then().withRows(Collections.singletonMap("result", "result2"))) + .build()); + cluster = + Cluster.builder() + .addContactPoints(scassandras.address(1).getAddress()) + .withPort(scassandras.getBinaryPort()) + .withLoadBalancingPolicy(new SortingLoadBalancingPolicy()) + .withPoolingOptions( + new PoolingOptions() + .setCoreConnectionsPerHost(HostDistance.LOCAL, 1) + .setMaxConnectionsPerHost(HostDistance.LOCAL, 1) + .setHeartbeatIntervalSeconds(0)) + .withRetryPolicy(retryPolicy) + .build(); + SessionManager session = (SessionManager) cluster.connect(); + Host host1 = TestUtils.findHost(cluster, 1); + Host host2 = TestUtils.findHost(cluster, 2); + HostConnectionPool pool1 = session.pools.get(host1); + HostConnectionPool pool2 = session.pools.get(host2); + Connection connection1 = spy(pool1.connections[0].get(0)); + Connection connection2 = spy(pool2.connections[0].get(0)); + pool1.connections[0].set(0, connection1); + pool2.connections[0].set(0, connection2); + SettableConnectionFuture useFuture = + new SettableConnectionFuture(new Requests.Query("USE \"missingkeyspace\"")); + session.poolsState.setKeyspace("missingkeyspace"); + doReturn(useFuture).when(connection1).write(any(Message.Request.class)); + doReturn(Futures.immediateFuture(connection2)) + .when(connection2) + .setKeyspaceAsync("missingkeyspace"); + + for (Scassandra node : scassandras.nodes()) { + node.activityClient().clearAllRecordedActivity(); + } + + ResultSetFuture queryFuture = + session.executeAsync(new SimpleStatement("select * from tbl").setIdempotent(true)); + useFuture.setResponse( + errorResponse(ExceptionCode.INVALID, "Keyspace missingkeyspace does not exist")); + + try { + queryFuture.getUninterruptibly(5, TimeUnit.SECONDS); + fail("Should have failed the request with the keyspace validation error"); + } catch (InvalidQueryException e) { + assertThat(e.getMessage()).contains("Keyspace missingkeyspace does not exist"); + } + assertRetryPolicyNotCalled(retryPolicy); + assertThat(connection1.isDefunct()).isFalse(); + verify(connection1, times(1)).write(any(Message.Request.class)); + verify(connection2, never()).setKeyspaceAsync("missingkeyspace"); + assertThat(scassandras.node(1).activityClient().retrieveQueries()) + .extractingResultOf("getQuery") + .isEmpty(); + assertThat(scassandras.node(2).activityClient().retrieveQueries()) + .extractingResultOf("getQuery") + .isEmpty(); + } finally { + if (cluster != null) { + cluster.close(); + } + scassandras.stop(); + } + } + + /** + * Ensures that transient server-side keyspace setup failures that happen before the user query is + * written do not consume retry policy attempts. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void + should_not_call_retry_policy_when_keyspace_setup_fails_with_server_error_before_query_write() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + ServerError failure = new ServerError(pool.host.getEndPoint(), "Server Error"); + session.poolsState.setKeyspace("newkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("newkeyspace"); + + try { + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + fail("Should have failed after exhausting query plan"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().values()).containsOnly(failure); + } + assertRetryPolicyNotCalled(retryPolicy); + } finally { + cluster.close(); + } + } + + /** + * Ensures that server-side timeout errors during keyspace setup that happen before the user query + * is written do not consume retry policy attempts or send the user query. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void + should_not_call_retry_policy_or_send_query_when_keyspace_setup_fails_with_server_side_timeout() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + ReadTimeoutException failure = + new ReadTimeoutException(pool.host.getEndPoint(), ConsistencyLevel.ONE, 0, 1, false); + session.poolsState.setKeyspace("newkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("newkeyspace"); + activityClient.clearAllRecordedActivity(); + + try { + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + fail("Should have failed after exhausting query plan"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().values()).containsOnly(failure); + } + assertRetryPolicyNotCalled(retryPolicy); + assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that client-side timeouts during keyspace setup that happen before the user query is + * written do not consume retry policy attempts or send the user query. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void + should_not_call_retry_policy_or_send_query_when_keyspace_setup_fails_with_client_side_timeout() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + OperationTimedOutException failure = new OperationTimedOutException(pool.host.getEndPoint()); + session.poolsState.setKeyspace("newkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("newkeyspace"); + activityClient.clearAllRecordedActivity(); + + try { + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + fail("Should have failed after exhausting query plan"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().values()).containsOnly(failure); + } + assertRetryPolicyNotCalled(retryPolicy); + assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty(); + } finally { + cluster.close(); + } + } + + /** + * Ensures that internal driver keyspace setup failures that happen before the user query is + * written do not consume retry policy attempts. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_not_call_retry_policy_when_keyspace_setup_fails_before_query_write() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build(); + try { + SessionManager session = (SessionManager) cluster.connect(); + Host host = TestUtils.findHost(cluster, 1); + HostConnectionPool pool = session.pools.get(host); + Connection connection = spy(pool.connections[0].get(0)); + pool.connections[0].set(0, connection); + ConnectionException failure = + new ConnectionException(pool.host.getEndPoint(), "Write attempt on defunct connection"); + session.poolsState.setKeyspace("newkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection) + .setKeyspaceAsync("newkeyspace"); + + try { + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + fail("Should have failed after exhausting query plan"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().values()).containsOnly(failure); + } + assertRetryPolicyNotCalled(retryPolicy); + } finally { + cluster.close(); + } + } + + /** + * Ensures that an internal keyspace setup failure on the only connection in a host pool moves the + * request to the next host. + * + * @test_category connection:connection_pool + */ + @Test(groups = "short") + public void should_try_next_host_when_single_connection_pool_fails_keyspace_setup() + throws Exception { + RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE); + ScassandraCluster scassandras = ScassandraCluster.builder().withNodes(2).build(); + Cluster cluster = null; + try { + scassandras.init(); + scassandras + .node(2) + .primingClient() + .prime( + queryBuilder() + .withQuery("select * from tbl") + .withThen(then().withRows(Collections.singletonMap("result", "result2"))) + .build()); + cluster = + Cluster.builder() + .addContactPoints(scassandras.address(1).getAddress()) + .withPort(scassandras.getBinaryPort()) + .withLoadBalancingPolicy(new SortingLoadBalancingPolicy()) + .withPoolingOptions( + new PoolingOptions() + .setCoreConnectionsPerHost(HostDistance.LOCAL, 1) + .setMaxConnectionsPerHost(HostDistance.LOCAL, 1) + .setHeartbeatIntervalSeconds(0)) + .withRetryPolicy(retryPolicy) + .build(); + SessionManager session = (SessionManager) cluster.connect(); + Host host1 = TestUtils.findHost(cluster, 1); + Host host2 = TestUtils.findHost(cluster, 2); + HostConnectionPool pool1 = session.pools.get(host1); + HostConnectionPool pool2 = session.pools.get(host2); + Connection connection1 = spy(pool1.connections[0].get(0)); + Connection connection2 = spy(pool2.connections[0].get(0)); + pool1.connections[0].set(0, connection1); + pool2.connections[0].set(0, connection2); + ConnectionException failure = + new ConnectionException(pool1.host.getEndPoint(), "Write attempt on defunct connection"); + session.poolsState.setKeyspace("newkeyspace"); + doReturn(Futures.immediateFailedFuture(failure)) + .when(connection1) + .setKeyspaceAsync("newkeyspace"); + doReturn(Futures.immediateFuture(connection2)) + .when(connection2) + .setKeyspaceAsync("newkeyspace"); + + for (Scassandra node : scassandras.nodes()) { + node.activityClient().clearAllRecordedActivity(); + } + + session.execute(new SimpleStatement("select * from tbl").setIdempotent(true)); + + assertRetryPolicyNotCalled(retryPolicy); + assertThat(scassandras.node(1).activityClient().retrieveQueries()) + .extractingResultOf("getQuery") + .isEmpty(); + assertThat(scassandras.node(2).activityClient().retrieveQueries()) + .extractingResultOf("getQuery") + .containsOnly("select * from tbl"); + } finally { + if (cluster != null) { + cluster.close(); + } + scassandras.stop(); + } + } + /** * Ensures that on borrowConnection if a set keyspace attempt is in progress on that connection * for a different keyspace than the pool state that the borrowConnection future returned is @@ -501,6 +1194,9 @@ public void should_adjust_connection_keyspace_on_dequeue_if_pool_state_is_differ .contains( "Aborting attempt to set keyspace to 'newkeyspace' since there is already an in flight attempt to set keyspace to 'slowks'."); } + assertThat(connection.inFlight.get()).isEqualTo(0); + assertThat(pool.totalInFlight.get()).isEqualTo(0); + assertThat(pool.pendingBorrowCount.get()).isEqualTo(0); } finally { MockRequest.completeAll(requests); cluster.close(); @@ -1743,4 +2439,14 @@ public void onFailure(Throwable t) { return future; } } + + private static class SettableConnectionFuture extends Connection.Future { + SettableConnectionFuture(Message.Request request) { + super(request); + } + + void setResponse(Message.Response response) { + super.set(response); + } + } } diff --git a/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java b/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java index f2fa5c6fc69..341c9401e74 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java +++ b/driver-core/src/test/java/com/datastax/driver/core/ScassandraCluster.java @@ -16,7 +16,6 @@ package com.datastax.driver.core; import static com.datastax.driver.core.Assertions.assertThat; -import static com.datastax.driver.core.ConditionChecker.check; import static org.scassandra.cql.MapType.map; import static org.scassandra.cql.PrimitiveType.BIG_INT; import static org.scassandra.cql.PrimitiveType.BOOLEAN; @@ -46,7 +45,6 @@ import java.util.Map; import java.util.Set; import java.util.TreeSet; -import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import org.scassandra.Scassandra; import org.scassandra.ScassandraFactory; @@ -330,32 +328,12 @@ public void stopDC(Cluster cluster, int dc) { public void stop(Cluster cluster, int node) { logger.debug("Stopping node {}.", node); Scassandra scassandra = node(node); - Host host = TestUtils.findHost(cluster, node); try { scassandra.stop(); } catch (Exception e) { logger.error("Could not stop node " + scassandra, e); } assertThat(cluster).host(node).goesDownWithin(60, TimeUnit.SECONDS); - waitUntilPoolsAreRemoved(cluster, host); - } - - private void waitUntilPoolsAreRemoved(final Cluster cluster, final Host host) { - if (host == null) return; - - check() - .before(60, TimeUnit.SECONDS) - .that( - new Callable() { - @Override - public Boolean call() { - for (SessionManager session : cluster.manager.sessions) { - if (session.pools.containsKey(host)) return false; - } - return true; - } - }) - .becomesTrue(); } /** diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/DefaultRetryPolicyIntegrationTest.java b/driver-core/src/test/java/com/datastax/driver/core/policies/DefaultRetryPolicyIntegrationTest.java index 3c14fbd8f2f..aaa25e1d9d4 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/policies/DefaultRetryPolicyIntegrationTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/policies/DefaultRetryPolicyIntegrationTest.java @@ -179,38 +179,7 @@ public void should_rethrow_unavailable_in_no_host_available_exception() { } @Test(groups = "short") - public void should_try_next_host_on_first_client_timeout() { - cluster.getConfiguration().getSocketOptions().setReadTimeoutMillis(1); - try { - scassandras - .node(1) - .primingClient() - .prime( - PrimingRequest.queryBuilder() - .withQuery("mock query") - .withThen(then().withFixedDelay(1000L).withRows(row("result", "result1"))) - .build()); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, OperationTimedOutException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getClientTimeouts().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } finally { - cluster - .getConfiguration() - .getSocketOptions() - .setReadTimeoutMillis(SocketOptions.DEFAULT_READ_TIMEOUT_MILLIS); - } - } - - @Test(groups = "short") - public void should_rethrow_on_second_client_timeout() { + public void should_try_next_host_on_client_timeouts() { cluster.getConfiguration().getSocketOptions().setReadTimeoutMillis(1); try { scassandras @@ -239,17 +208,32 @@ public void should_rethrow_on_second_client_timeout() { .build()); try { query(); - fail("expected an OperationTimedOutException"); - } catch (OperationTimedOutException e) { - assertThat(e.getMessage()).contains("Timed out waiting for server response"); + fail("expected a NoHostAvailableException"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(OperationTimedOutException.class); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host1.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host1.getEndPoint())); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host2.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host2.getEndPoint())); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host3.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host3.getEndPoint())); } - assertOnRequestErrorWasCalled(2, OperationTimedOutException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getClientTimeouts().getCount()).isEqualTo(2); - assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(1); + assertOnRequestErrorWasCalled(3, OperationTimedOutException.class); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getClientTimeouts().getCount()).isEqualTo(3); + assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } finally { cluster .getConfiguration() @@ -259,41 +243,27 @@ public void should_rethrow_on_second_client_timeout() { } @Test(groups = "short", dataProvider = "serverSideErrors") - public void should_try_next_host_on_first_server_side_error( - Result error, Class exception) { - simulateError(1, error); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, exception); - assertThat(errors.getOthers().getCount()).isEqualTo(1); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } - - @Test(groups = "short", dataProvider = "serverSideErrors") - public void should_rethrow_on_second_server_side_error( + public void should_try_next_host_on_server_side_error( Result error, Class exception) { simulateError(1, error); simulateError(2, error); simulateError(3, error); try { query(); - Fail.fail("expected a " + exception.getSimpleName()); - } catch (DriverException e) { - assertThat(e).isInstanceOf(exception); + Fail.fail("expected a NoHostAvailableException"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(exception); } - assertOnRequestErrorWasCalled(2, exception); - assertThat(errors.getOthers().getCount()).isEqualTo(2); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(1); + assertOnRequestErrorWasCalled(3, exception); + assertThat(errors.getOthers().getCount()).isEqualTo(3); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } @Test(groups = "short") @@ -337,43 +307,27 @@ public void should_rethrow_on_write_failure() { } @Test(groups = "short", dataProvider = "connectionErrors") - public void should_try_next_host_on_first_connection_error( - ClosedConnectionConfig.CloseType closeType) { - simulateError(1, closed_connection, new ClosedConnectionConfig(closeType)); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, TransportException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getConnectionErrors().getCount()).isEqualTo(1); - assertThat(errors.getIgnoresOnConnectionError().getCount()).isEqualTo(0); - assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } - - @Test(groups = "short", dataProvider = "connectionErrors") - public void should_rethrow_on_second_connection_error( - ClosedConnectionConfig.CloseType closeType) { + public void should_try_next_host_on_connection_error(ClosedConnectionConfig.CloseType closeType) { simulateError(1, closed_connection, new ClosedConnectionConfig(closeType)); simulateError(2, closed_connection, new ClosedConnectionConfig(closeType)); simulateError(3, closed_connection, new ClosedConnectionConfig(closeType)); try { query(); - Fail.fail("expected a TransportException"); - } catch (TransportException e) { - // expected — rethrown after one retry + Fail.fail("expected a NoHostAvailableException"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(TransportException.class); } - assertOnRequestErrorWasCalled(2, TransportException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getConnectionErrors().getCount()).isEqualTo(2); + assertOnRequestErrorWasCalled(3, TransportException.class); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getConnectionErrors().getCount()).isEqualTo(3); assertThat(errors.getIgnoresOnConnectionError().getCount()).isEqualTo(0); - assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(1); + assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } @Test(groups = "short") diff --git a/driver-core/src/test/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicyIntegrationTest.java b/driver-core/src/test/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicyIntegrationTest.java index 7745c7e2216..7a305358ec2 100644 --- a/driver-core/src/test/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicyIntegrationTest.java +++ b/driver-core/src/test/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicyIntegrationTest.java @@ -31,6 +31,7 @@ import com.datastax.driver.core.SocketOptions; import com.datastax.driver.core.WriteType; import com.datastax.driver.core.exceptions.DriverException; +import com.datastax.driver.core.exceptions.NoHostAvailableException; import com.datastax.driver.core.exceptions.OperationTimedOutException; import com.datastax.driver.core.exceptions.ReadFailureException; import com.datastax.driver.core.exceptions.ReadTimeoutException; @@ -369,43 +370,13 @@ public void should_rethrow_if_no_hosts_alive_on_unavailable() { /** * Ensures that when handling a client timeout with {@link DowngradingConsistencyRetryPolicy} that - * a retry is attempted on the next host. If the retry also times out, the exception is rethrown. + * a retry is attempted on the next host until all hosts are tried at which point a {@link + * NoHostAvailableException} is returned. * * @test_category retry_policy */ @Test(groups = "short") - public void should_try_next_host_on_first_client_timeout() { - cluster.getConfiguration().getSocketOptions().setReadTimeoutMillis(1); - try { - scassandras - .node(1) - .primingClient() - .prime( - PrimingRequest.queryBuilder() - .withQuery("mock query") - .withThen(then().withFixedDelay(1000L).withRows(row("result", "result1"))) - .build()); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, OperationTimedOutException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getClientTimeouts().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } finally { - cluster - .getConfiguration() - .getSocketOptions() - .setReadTimeoutMillis(SocketOptions.DEFAULT_READ_TIMEOUT_MILLIS); - } - } - - @Test(groups = "short") - public void should_rethrow_on_second_client_timeout() { + public void should_try_next_host_on_client_timeouts() { cluster.getConfiguration().getSocketOptions().setReadTimeoutMillis(1); try { scassandras @@ -434,17 +405,32 @@ public void should_rethrow_on_second_client_timeout() { .build()); try { query(); - Assertions.fail("expected an OperationTimedOutException"); - } catch (OperationTimedOutException e) { - assertThat(e.getMessage()).contains("Timed out waiting for server response"); + Assertions.fail("expected a NoHostAvailableException"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(OperationTimedOutException.class); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host1.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host1.getEndPoint())); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host2.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host2.getEndPoint())); + assertThat( + ((OperationTimedOutException) e.getErrors().get(host3.getEndPoint())).getMessage()) + .contains( + String.format("[%s] Timed out waiting for server response", host3.getEndPoint())); } - assertOnRequestErrorWasCalled(2, OperationTimedOutException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getClientTimeouts().getCount()).isEqualTo(2); - assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(1); + assertOnRequestErrorWasCalled(3, OperationTimedOutException.class); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getClientTimeouts().getCount()).isEqualTo(3); + assertThat(errors.getRetriesOnClientTimeout().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } finally { cluster .getConfiguration() @@ -455,96 +441,66 @@ public void should_rethrow_on_second_client_timeout() { /** * Ensures that when handling a server error defined in {@link #serverSideErrors} with {@link - * DowngradingConsistencyRetryPolicy} that a retry is attempted on the next host. If the retry - * also fails, the exception is rethrown. + * DowngradingConsistencyRetryPolicy} that a retry is attempted on the next host until all hosts + * are tried at which point a {@link NoHostAvailableException} is raised and it's errors include + * the expected exception. * * @param error Server side error to be produced. * @param exception The exception we expect to be raised. * @test_category retry_policy */ @Test(groups = "short", dataProvider = "serverSideErrors") - public void should_try_next_host_on_first_server_side_error( - Result error, Class exception) { - simulateError(1, error); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, exception); - assertThat(errors.getOthers().getCount()).isEqualTo(1); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } - - @Test(groups = "short", dataProvider = "serverSideErrors") - public void should_rethrow_on_second_server_side_error( + public void should_try_next_host_on_server_side_error( Result error, Class exception) { simulateError(1, error); simulateError(2, error); simulateError(3, error); try { query(); - Fail.fail("expected a " + exception.getSimpleName()); - } catch (DriverException e) { - assertThat(e).isInstanceOf(exception); + Fail.fail("expected a NoHostAvailableException"); + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(exception); } - assertOnRequestErrorWasCalled(2, exception); - assertThat(errors.getOthers().getCount()).isEqualTo(2); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(1); + assertOnRequestErrorWasCalled(3, exception); + assertThat(errors.getOthers().getCount()).isEqualTo(3); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } /** * Ensures that when handling a connection error caused by the connection closing during a request - * in a way described by {@link #connectionErrors} that the next host is tried. If the retry also - * fails, the exception is rethrown. + * in a way described by {@link #connectionErrors} that the next host is tried. * * @param closeType The way the connection should be closed during the request. */ @Test(groups = "short", dataProvider = "connectionErrors") - public void should_try_next_host_on_first_connection_error( - ClosedConnectionConfig.CloseType closeType) { - simulateError(1, closed_connection, new ClosedConnectionConfig(closeType)); - simulateNormalResponse(2); - - query(); - - assertOnRequestErrorWasCalled(1, TransportException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getConnectionErrors().getCount()).isEqualTo(1); - assertThat(errors.getIgnoresOnConnectionError().getCount()).isEqualTo(0); - assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(1); - assertQueried(1, 1); - assertQueried(2, 1); - assertQueried(3, 0); - } - - @Test(groups = "short", dataProvider = "connectionErrors") - public void should_rethrow_on_second_connection_error( - ClosedConnectionConfig.CloseType closeType) { + public void should_try_next_host_on_connection_error(ClosedConnectionConfig.CloseType closeType) { simulateError(1, closed_connection, new ClosedConnectionConfig(closeType)); simulateError(2, closed_connection, new ClosedConnectionConfig(closeType)); simulateError(3, closed_connection, new ClosedConnectionConfig(closeType)); try { query(); Fail.fail("expected a TransportException"); - } catch (TransportException e) { - // expected — rethrown after one retry + } catch (NoHostAvailableException e) { + assertThat(e.getErrors().keySet()) + .hasSize(3) + .containsOnly(host1.getEndPoint(), host2.getEndPoint(), host3.getEndPoint()); + assertThat(e.getErrors().values()).hasOnlyElementsOfType(TransportException.class); } - assertOnRequestErrorWasCalled(2, TransportException.class); - assertThat(errors.getRetries().getCount()).isEqualTo(1); - assertThat(errors.getConnectionErrors().getCount()).isEqualTo(2); + assertOnRequestErrorWasCalled(3, TransportException.class); + assertThat(errors.getRetries().getCount()).isEqualTo(3); + assertThat(errors.getConnectionErrors().getCount()).isEqualTo(3); assertThat(errors.getIgnoresOnConnectionError().getCount()).isEqualTo(0); - assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(1); + assertThat(errors.getRetriesOnConnectionError().getCount()).isEqualTo(3); assertQueried(1, 1); assertQueried(2, 1); - assertQueried(3, 0); + assertQueried(3, 1); } @Test(groups = "short")