diff --git a/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java b/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java index c17ccd50ff8..65a7ec26af6 100644 --- a/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java +++ b/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java @@ -25,7 +25,6 @@ import com.datastax.dse.driver.internal.core.cql.DseConversions; import com.datastax.dse.protocol.internal.request.Revise; import com.datastax.dse.protocol.internal.response.result.DseRowsMetadata; -import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.DriverTimeoutException.NodeDiagnostics; @@ -366,7 +365,7 @@ private void sendRequest( // We've reached the end of the query plan without finding any node to write to; abort the // continuous paging session. if (activeExecutionsCount.decrementAndGet() == 0) { - abortGlobalRequestOrChosenCallback(AllNodesFailedException.fromErrors(errors)); + abortGlobalRequestOrChosenCallback(DefaultSession.queryPlanExhaustedError(session, errors)); } } else if (!chosenCallback.isDone()) { boolean writeSubmitted = false; diff --git a/core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java b/core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java index cf6875dd362..dcd12fbad28 100644 --- a/core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java +++ b/core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java @@ -25,7 +25,6 @@ import com.datastax.dse.driver.api.core.metrics.DseNodeMetric; import com.datastax.dse.driver.api.core.metrics.DseSessionMetric; import com.datastax.dse.driver.internal.core.graph.binary.GraphBinaryModule; -import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.DriverException; import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.DriverTimeoutException.NodeDiagnostics; @@ -297,7 +296,7 @@ private void sendRequest( // We're the last execution so fail the result setFinalError( statement, - AllNodesFailedException.fromErrors(this.errors), + DefaultSession.queryPlanExhaustedError(session, this.errors), null, NO_SUCCESSFUL_EXECUTION); } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandler.java index b05fec2cb7e..eda5d6bd151 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandler.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandler.java @@ -19,7 +19,6 @@ import static com.datastax.oss.driver.api.core.DriverTimeoutException.UNAVAILABLE; -import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.DriverTimeoutException.NodeDiagnostics; @@ -227,7 +226,7 @@ private void sendRequest(PrepareRequest request, Node node, int retryCount) { } } if (channel == null) { - setFinalError(AllNodesFailedException.fromErrors(this.errors)); + setFinalError(DefaultSession.queryPlanExhaustedError(session, this.errors)); } else { boolean writeSubmitted = false; try { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java index ce4b40d6d29..d7e24c262ab 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java @@ -25,7 +25,6 @@ import static com.datastax.oss.driver.api.core.DriverTimeoutException.UNAVAILABLE; -import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DriverException; import com.datastax.oss.driver.api.core.DriverTimeoutException; @@ -406,7 +405,8 @@ private void sendRequest( // We've reached the end of the query plan without finding any node to write to if (!result.isDone() && activeExecutionsCount.decrementAndGet() == 0) { // We're the last execution so fail the result - setFinalError(statement, AllNodesFailedException.fromErrors(this.errors), null, -1); + setFinalError( + statement, DefaultSession.queryPlanExhaustedError(session, this.errors), null, -1); } } else { boolean writeSubmitted = false; diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java b/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java index 7484f31d7d3..c5f2f9a04dd 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java @@ -23,6 +23,7 @@ */ package com.datastax.oss.driver.internal.core.session; +import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.AsyncAutoCloseable; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.CqlSession; @@ -99,6 +100,19 @@ public static CompletionStage init( return new DefaultSession(context, contactPoints).init(keyspace); } + /** + * Returns {@code IllegalStateException("Session is closed")} for an empty plan during shutdown + * (scylladb/java-driver#846); otherwise delegates to {@link AllNodesFailedException#fromErrors}. + */ + @NonNull + public static Throwable queryPlanExhaustedError( + @NonNull DefaultSession session, @Nullable List> errors) { + if ((errors == null || errors.isEmpty()) && (session.isClosing() || session.isClosed())) { + return new IllegalStateException("Session is closed"); + } + return AllNodesFailedException.fromErrors(errors); + } + private final InternalDriverContext context; private final EventExecutor adminExecutor; private final String logPrefix; @@ -108,6 +122,10 @@ public static CompletionStage init( private final PoolManager poolManager; private final SessionMetricUpdater metricUpdater; + // Flipped before scheduling close on adminExecutor so handlers see shutdown before closeFuture + // completes. + private volatile boolean closing; + private DefaultSession(InternalDriverContext context, Set contactPoints) { int instanceCount = INSTANCE_COUNT.incrementAndGet(); int threshold = @@ -296,15 +314,22 @@ public CompletionStage closeFuture() { @NonNull @Override public CompletionStage closeAsync() { + closing = true; return closeSafely(singleThreaded::close); } @NonNull @Override public CompletionStage forceCloseAsync() { + closing = true; return closeSafely(singleThreaded::forceClose); } + /** Whether close has been initiated; flips before {@link #isClosed()} does. */ + public boolean isClosing() { + return closing; + } + private CompletionStage closeSafely(Runnable action) { // Protect against getting closed twice: with the default NettyOptions, closing shuts down // adminExecutor, so we don't want to call RunOrSchedule the second time. diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerTest.java index f9068d137f2..a6280dcdfcc 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandlerTest.java @@ -119,6 +119,55 @@ public void should_fail_if_no_node_available() { } } + // scylladb/java-driver#846: post-shutdown branch (closeFuture already completed). + @Test + public void should_fail_with_session_closed_when_query_plan_empty_and_session_is_closed() { + // no withResponse/withEmptyPool calls => empty query plan + try (RequestHandlerTestHarness harness = RequestHandlerTestHarness.builder().build()) { + when(harness.getSession().isClosed()).thenReturn(true); + + CompletionStage resultSetFuture = + new CqlRequestHandler( + UNDEFINED_IDEMPOTENCE_STATEMENT, + harness.getSession(), + harness.getContext(), + "test") + .handle(); + + assertThatStage(resultSetFuture) + .isFailed( + error -> { + assertThat(error).isInstanceOf(IllegalStateException.class); + assertThat(error).hasMessage("Session is closed"); + }); + } + } + + // scylladb/java-driver#846: shutdown-in-progress branch (closeAsync called, closeFuture not yet + // done). + @Test + public void should_fail_with_session_closed_when_query_plan_empty_and_session_is_closing() { + // no withResponse/withEmptyPool calls => empty query plan + try (RequestHandlerTestHarness harness = RequestHandlerTestHarness.builder().build()) { + when(harness.getSession().isClosing()).thenReturn(true); + + CompletionStage resultSetFuture = + new CqlRequestHandler( + UNDEFINED_IDEMPOTENCE_STATEMENT, + harness.getSession(), + harness.getContext(), + "test") + .handle(); + + assertThatStage(resultSetFuture) + .isFailed( + error -> { + assertThat(error).isInstanceOf(IllegalStateException.class); + assertThat(error).hasMessage("Session is closed"); + }); + } + } + @Test public void should_fail_if_nodes_unavailable() { RequestHandlerTestHarness.Builder harnessBuilder = RequestHandlerTestHarness.builder();