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")