Skip to content

Commit a562140

Browse files
dkropachevdkropachev
committed
Preserve keyspace validation failures
1 parent d78343f commit a562140

2 files changed

Lines changed: 264 additions & 0 deletions

File tree

‎driver-core/src/main/java/com/datastax/driver/core/Connection.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -850,6 +850,8 @@ void setKeyspace(String keyspace) throws ConnectionException {
850850
Throwable cause = e.getCause();
851851
if (cause instanceof BusyConnectionException) {
852852
throw keyspaceBusyException();
853+
} else if (cause instanceof QueryValidationException) {
854+
throw (QueryValidationException) cause;
853855
} else if (cause instanceof ConnectionException) {
854856
throw defunct((ConnectionException) cause);
855857
} else if (cause instanceof OperationTimedOutException) {

‎driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java‎

Lines changed: 262 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import static com.google.common.collect.Lists.newArrayList;
2828
import static java.util.concurrent.TimeUnit.MILLISECONDS;
2929
import static java.util.concurrent.TimeUnit.SECONDS;
30+
import static org.mockito.Matchers.anyBoolean;
3031
import static org.mockito.Matchers.anyInt;
3132
import static org.mockito.Mockito.after;
3233
import static org.mockito.Mockito.any;
@@ -53,6 +54,8 @@
5354
import com.datastax.driver.core.exceptions.DriverException;
5455
import com.datastax.driver.core.exceptions.InvalidQueryException;
5556
import com.datastax.driver.core.exceptions.NoHostAvailableException;
57+
import com.datastax.driver.core.exceptions.OperationTimedOutException;
58+
import com.datastax.driver.core.exceptions.ReadTimeoutException;
5659
import com.datastax.driver.core.exceptions.ServerError;
5760
import com.datastax.driver.core.policies.ConstantReconnectionPolicy;
5861
import com.datastax.driver.core.policies.DefaultRetryPolicy;
@@ -164,6 +167,25 @@ private static void assertRetryPolicyNotCalled(RetryPolicy retryPolicy) {
164167
any(ConsistencyLevel.class),
165168
any(DriverException.class),
166169
anyInt());
170+
verify(retryPolicy, never())
171+
.onReadTimeout(
172+
any(Statement.class),
173+
any(ConsistencyLevel.class),
174+
anyInt(),
175+
anyInt(),
176+
anyBoolean(),
177+
anyInt());
178+
verify(retryPolicy, never())
179+
.onWriteTimeout(
180+
any(Statement.class),
181+
any(ConsistencyLevel.class),
182+
any(WriteType.class),
183+
anyInt(),
184+
anyInt(),
185+
anyInt());
186+
verify(retryPolicy, never())
187+
.onUnavailable(
188+
any(Statement.class), any(ConsistencyLevel.class), anyInt(), anyInt(), anyInt());
167189
}
168190

169191
/**
@@ -574,6 +596,41 @@ public void should_not_defunct_connection_when_keyspace_setup_fails_with_validat
574596
}
575597
}
576598

599+
/**
600+
* Ensures that the synchronous keyspace setup wrapper preserves validation errors and does not
601+
* defunct the connection.
602+
*
603+
* @test_category connection:connection_pool
604+
*/
605+
@Test(groups = "short")
606+
public void should_not_defunct_connection_when_synchronous_keyspace_setup_gets_validation_error()
607+
throws Exception {
608+
Cluster cluster = createClusterBuilder().build();
609+
try {
610+
HostConnectionPool pool = createPool(cluster, 1, 1);
611+
Connection connection = spy(pool.connections[0].get(0));
612+
pool.connections[0].set(0, connection);
613+
InvalidQueryException failure =
614+
new InvalidQueryException(
615+
pool.host.getEndPoint(), "Keyspace missingkeyspace does not exist");
616+
617+
doReturn(Futures.<Connection>immediateFailedFuture(failure))
618+
.when(connection)
619+
.setKeyspaceAsync("missingkeyspace");
620+
621+
try {
622+
connection.setKeyspace("missingkeyspace");
623+
fail("Should have failed keyspace attempt");
624+
} catch (InvalidQueryException e) {
625+
assertThat(e).isSameAs(failure);
626+
}
627+
assertThat(connection.isDefunct()).isFalse();
628+
assertThat(pool.opened()).isEqualTo(1);
629+
} finally {
630+
cluster.close();
631+
}
632+
}
633+
577634
/**
578635
* Ensures that transient server errors during keyspace setup fail the attempt and defunct the
579636
* connection.
@@ -678,6 +735,134 @@ public void should_fail_fast_when_keyspace_setup_fails_with_validation_error() t
678735
}
679736
}
680737

738+
/**
739+
* Ensures that a 0x2200 INVALID response while setting the keyspace is treated as a final query
740+
* validation error and does not retry the user query.
741+
*
742+
* @test_category connection:connection_pool
743+
*/
744+
@Test(groups = "short")
745+
public void should_fail_fast_when_keyspace_setup_receives_invalid_error_response()
746+
throws Exception {
747+
RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE);
748+
Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build();
749+
try {
750+
SessionManager session = (SessionManager) cluster.connect();
751+
Host host = TestUtils.findHost(cluster, 1);
752+
HostConnectionPool pool = session.pools.get(host);
753+
Connection connection = spy(pool.connections[0].get(0));
754+
pool.connections[0].set(0, connection);
755+
SettableConnectionFuture useFuture =
756+
new SettableConnectionFuture(new Requests.Query("USE \"missingkeyspace\""));
757+
session.poolsState.setKeyspace("missingkeyspace");
758+
doReturn(useFuture).when(connection).write(any(Message.Request.class));
759+
activityClient.clearAllRecordedActivity();
760+
761+
ResultSetFuture queryFuture =
762+
session.executeAsync(new SimpleStatement("select * from tbl").setIdempotent(true));
763+
useFuture.setResponse(
764+
errorResponse(ExceptionCode.INVALID, "Keyspace missingkeyspace does not exist"));
765+
766+
try {
767+
queryFuture.getUninterruptibly(5, TimeUnit.SECONDS);
768+
fail("Should have failed the request with the keyspace validation error");
769+
} catch (InvalidQueryException e) {
770+
assertThat(e.getMessage()).contains("Keyspace missingkeyspace does not exist");
771+
}
772+
assertRetryPolicyNotCalled(retryPolicy);
773+
assertThat(connection.isDefunct()).isFalse();
774+
assertThat(pool.opened()).isEqualTo(1);
775+
verify(connection, times(1)).write(any(Message.Request.class));
776+
assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty();
777+
} finally {
778+
cluster.close();
779+
}
780+
}
781+
782+
/**
783+
* Ensures that a 0x2200 INVALID response while setting the keyspace does not move the user query
784+
* to the next host.
785+
*
786+
* @test_category connection:connection_pool
787+
*/
788+
@Test(groups = "short")
789+
public void should_not_try_next_host_when_keyspace_setup_receives_invalid_error_response()
790+
throws Exception {
791+
RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE);
792+
ScassandraCluster scassandras = ScassandraCluster.builder().withNodes(2).build();
793+
Cluster cluster = null;
794+
try {
795+
scassandras.init();
796+
scassandras
797+
.node(2)
798+
.primingClient()
799+
.prime(
800+
queryBuilder()
801+
.withQuery("select * from tbl")
802+
.withThen(then().withRows(Collections.singletonMap("result", "result2")))
803+
.build());
804+
cluster =
805+
Cluster.builder()
806+
.addContactPoints(scassandras.address(1).getAddress())
807+
.withPort(scassandras.getBinaryPort())
808+
.withLoadBalancingPolicy(new SortingLoadBalancingPolicy())
809+
.withPoolingOptions(
810+
new PoolingOptions()
811+
.setCoreConnectionsPerHost(HostDistance.LOCAL, 1)
812+
.setMaxConnectionsPerHost(HostDistance.LOCAL, 1)
813+
.setHeartbeatIntervalSeconds(0))
814+
.withRetryPolicy(retryPolicy)
815+
.build();
816+
SessionManager session = (SessionManager) cluster.connect();
817+
Host host1 = TestUtils.findHost(cluster, 1);
818+
Host host2 = TestUtils.findHost(cluster, 2);
819+
HostConnectionPool pool1 = session.pools.get(host1);
820+
HostConnectionPool pool2 = session.pools.get(host2);
821+
Connection connection1 = spy(pool1.connections[0].get(0));
822+
Connection connection2 = spy(pool2.connections[0].get(0));
823+
pool1.connections[0].set(0, connection1);
824+
pool2.connections[0].set(0, connection2);
825+
SettableConnectionFuture useFuture =
826+
new SettableConnectionFuture(new Requests.Query("USE \"missingkeyspace\""));
827+
session.poolsState.setKeyspace("missingkeyspace");
828+
doReturn(useFuture).when(connection1).write(any(Message.Request.class));
829+
doReturn(Futures.immediateFuture(connection2))
830+
.when(connection2)
831+
.setKeyspaceAsync("missingkeyspace");
832+
833+
for (Scassandra node : scassandras.nodes()) {
834+
node.activityClient().clearAllRecordedActivity();
835+
}
836+
837+
ResultSetFuture queryFuture =
838+
session.executeAsync(new SimpleStatement("select * from tbl").setIdempotent(true));
839+
useFuture.setResponse(
840+
errorResponse(ExceptionCode.INVALID, "Keyspace missingkeyspace does not exist"));
841+
842+
try {
843+
queryFuture.getUninterruptibly(5, TimeUnit.SECONDS);
844+
fail("Should have failed the request with the keyspace validation error");
845+
} catch (InvalidQueryException e) {
846+
assertThat(e.getMessage()).contains("Keyspace missingkeyspace does not exist");
847+
}
848+
assertRetryPolicyNotCalled(retryPolicy);
849+
assertThat(connection1.isDefunct()).isFalse();
850+
verify(connection1, times(1)).write(any(Message.Request.class));
851+
verify(connection2, never()).setKeyspaceAsync("missingkeyspace");
852+
assertThat(scassandras.node(1).activityClient().retrieveQueries())
853+
.extractingResultOf("getQuery")
854+
.isEmpty();
855+
assertThat(scassandras.node(2).activityClient().retrieveQueries())
856+
.extractingResultOf("getQuery")
857+
.isEmpty();
858+
} finally {
859+
if (cluster != null) {
860+
cluster.close();
861+
}
862+
scassandras.stop();
863+
}
864+
}
865+
681866
/**
682867
* Ensures that transient server-side keyspace setup failures that happen before the user query is
683868
* written do not consume retry policy attempts.
@@ -714,6 +899,83 @@ public void should_fail_fast_when_keyspace_setup_fails_with_validation_error() t
714899
}
715900
}
716901

902+
/**
903+
* Ensures that server-side timeout errors during keyspace setup that happen before the user query
904+
* is written do not consume retry policy attempts or send the user query.
905+
*
906+
* @test_category connection:connection_pool
907+
*/
908+
@Test(groups = "short")
909+
public void
910+
should_not_call_retry_policy_or_send_query_when_keyspace_setup_fails_with_server_side_timeout()
911+
throws Exception {
912+
RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE);
913+
Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build();
914+
try {
915+
SessionManager session = (SessionManager) cluster.connect();
916+
Host host = TestUtils.findHost(cluster, 1);
917+
HostConnectionPool pool = session.pools.get(host);
918+
Connection connection = spy(pool.connections[0].get(0));
919+
pool.connections[0].set(0, connection);
920+
ReadTimeoutException failure =
921+
new ReadTimeoutException(pool.host.getEndPoint(), ConsistencyLevel.ONE, 0, 1, false);
922+
session.poolsState.setKeyspace("newkeyspace");
923+
doReturn(Futures.<Connection>immediateFailedFuture(failure))
924+
.when(connection)
925+
.setKeyspaceAsync("newkeyspace");
926+
activityClient.clearAllRecordedActivity();
927+
928+
try {
929+
session.execute(new SimpleStatement("select * from tbl").setIdempotent(true));
930+
fail("Should have failed after exhausting query plan");
931+
} catch (NoHostAvailableException e) {
932+
assertThat(e.getErrors().values()).containsOnly(failure);
933+
}
934+
assertRetryPolicyNotCalled(retryPolicy);
935+
assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty();
936+
} finally {
937+
cluster.close();
938+
}
939+
}
940+
941+
/**
942+
* Ensures that client-side timeouts during keyspace setup that happen before the user query is
943+
* written do not consume retry policy attempts or send the user query.
944+
*
945+
* @test_category connection:connection_pool
946+
*/
947+
@Test(groups = "short")
948+
public void
949+
should_not_call_retry_policy_or_send_query_when_keyspace_setup_fails_with_client_side_timeout()
950+
throws Exception {
951+
RetryPolicy retryPolicy = spy(DefaultRetryPolicy.INSTANCE);
952+
Cluster cluster = createClusterBuilder().withRetryPolicy(retryPolicy).build();
953+
try {
954+
SessionManager session = (SessionManager) cluster.connect();
955+
Host host = TestUtils.findHost(cluster, 1);
956+
HostConnectionPool pool = session.pools.get(host);
957+
Connection connection = spy(pool.connections[0].get(0));
958+
pool.connections[0].set(0, connection);
959+
OperationTimedOutException failure = new OperationTimedOutException(pool.host.getEndPoint());
960+
session.poolsState.setKeyspace("newkeyspace");
961+
doReturn(Futures.<Connection>immediateFailedFuture(failure))
962+
.when(connection)
963+
.setKeyspaceAsync("newkeyspace");
964+
activityClient.clearAllRecordedActivity();
965+
966+
try {
967+
session.execute(new SimpleStatement("select * from tbl").setIdempotent(true));
968+
fail("Should have failed after exhausting query plan");
969+
} catch (NoHostAvailableException e) {
970+
assertThat(e.getErrors().values()).containsOnly(failure);
971+
}
972+
assertRetryPolicyNotCalled(retryPolicy);
973+
assertThat(activityClient.retrieveQueries()).extractingResultOf("getQuery").isEmpty();
974+
} finally {
975+
cluster.close();
976+
}
977+
}
978+
717979
/**
718980
* Ensures that internal driver keyspace setup failures that happen before the user query is
719981
* written do not consume retry policy attempts.

0 commit comments

Comments
 (0)