diff --git a/changelog/README.md b/changelog/README.md index b90979e2e71..6a16ee93e42 100644 --- a/changelog/README.md +++ b/changelog/README.md @@ -23,6 +23,7 @@ under the License. ### 4.19.2 +- [improvement] #958: Allow retry policies to handle client-side request timeouts - [bug] CASSJAVA-116: Retry or Speculative Execution with RequestIdGenerator throws "Duplicate Key" ### 4.19.1 diff --git a/core/revapi.json b/core/revapi.json index 6e9d7234d8e..458e1d03247 100644 --- a/core/revapi.json +++ b/core/revapi.json @@ -14,6 +14,11 @@ } }, "ignore": [ + { + "code": "java.method.addedToInterface", + "new": "method com.datastax.oss.driver.api.core.retry.RetryVerdict com.datastax.oss.driver.api.core.retry.RetryPolicy::onRequestTimeoutVerdict(com.datastax.oss.driver.api.core.session.Request, com.datastax.oss.driver.api.core.ConsistencyLevel, com.datastax.oss.driver.api.core.DriverTimeoutException, boolean, int)", + "justification": "#958: Expose a dedicated callback for client-side request timeouts" + }, { "code": "java.method.removed", "old": "method com.datastax.oss.driver.api.core.cql.BatchStatementBuilder com.datastax.oss.driver.api.core.cql.BatchStatementBuilder::withKeyspace(com.datastax.oss.driver.api.core.CqlIdentifier)", diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/retry/RetryPolicy.java b/core/src/main/java/com/datastax/oss/driver/api/core/retry/RetryPolicy.java index e8546816e23..44b69ab9e57 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/retry/RetryPolicy.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/retry/RetryPolicy.java @@ -18,6 +18,7 @@ package com.datastax.oss.driver.api.core.retry; import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.connection.ClosedConnectionException; import com.datastax.oss.driver.api.core.connection.HeartbeatException; import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy; @@ -216,6 +217,39 @@ default RetryVerdict onUnavailableVerdict( return () -> decision; } + /** + * Whether to retry when the driver did not receive a response before the client-side request + * timeout expired. + * + *

This is different from {@link #onReadTimeoutVerdict(Request, ConsistencyLevel, int, int, + * boolean, int)} and {@link #onWriteTimeoutVerdict(Request, ConsistencyLevel, WriteType, int, + * int, int)}: those methods handle server-side timeout responses returned by a coordinator, while + * this method handles a timeout raised by the driver itself. + * + *

Since a client-side timeout does not prove that the coordinator failed to execute the + * request, policies should only retry when {@code requestIdempotent} is {@code true}, unless + * duplicate execution is explicitly acceptable for the request. + * + *

If this method returns a retry verdict, the driver starts a new attempt and schedules the + * client-side request timeout again for the retried request. + * + * @param request the request that timed out. + * @param cl the requested consistency level. + * @param error the timeout raised by the driver. + * @param requestIdempotent the request idempotence, resolved from {@link Request#isIdempotent()} + * and the configured default. + * @param retryCount how many times the retry policy has been invoked already for this request + * (not counting the current invocation). + */ + default RetryVerdict onRequestTimeoutVerdict( + @NonNull Request request, + @NonNull ConsistencyLevel cl, + @NonNull DriverTimeoutException error, + boolean requestIdempotent, + int retryCount) { + return RetryVerdict.RETHROW; + } + /** * Whether to retry when a request was aborted before we could get a response from the server. * 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 29f595e5225..62cb075ccf6 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 @@ -26,6 +26,7 @@ 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.ConsistencyLevel; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DriverException; import com.datastax.oss.driver.api.core.DriverTimeoutException; @@ -46,6 +47,7 @@ import com.datastax.oss.driver.api.core.metadata.token.Token; import com.datastax.oss.driver.api.core.metrics.DefaultNodeMetric; import com.datastax.oss.driver.api.core.metrics.DefaultSessionMetric; +import com.datastax.oss.driver.api.core.retry.RetryDecision; import com.datastax.oss.driver.api.core.retry.RetryPolicy; import com.datastax.oss.driver.api.core.retry.RetryVerdict; import com.datastax.oss.driver.api.core.servererrors.BootstrappingException; @@ -111,6 +113,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import net.jcip.annotations.GuardedBy; import net.jcip.annotations.ThreadSafe; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -142,15 +145,32 @@ public class CqlRequestHandler implements Throttled { */ private final AtomicInteger startedSpeculativeExecutionsCount; - final Timeout scheduledTimeout; + volatile Timeout scheduledTimeout; final List scheduledExecutions; private final List inFlightCallbacks; + private final Object callbackLock = new Object(); private final RequestThrottler throttler; private final RequestTracker requestTracker; private final Optional requestIdGenerator; private final SessionMetricUpdater sessionMetricUpdater; private final DriverExecutionProfile executionProfile; + @GuardedBy("callbackLock") + @Nullable + private Statement preDispatchStatement; + + @GuardedBy("callbackLock") + private int preDispatchRetryCount; + + @GuardedBy("callbackLock") + private boolean started; + + @GuardedBy("callbackLock") + private int requestGeneration; + + @GuardedBy("callbackLock") + private int retainedExecution; + // The errors on the nodes that were already tried (lazily initialized on the first error). // We don't use a map because nodes can appear multiple times. private volatile List> errors; @@ -203,8 +223,8 @@ protected CqlRequestHandler( this.timer = context.getNettyOptions().getTimer(); this.executionProfile = Conversions.resolveExecutionProfile(initialStatement, context); - Duration timeout = Conversions.resolveRequestTimeout(statement, executionProfile); - this.scheduledTimeout = scheduleTimeout(timeout); + this.preDispatchStatement = initialStatement; + this.scheduledTimeout = scheduleTimeout(statement); this.throttler = context.getRequestThrottler(); this.throttler.register(this); @@ -222,35 +242,34 @@ public void onThrottleReady(boolean wasDelayed) { System.nanoTime() - startTimeNanos, TimeUnit.NANOSECONDS); } - Queue queryPlan; - if (this.initialStatement.getNode() != null) { - queryPlan = new SimpleQueryPlan(this.initialStatement.getNode()); - } else { - queryPlan = - context - .getLoadBalancingPolicyWrapper() - .newQueryPlan(initialStatement, executionProfile.getName(), session); + synchronized (callbackLock) { + if (result.isDone() || preDispatchStatement == null) { + return; + } + started = true; + Statement statement = preDispatchStatement; + int retryCount = preDispatchRetryCount; + preDispatchStatement = null; + sendRequest(statement, null, newQueryPlan(statement), 0, retryCount, true); } - - sendRequest(initialStatement, null, queryPlan, 0, 0, true); } public CompletionStage handle() { return result; } - private Timeout scheduleTimeout(Duration timeoutDuration) { + private Timeout scheduleTimeout(Statement statement) { + Duration timeoutDuration = Conversions.resolveRequestTimeout(statement, executionProfile); if (timeoutDuration.toNanos() > 0) { try { return this.timer.newTimeout( (Timeout timeout1) -> { - NodeDiagnostics diagnostics = buildNodeDiagnostics(); - setFinalError( - initialStatement, + if (result.isDone()) { + return; + } + processRequestTimeout( new DriverTimeoutException( - "Query timed out after " + timeoutDuration, diagnostics), - null, - -1); + "Query timed out after " + timeoutDuration, buildNodeDiagnostics())); }, timeoutDuration.toNanos(), TimeUnit.NANOSECONDS); @@ -266,13 +285,231 @@ private Timeout scheduleTimeout(Duration timeoutDuration) { return null; } + private void processRequestTimeout(DriverTimeoutException error) { + NodeResponseCallback callback; + synchronized (callbackLock) { + if (result.isDone()) { + return; + } + callback = inFlightCallbacks.isEmpty() ? null : inFlightCallbacks.get(0); + if (callback == null) { + processRequestTimeoutWithoutCallback(error); + return; + } + } + + RetryVerdict verdict = + invokeRequestTimeoutRetryPolicy( + callback.statement, error, callback.retryCount, callback.node, callback.execution); + if (verdict == null) { + return; + } + + if (isRetryOrIgnore(verdict)) { + incrementClientTimeoutMetric(); + } + + cancelTimedOutCallbacks(callback); + callback.processRetryVerdict(verdict, error, true); + } + + private void processRequestTimeoutWithoutCallback(DriverTimeoutException error) { + synchronized (callbackLock) { + Statement statement = preDispatchStatement; + if (statement == null) { + setFinalError(initialStatement, error, null, -1); + return; + } + int retryCount = preDispatchRetryCount; + RetryVerdict verdict = + invokeRequestTimeoutRetryPolicy(statement, error, retryCount, null, -1); + if (verdict == null) { + return; + } + + if (isRetryOrIgnore(verdict)) { + incrementClientTimeoutMetric(); + } + + switch (verdict.getRetryDecision()) { + case RETRY_SAME: + case RETRY_NEXT: + Statement retryRequest = verdict.getRetryRequest(statement); + cancelTimedOutCallbacks(null); + preDispatchRetryCount = retryCount + 1; + preDispatchStatement = started ? null : retryRequest; + scheduledTimeout = scheduleTimeout(retryRequest); + if (started) { + sendRequest(retryRequest, null, newQueryPlan(retryRequest), 0, retryCount + 1, true); + } + break; + case RETHROW: + preDispatchStatement = null; + setFinalError(statement, error, null, -1); + break; + case IGNORE: + cancelTimedOutCallbacks(null); + preDispatchStatement = null; + setFinalResult(Void.INSTANCE, null, true, statement, null, -1); + break; + } + } + } + @Nullable - private NodeDiagnostics buildNodeDiagnostics() { - List callbacks = inFlightCallbacks; - if (callbacks.isEmpty()) { + private RetryVerdict invokeRequestTimeoutRetryPolicy( + Statement statement, + DriverTimeoutException error, + int retryCount, + @Nullable Node node, + int execution) { + RetryVerdict verdict; + try { + RetryPolicy retryPolicy = Conversions.resolveRetryPolicy(context, executionProfile); + verdict = + retryPolicy.onRequestTimeoutVerdict( + statement, + resolveConsistency(statement), + error, + Conversions.resolveIdempotence(statement, executionProfile), + retryCount); + } catch (Throwable cause) { + setFinalError( + statement, + new IllegalStateException("Unexpected error while invoking the retry policy", cause), + node, + execution); return null; } - NodeResponseCallback cb = callbacks.get(0); + + if (verdict == null) { + setFinalError( + statement, + new NullPointerException("Retry policy returned null verdict"), + node, + execution); + return null; + } + return verdict; + } + + private boolean isRetryOrIgnore(RetryVerdict verdict) { + return verdict.getRetryDecision() == RetryDecision.RETRY_SAME + || verdict.getRetryDecision() == RetryDecision.RETRY_NEXT + || verdict.getRetryDecision() == RetryDecision.IGNORE; + } + + private void incrementClientTimeoutMetric() { + sessionMetricUpdater.incrementCounter( + DefaultSessionMetric.CQL_CLIENT_TIMEOUTS, executionProfile.getName()); + } + + private void cancelTimedOutCallbacks(@Nullable NodeResponseCallback retryCallback) { + synchronized (callbackLock) { + requestGeneration++; + retainedExecution = retryCallback == null ? 0 : retryCallback.execution; + + for (Timeout scheduledExecution : scheduledExecutions) { + scheduledExecution.cancel(); + } + scheduledExecutions.clear(); + + for (NodeResponseCallback callback : inFlightCallbacks.toArray(new NodeResponseCallback[0])) { + inFlightCallbacks.remove(callback); + callback.cancel(); + if (callback != retryCallback) { + activeExecutionsCount.decrementAndGet(); + } + } + } + } + + private Queue newQueryPlan(Statement statement) { + if (statement.getNode() != null) { + return new SimpleQueryPlan(statement.getNode()); + } else { + return context + .getLoadBalancingPolicyWrapper() + .newQueryPlan(statement, executionProfile.getName(), session); + } + } + + private int currentRequestGeneration() { + synchronized (callbackLock) { + return requestGeneration; + } + } + + private boolean registerInFlightCallback(NodeResponseCallback callback) { + synchronized (callbackLock) { + if (result.isDone() || callback.generation != requestGeneration) { + return false; + } + inFlightCallbacks.add(callback); + return true; + } + } + + private boolean removeInFlightCallback(NodeResponseCallback callback) { + synchronized (callbackLock) { + return inFlightCallbacks.remove(callback); + } + } + + private boolean isStale(NodeResponseCallback callback) { + synchronized (callbackLock) { + return callback.generation != requestGeneration; + } + } + + private boolean isStaleSpeculativeExecution(int generation) { + synchronized (callbackLock) { + return generation != requestGeneration; + } + } + + private boolean isStaleGeneration(int generation) { + synchronized (callbackLock) { + return generation != requestGeneration; + } + } + + private void decrementActiveExecutionsIfStale(int generation, int execution) { + synchronized (callbackLock) { + if (generation != requestGeneration && execution != retainedExecution) { + activeExecutionsCount.decrementAndGet(); + } + } + } + + private void decrementActiveExecutionsIfStale(NodeResponseCallback callback, boolean removed) { + synchronized (callbackLock) { + if (removed + && callback.generation != requestGeneration + && callback.execution != retainedExecution) { + activeExecutionsCount.decrementAndGet(); + } + } + } + + private ConsistencyLevel resolveConsistency(Statement statement) { + ConsistencyLevel consistency = statement.getConsistencyLevel(); + return consistency != null + ? consistency + : context + .getConsistencyLevelRegistry() + .nameToLevel(executionProfile.getString(DefaultDriverOption.REQUEST_CONSISTENCY)); + } + + @Nullable + private NodeDiagnostics buildNodeDiagnostics() { + NodeResponseCallback cb; + synchronized (callbackLock) { + if (inFlightCallbacks.isEmpty()) { + return null; + } + cb = inFlightCallbacks.get(0); + } int channelInFlight = cb.channel.getInFlight(); ChannelPool pool = session.getPools().get(cb.node); return NodeDiagnostics.of( @@ -375,9 +612,31 @@ private void sendRequest( int currentExecutionIndex, int retryCount, boolean scheduleNextExecution) { + sendRequest( + statement, + retriedNode, + queryPlan, + currentExecutionIndex, + retryCount, + scheduleNextExecution, + currentRequestGeneration()); + } + + private void sendRequest( + Statement statement, + Node retriedNode, + Queue queryPlan, + int currentExecutionIndex, + int retryCount, + boolean scheduleNextExecution, + int generation) { if (result.isDone()) { return; } + if (isStaleGeneration(generation)) { + decrementActiveExecutionsIfStale(generation, currentExecutionIndex); + return; + } Node node = retriedNode; DriverChannel channel = null; if (node == null @@ -428,8 +687,17 @@ private void sendRequest( currentExecutionIndex, retryCount, scheduleNextExecution, + generation, logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex)); Message message = Conversions.toMessage(statement, executionProfile, context); + if (!registerInFlightCallback(nodeResponseCallback)) { + decrementActiveExecutionsIfStale(generation, currentExecutionIndex); + channel + .write( + message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback) + .addListener(nodeResponseCallback); + return; + } channel .write(message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback) .addListener(nodeResponseCallback); @@ -469,9 +737,45 @@ private void setFinalResult( Frame responseFrame, boolean schemaInAgreement, NodeResponseCallback callback) { + setFinalResult( + resultMessage, + responseFrame, + schemaInAgreement, + callback.statement, + callback.node, + callback.execution, + callback.nodeStartTimeNanos); + } + + private void setFinalResult( + Result resultMessage, + Frame responseFrame, + boolean schemaInAgreement, + Statement statement, + @Nullable Node node, + int execution) { + setFinalResult( + resultMessage, + responseFrame, + schemaInAgreement, + statement, + node, + execution, + NANOTIME_NOT_MEASURED_YET); + } + + private void setFinalResult( + Result resultMessage, + Frame responseFrame, + boolean schemaInAgreement, + Statement statement, + @Nullable Node node, + int execution, + long nodeStartTimeNanos) { try { ExecutionInfo executionInfo = - buildExecutionInfo(callback, resultMessage, responseFrame, schemaInAgreement); + buildExecutionInfo( + statement, node, execution, resultMessage, responseFrame, schemaInAgreement); AsyncResultSet resultSet = Conversions.toResultSet(resultMessage, executionInfo, session, context); if (result.complete(resultSet)) { @@ -485,19 +789,18 @@ private void setFinalResult( if (!(requestTracker instanceof NoopRequestTracker)) { completionTimeNanos = System.nanoTime(); totalLatencyNanos = completionTimeNanos - startTimeNanos; - long nodeLatencyNanos = completionTimeNanos - callback.nodeStartTimeNanos; - requestTracker.onNodeSuccess( - callback.statement, - nodeLatencyNanos, - executionProfile, - callback.node, - handlerLogPrefix); - requestTracker.onSuccess( - callback.statement, - totalLatencyNanos, - executionProfile, - callback.node, - handlerLogPrefix); + if (node != null) { + long nodeLatencyNanos = + nodeStartTimeNanos == NANOTIME_NOT_MEASURED_YET + ? totalLatencyNanos + : completionTimeNanos - nodeStartTimeNanos; + requestTracker.onNodeSuccess( + statement, nodeLatencyNanos, executionProfile, node, handlerLogPrefix); + } + if (node != null) { + requestTracker.onSuccess( + statement, totalLatencyNanos, executionProfile, node, handlerLogPrefix); + } } if (sessionMetricUpdater.isEnabled( DefaultSessionMetric.CQL_REQUESTS, executionProfile.getName())) { @@ -538,10 +841,10 @@ private void setFinalResult( if (!executionInfo.getWarnings().isEmpty() && executionProfile.getBoolean(DefaultDriverOption.REQUEST_LOG_WARNINGS) && LOG.isWarnEnabled()) { - logServerWarnings(callback.statement, executionProfile, executionInfo.getWarnings()); + logServerWarnings(statement, executionProfile, executionInfo.getWarnings()); } } catch (Throwable error) { - setFinalError(callback.statement, error, callback.node, -1); + setFinalError(statement, error, node, execution); } } @@ -573,17 +876,19 @@ private void logServerWarnings( } private ExecutionInfo buildExecutionInfo( - NodeResponseCallback callback, + Statement statement, + @Nullable Node node, + int execution, Result resultMessage, Frame responseFrame, boolean schemaInAgreement) { ByteBuffer pagingState = (resultMessage instanceof Rows) ? ((Rows) resultMessage).getMetadata().pagingState : null; return new DefaultExecutionInfo( - callback.statement, - callback.node, + statement, + node, startedSpeculativeExecutionsCount.get(), - callback.execution, + execution, errors, pagingState, responseFrame, @@ -654,6 +959,7 @@ private class NodeResponseCallback // the first attempt of each execution). private final int retryCount; private final boolean scheduleNextExecution; + private final int generation; private final String logPrefix; private NodeResponseCallback( @@ -664,6 +970,7 @@ private NodeResponseCallback( int execution, int retryCount, boolean scheduleNextExecution, + int generation, String logPrefix) { this.statement = statement; this.node = node; @@ -672,13 +979,26 @@ private NodeResponseCallback( this.execution = execution; this.retryCount = retryCount; this.scheduleNextExecution = scheduleNextExecution; + this.generation = generation; this.logPrefix = logPrefix; } // this gets invoked once the write completes. @Override public void operationComplete(Future future) throws Exception { + if (result.isDone()) { + removeInFlightCallback(this); + cancel(); + return; + } + if (isStale(this)) { + boolean removed = removeInFlightCallback(this); + cancel(); + decrementActiveExecutionsIfStale(this, removed); + return; + } if (!future.isSuccess()) { + removeInFlightCallback(this); Throwable error = future.cause(); if (error instanceof EncoderException && error.getCause() instanceof FrameTooLongException) { @@ -701,41 +1021,34 @@ public void operationComplete(Future future) throws Exception { queryPlan, execution, retryCount, - scheduleNextExecution); // try next node + scheduleNextExecution, + generation); // try next node } } else { LOG.trace("[{}] Request sent on {}", logPrefix, channel); - if (result.isDone()) { - // If the handler completed since the last time we checked, cancel directly because we - // don't know if cancelScheduledTasks() has run yet - cancel(); - } else { - inFlightCallbacks.add(this); - if (scheduleNextExecution - && Conversions.resolveIdempotence(statement, executionProfile)) { - int nextExecution = execution + 1; - long nextDelay; - try { - nextDelay = - Conversions.resolveSpeculativeExecutionPolicy(context, executionProfile) - .nextExecution(node, keyspace, statement, nextExecution); - } catch (Throwable cause) { - // This is a bug in the policy, but not fatal since we have at least one other - // execution already running. Don't fail the whole request. - LOG.error( - "[{}] Unexpected error while invoking the speculative execution policy", - logPrefix, - cause); - return; - } - if (nextDelay >= 0) { - scheduleSpeculativeExecution(nextExecution, nextDelay); - } else { - LOG.trace( - "[{}] Speculative execution policy returned {}, no next execution", - logPrefix, - nextDelay); - } + if (scheduleNextExecution && Conversions.resolveIdempotence(statement, executionProfile)) { + int nextExecution = execution + 1; + long nextDelay; + try { + nextDelay = + Conversions.resolveSpeculativeExecutionPolicy(context, executionProfile) + .nextExecution(node, keyspace, statement, nextExecution); + } catch (Throwable cause) { + // This is a bug in the policy, but not fatal since we have at least one other + // execution already running. Don't fail the whole request. + LOG.error( + "[{}] Unexpected error while invoking the speculative execution policy", + logPrefix, + cause); + return; + } + if (nextDelay >= 0) { + scheduleSpeculativeExecution(nextExecution, nextDelay); + } else { + LOG.trace( + "[{}] Speculative execution policy returned {}, no next execution", + logPrefix, + nextDelay); } } } @@ -747,7 +1060,7 @@ private void scheduleSpeculativeExecution(int index, long delay) { scheduledExecutions.add( timer.newTimeout( (Timeout timeout1) -> { - if (!result.isDone()) { + if (!result.isDone() && !isStaleSpeculativeExecution(generation)) { LOG.trace( "[{}] Starting speculative execution {}", CqlRequestHandler.this.handlerLogPrefix, @@ -760,7 +1073,7 @@ private void scheduleSpeculativeExecution(int index, long delay) { .getMetricUpdater() .incrementCounter( DefaultNodeMetric.SPECULATIVE_EXECUTIONS, executionProfile.getName()); - sendRequest(statement, null, queryPlan, index, 0, true); + sendRequest(statement, null, queryPlan, index, 0, true, generation); } }, delay, @@ -777,6 +1090,14 @@ private void scheduleSpeculativeExecution(int index, long delay) { @Override public void onResponse(Frame responseFrame) { + boolean removed = removeInFlightCallback(this); + if (result.isDone()) { + return; + } + if (isStale(this)) { + decrementActiveExecutionsIfStale(this, removed); + return; + } long nodeResponseTimeNanos = NANOTIME_NOT_MEASURED_YET; NodeMetricUpdater nodeMetricUpdater = ((DefaultNode) node).getMetricUpdater(); if (nodeMetricUpdater.isEnabled(DefaultNodeMetric.CQL_MESSAGES, executionProfile.getName())) { @@ -788,10 +1109,6 @@ public void onResponse(Frame responseFrame) { nodeLatency, TimeUnit.NANOSECONDS); } - inFlightCallbacks.remove(this); - if (result.isDone()) { - return; - } try { Message responseMessage = responseFrame.message; if (responseMessage instanceof SchemaChange) { @@ -873,6 +1190,9 @@ private void processErrorResponse(Error errorMessage) { .start() .handle( (repreparedId, exception) -> { + if (result.isDone() || isStale(this)) { + return null; + } if (exception != null) { // If the error is not recoverable, surface it to the client instead of retrying if (exception instanceof UnexpectedResponseException) { @@ -898,7 +1218,8 @@ private void processErrorResponse(Error errorMessage) { recordError(node, exception); trackNodeError(node, exception, NANOTIME_NOT_MEASURED_YET); LOG.trace("[{}] Reprepare failed, trying next node", logPrefix); - sendRequest(statement, null, queryPlan, execution, retryCount, false); + sendRequest( + statement, null, queryPlan, execution, retryCount, false, generation); } else { if (!repreparedId.equals(idToReprepare)) { IllegalStateException illegalStateException = @@ -914,7 +1235,8 @@ private void processErrorResponse(Error errorMessage) { setFinalError(statement, illegalStateException, node, execution); } LOG.trace("[{}] Reprepare sucessful, retrying", logPrefix); - sendRequest(statement, node, queryPlan, execution, retryCount, false); + sendRequest( + statement, node, queryPlan, execution, retryCount, false, generation); } return null; }); @@ -926,7 +1248,7 @@ private void processErrorResponse(Error errorMessage) { LOG.trace("[{}] {} is bootstrapping, trying next node", logPrefix, node); recordError(node, error); trackNodeError(node, error, NANOTIME_NOT_MEASURED_YET); - sendRequest(statement, null, queryPlan, execution, retryCount, false); + sendRequest(statement, null, queryPlan, execution, retryCount, false, generation); } else if (error instanceof QueryValidationException || error instanceof FunctionFailureException || error instanceof ProtocolError) { @@ -1003,29 +1325,33 @@ private void processErrorResponse(Error errorMessage) { } private void processRetryVerdict(RetryVerdict verdict, Throwable error) { + processRetryVerdict(verdict, error, false); + } + + private void processRetryVerdict( + RetryVerdict verdict, Throwable error, boolean restartRequestTimeout) { LOG.trace("[{}] Processing retry decision {}", logPrefix, verdict); + int retryGeneration = restartRequestTimeout ? currentRequestGeneration() : generation; switch (verdict.getRetryDecision()) { case RETRY_SAME: recordError(node, error); trackNodeError(node, error, NANOTIME_NOT_MEASURED_YET); + Statement retrySameRequest = verdict.getRetryRequest(statement); + if (restartRequestTimeout) { + scheduledTimeout = scheduleTimeout(retrySameRequest); + } sendRequest( - verdict.getRetryRequest(statement), - node, - queryPlan, - execution, - retryCount + 1, - false); + retrySameRequest, node, queryPlan, execution, retryCount + 1, false, retryGeneration); break; case RETRY_NEXT: recordError(node, error); trackNodeError(node, error, NANOTIME_NOT_MEASURED_YET); + Statement retryNextRequest = verdict.getRetryRequest(statement); + if (restartRequestTimeout) { + scheduledTimeout = scheduleTimeout(retryNextRequest); + } sendRequest( - verdict.getRetryRequest(statement), - null, - queryPlan, - execution, - retryCount + 1, - false); + retryNextRequest, null, queryPlan, execution, retryCount + 1, false, retryGeneration); break; case RETHROW: trackNodeError(node, error, NANOTIME_NOT_MEASURED_YET); @@ -1061,10 +1387,14 @@ private void updateErrorMetrics( @Override public void onFailure(Throwable error) { - inFlightCallbacks.remove(this); + boolean removed = removeInFlightCallback(this); if (result.isDone()) { return; } + if (isStale(this)) { + decrementActiveExecutionsIfStale(this, removed); + return; + } LOG.trace("[{}] Request failure, processing: {}", logPrefix, error); RetryVerdict verdict; if (!Conversions.resolveIdempotence(statement, executionProfile) diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/retry/ConsistencyDowngradingRetryPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/retry/ConsistencyDowngradingRetryPolicy.java index dbf534459a3..14ba7ca004d 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/retry/ConsistencyDowngradingRetryPolicy.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/retry/ConsistencyDowngradingRetryPolicy.java @@ -23,6 +23,7 @@ import static com.datastax.oss.driver.api.core.servererrors.WriteType.UNLOGGED_BATCH; import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.connection.ClosedConnectionException; import com.datastax.oss.driver.api.core.connection.HeartbeatException; import com.datastax.oss.driver.api.core.context.DriverContext; @@ -116,6 +117,10 @@ public class ConsistencyDowngradingRetryPolicy implements RetryPolicy { "[{}] Verdict on unavailable exception (consistency: {}, " + "required replica: {}, alive replica: {}, retries: {}): {}"; + @VisibleForTesting + public static final String VERDICT_ON_REQUEST_TIMEOUT = + "[{}] Verdict on request timeout (consistency: {}, idempotent: {}, retries: {}): {}"; + @VisibleForTesting public static final String VERDICT_ON_ABORTED = "[{}] Verdict on aborted request (type: {}, message: '{}', retries: {}): {}"; @@ -269,6 +274,27 @@ public RetryVerdict onUnavailableVerdict( return verdict; } + /** + * {@inheritDoc} + * + *

This implementation retries client-side request timeouts on the next host for idempotent + * requests only. + */ + @Override + public RetryVerdict onRequestTimeoutVerdict( + @NonNull Request request, + @NonNull ConsistencyLevel cl, + @NonNull DriverTimeoutException error, + boolean requestIdempotent, + int retryCount) { + RetryVerdict verdict = + requestIdempotent && retryCount == 0 ? RetryVerdict.RETRY_NEXT : RetryVerdict.RETHROW; + if (LOG.isTraceEnabled()) { + LOG.trace(VERDICT_ON_REQUEST_TIMEOUT, logPrefix, cl, requestIdempotent, retryCount, verdict); + } + return verdict; + } + @Override public RetryVerdict onRequestAbortedVerdict( @NonNull Request request, @NonNull Throwable error, int retryCount) { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/retry/DefaultRetryPolicy.java b/core/src/main/java/com/datastax/oss/driver/internal/core/retry/DefaultRetryPolicy.java index 8cea1a564b5..7c0a1aa7aba 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/retry/DefaultRetryPolicy.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/retry/DefaultRetryPolicy.java @@ -18,11 +18,13 @@ package com.datastax.oss.driver.internal.core.retry; import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.connection.ClosedConnectionException; import com.datastax.oss.driver.api.core.connection.HeartbeatException; import com.datastax.oss.driver.api.core.context.DriverContext; import com.datastax.oss.driver.api.core.retry.RetryDecision; import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.retry.RetryVerdict; import com.datastax.oss.driver.api.core.servererrors.CoordinatorException; import com.datastax.oss.driver.api.core.servererrors.DefaultWriteType; import com.datastax.oss.driver.api.core.servererrors.ReadFailureException; @@ -186,6 +188,22 @@ public RetryDecision onUnavailable( return decision; } + /** + * {@inheritDoc} + * + *

This implementation always rethrows the timeout. Client-side request timeouts do not prove + * that the coordinator failed to execute the request. + */ + @Override + public RetryVerdict onRequestTimeoutVerdict( + @NonNull Request request, + @NonNull ConsistencyLevel cl, + @NonNull DriverTimeoutException error, + boolean requestIdempotent, + int retryCount) { + return RetryVerdict.RETHROW; + } + /** * {@inheritDoc} * diff --git a/core/src/test/java/com/datastax/oss/driver/api/core/retry/ConsistencyDowngradingRetryPolicyTest.java b/core/src/test/java/com/datastax/oss/driver/api/core/retry/ConsistencyDowngradingRetryPolicyTest.java index e4463d833bf..b6e8ca52c65 100644 --- a/core/src/test/java/com/datastax/oss/driver/api/core/retry/ConsistencyDowngradingRetryPolicyTest.java +++ b/core/src/test/java/com/datastax/oss/driver/api/core/retry/ConsistencyDowngradingRetryPolicyTest.java @@ -121,6 +121,13 @@ public void should_process_unavailable() { assertOnUnavailable(QUORUM, 2, 0, 0).hasDecision(RETHROW); } + @Test + public void should_process_request_timeout() { + assertOnRequestTimeout(QUORUM, true, 0).hasDecision(RETRY_NEXT); + assertOnRequestTimeout(QUORUM, true, 1).hasDecision(RETHROW); + assertOnRequestTimeout(QUORUM, false, 0).hasDecision(RETHROW); + } + @Test public void should_process_aborted_request() { assertOnRequestAborted(ClosedConnectionException.class, 0).hasDecision(RETRY_NEXT); diff --git a/core/src/test/java/com/datastax/oss/driver/api/core/retry/DefaultRetryPolicyTest.java b/core/src/test/java/com/datastax/oss/driver/api/core/retry/DefaultRetryPolicyTest.java index e36ccff2b91..b92f1f9454c 100644 --- a/core/src/test/java/com/datastax/oss/driver/api/core/retry/DefaultRetryPolicyTest.java +++ b/core/src/test/java/com/datastax/oss/driver/api/core/retry/DefaultRetryPolicyTest.java @@ -62,6 +62,13 @@ public void should_process_unavailable() { assertOnUnavailable(QUORUM, 2, 1, 1).hasDecision(RETHROW); } + @Test + public void should_process_request_timeout() { + assertOnRequestTimeout(QUORUM, true, 0).hasDecision(RETHROW); + assertOnRequestTimeout(QUORUM, true, 1).hasDecision(RETHROW); + assertOnRequestTimeout(QUORUM, false, 0).hasDecision(RETHROW); + } + @Test public void should_process_aborted_request() { assertOnRequestAborted(ClosedConnectionException.class, 0).hasDecision(RETRY_NEXT); diff --git a/core/src/test/java/com/datastax/oss/driver/api/core/retry/RetryPolicyTestBase.java b/core/src/test/java/com/datastax/oss/driver/api/core/retry/RetryPolicyTestBase.java index a57f4ab352f..1490a014da0 100644 --- a/core/src/test/java/com/datastax/oss/driver/api/core/retry/RetryPolicyTestBase.java +++ b/core/src/test/java/com/datastax/oss/driver/api/core/retry/RetryPolicyTestBase.java @@ -21,6 +21,7 @@ import static org.mockito.Mockito.mock; import com.datastax.oss.driver.api.core.ConsistencyLevel; +import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.servererrors.CoordinatorException; import com.datastax.oss.driver.api.core.servererrors.WriteType; import com.datastax.oss.driver.api.core.session.Request; @@ -58,6 +59,13 @@ protected RetryVerdictAssert assertOnUnavailable( policy.onUnavailableVerdict(request, cl, required, alive, retryCount)); } + protected RetryVerdictAssert assertOnRequestTimeout( + ConsistencyLevel cl, boolean requestIdempotent, int retryCount) { + return new RetryVerdictAssert( + policy.onRequestTimeoutVerdict( + request, cl, mock(DriverTimeoutException.class), requestIdempotent, retryCount)); + } + protected RetryVerdictAssert assertOnRequestAborted( Class errorClass, int retryCount) { return new RetryVerdictAssert( 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 c1a2765eef0..cc92ea863a6 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 @@ -19,16 +19,20 @@ import static com.datastax.oss.driver.Assertions.assertThat; import static com.datastax.oss.driver.Assertions.assertThatStage; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.datastax.oss.driver.api.core.AllNodesFailedException; import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.NoNodeAvailableException; import com.datastax.oss.driver.api.core.NodeUnavailableException; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; import com.datastax.oss.driver.api.core.cql.AsyncResultSet; import com.datastax.oss.driver.api.core.cql.BoundStatement; import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; @@ -37,6 +41,9 @@ import com.datastax.oss.driver.api.core.cql.Row; import com.datastax.oss.driver.api.core.cql.Statement; import com.datastax.oss.driver.api.core.metadata.Node; +import com.datastax.oss.driver.api.core.retry.RetryVerdict; +import com.datastax.oss.driver.api.core.session.throttling.RequestThrottler; +import com.datastax.oss.driver.api.core.session.throttling.Throttled; import com.datastax.oss.driver.internal.core.session.RepreparePayload; import com.datastax.oss.driver.internal.core.util.concurrent.CapturingTimer.CapturedTimeout; import com.datastax.oss.protocol.internal.request.Prepare; @@ -44,6 +51,7 @@ import com.datastax.oss.protocol.internal.response.result.Prepared; import com.datastax.oss.protocol.internal.response.result.SetKeyspace; import com.datastax.oss.protocol.internal.util.Bytes; +import com.tngtech.java.junit.dataprovider.UseDataProvider; import java.nio.ByteBuffer; import java.time.Duration; import java.util.Collections; @@ -55,6 +63,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.mockito.ArgumentCaptor; public class CqlRequestHandlerTest extends CqlRequestHandlerTestBase { @@ -182,6 +191,166 @@ public void should_time_out_if_first_node_takes_too_long_to_respond() throws Exc } } + @Test + @UseDataProvider("idempotentConfig") + public void should_retry_request_timeout_if_policy_decides_so( + boolean defaultIdempotence, Statement statement) throws Exception { + RequestHandlerTestHarness.Builder harnessBuilder = + RequestHandlerTestHarness.builder().withDefaultIdempotence(defaultIdempotence); + PoolBehavior node1Behavior = harnessBuilder.customBehavior(node1); + node1Behavior.setWriteSuccess(); + harnessBuilder.withResponse(node2, defaultFrameOf(singleRow())); + + try (RequestHandlerTestHarness harness = harnessBuilder.build()) { + when(harness + .getContext() + .getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME) + .onRequestTimeoutVerdict( + any(Statement.class), + eq(DefaultConsistencyLevel.LOCAL_ONE), + any(DriverTimeoutException.class), + eq(true), + eq(0))) + .thenReturn(RetryVerdict.RETRY_NEXT); + + CompletionStage resultSetFuture = + new CqlRequestHandler(statement, harness.getSession(), harness.getContext(), "test") + .handle(); + + CapturedTimeout requestTimeout = harness.nextScheduledTimeout(); + requestTimeout.task().run(requestTimeout); + + assertThatStage(resultSetFuture) + .isSuccess( + resultSet -> { + Iterator rows = resultSet.currentPage().iterator(); + assertThat(rows.hasNext()).isTrue(); + assertThat(rows.next().getString("message")).isEqualTo("hello, world"); + + ExecutionInfo executionInfo = resultSet.getExecutionInfo(); + assertThat(executionInfo.getCoordinator()).isEqualTo(node2); + assertThat(executionInfo.getErrors()).hasSize(1); + assertThat(executionInfo.getErrors().get(0).getKey()).isEqualTo(node1); + assertThat(executionInfo.getErrors().get(0).getValue()) + .isInstanceOf(DriverTimeoutException.class); + }); + node1Behavior.verifyCancellation(); + } + } + + @Test + public void should_retry_request_timeout_before_write_completes_if_policy_decides_so() + throws Exception { + RequestHandlerTestHarness.Builder harnessBuilder = RequestHandlerTestHarness.builder(); + PoolBehavior node1Behavior = harnessBuilder.customBehavior(node1); + harnessBuilder.withResponse(node2, defaultFrameOf(singleRow())); + + try (RequestHandlerTestHarness harness = harnessBuilder.build()) { + when(harness + .getContext() + .getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME) + .onRequestTimeoutVerdict( + any(Statement.class), + eq(DefaultConsistencyLevel.LOCAL_ONE), + any(DriverTimeoutException.class), + eq(true), + eq(0))) + .thenReturn(RetryVerdict.RETRY_NEXT); + + CompletionStage resultSetFuture = + new CqlRequestHandler( + IDEMPOTENT_STATEMENT, harness.getSession(), harness.getContext(), "test") + .handle(); + + CapturedTimeout requestTimeout = harness.nextScheduledTimeout(); + requestTimeout.task().run(requestTimeout); + + assertThatStage(resultSetFuture) + .isSuccess( + resultSet -> + assertThat(resultSet.getExecutionInfo().getCoordinator()).isEqualTo(node2)); + + node1Behavior.verifyCancellation(); + node1Behavior.setWriteSuccess(); + } + } + + @Test + public void should_retry_request_timeout_while_waiting_on_throttler_if_policy_decides_so() + throws Exception { + RequestThrottler throttler = mock(RequestThrottler.class); + RequestHandlerTestHarness.Builder harnessBuilder = + RequestHandlerTestHarness.builder().withRequestThrottler(throttler); + PoolBehavior node1Behavior = harnessBuilder.customBehavior(node1); + node1Behavior.setWriteSuccess(); + node1Behavior.setResponseSuccess(defaultFrameOf(singleRow())); + + try (RequestHandlerTestHarness harness = harnessBuilder.build()) { + when(harness + .getContext() + .getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME) + .onRequestTimeoutVerdict( + any(Statement.class), + eq(DefaultConsistencyLevel.LOCAL_ONE), + any(DriverTimeoutException.class), + eq(true), + eq(0))) + .thenReturn(RetryVerdict.RETRY_NEXT); + + CompletionStage resultSetFuture = + new CqlRequestHandler( + IDEMPOTENT_STATEMENT, harness.getSession(), harness.getContext(), "test") + .handle(); + + ArgumentCaptor throttledCaptor = ArgumentCaptor.forClass(Throttled.class); + verify(throttler).register(throttledCaptor.capture()); + + CapturedTimeout requestTimeout = harness.nextScheduledTimeout(); + requestTimeout.task().run(requestTimeout); + + assertThat(resultSetFuture.toCompletableFuture()).isNotDone(); + node1Behavior.verifyNoWrite(); + + throttledCaptor.getValue().onThrottleReady(true); + + assertThatStage(resultSetFuture) + .isSuccess( + resultSet -> + assertThat(resultSet.getExecutionInfo().getCoordinator()).isEqualTo(node1)); + node1Behavior.verifyWrite(); + } + } + + @Test + @UseDataProvider("nonIdempotentConfig") + public void should_pass_resolved_non_idempotence_to_policy_on_request_timeout( + boolean defaultIdempotence, Statement statement) throws Exception { + RequestHandlerTestHarness.Builder harnessBuilder = + RequestHandlerTestHarness.builder().withDefaultIdempotence(defaultIdempotence); + PoolBehavior node1Behavior = harnessBuilder.customBehavior(node1); + node1Behavior.setWriteSuccess(); + harnessBuilder.withResponse(node2, defaultFrameOf(singleRow())); + + try (RequestHandlerTestHarness harness = harnessBuilder.build()) { + CompletionStage resultSetFuture = + new CqlRequestHandler(statement, harness.getSession(), harness.getContext(), "test") + .handle(); + + CapturedTimeout requestTimeout = harness.nextScheduledTimeout(); + requestTimeout.task().run(requestTimeout); + + assertThatStage(resultSetFuture) + .isFailed(error -> assertThat(error).isInstanceOf(DriverTimeoutException.class)); + verify(harness.getContext().getRetryPolicy(DriverExecutionProfile.DEFAULT_NAME)) + .onRequestTimeoutVerdict( + any(Statement.class), + eq(DefaultConsistencyLevel.LOCAL_ONE), + any(DriverTimeoutException.class), + eq(false), + eq(0)); + } + } + @Test public void should_switch_keyspace_on_session_after_successful_use_statement() { try (RequestHandlerTestHarness harness = diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java index 8ddc66c9c4e..d902d16246d 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/RequestHandlerTestHarness.java @@ -24,14 +24,17 @@ package com.datastax.oss.driver.internal.core.cql; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import com.datastax.oss.driver.api.core.CQL4SkipMetadataResolveMethod; +import com.datastax.oss.driver.api.core.ConsistencyLevel; import com.datastax.oss.driver.api.core.CqlIdentifier; import com.datastax.oss.driver.api.core.DefaultConsistencyLevel; +import com.datastax.oss.driver.api.core.DriverTimeoutException; import com.datastax.oss.driver.api.core.ProtocolVersion; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; import com.datastax.oss.driver.api.core.config.DriverConfig; @@ -40,8 +43,10 @@ import com.datastax.oss.driver.api.core.metadata.Node; import com.datastax.oss.driver.api.core.metrics.SessionMetric; import com.datastax.oss.driver.api.core.retry.RetryPolicy; +import com.datastax.oss.driver.api.core.retry.RetryVerdict; import com.datastax.oss.driver.api.core.session.Request; import com.datastax.oss.driver.api.core.session.Session; +import com.datastax.oss.driver.api.core.session.throttling.RequestThrottler; import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy; import com.datastax.oss.driver.api.core.time.TimestampGenerator; import com.datastax.oss.driver.api.core.tracker.RequestIdGenerator; @@ -137,6 +142,13 @@ protected RequestHandlerTestHarness(Builder builder) { when(context.getLoadBalancingPolicyWrapper()).thenReturn(loadBalancingPolicyWrapper); when(context.getRetryPolicy(anyString())).thenReturn(retryPolicy); + when(retryPolicy.onRequestTimeoutVerdict( + any(Request.class), + any(ConsistencyLevel.class), + any(DriverTimeoutException.class), + anyBoolean(), + anyInt())) + .thenReturn(RetryVerdict.RETHROW); // Disable speculative executions by default when(speculativeExecutionPolicy.nextExecution( @@ -193,7 +205,11 @@ protected RequestHandlerTestHarness(Builder builder) { when(context.getWriteTypeRegistry()).thenReturn(new DefaultWriteTypeRegistry()); - when(context.getRequestThrottler()).thenReturn(new PassThroughRequestThrottler(context)); + when(context.getRequestThrottler()) + .thenReturn( + builder.requestThrottler == null + ? new PassThroughRequestThrottler(context) + : builder.requestThrottler); when(context.getRequestTracker()).thenReturn(new NoopRequestTracker(context)); @@ -232,6 +248,7 @@ public static class Builder { private boolean defaultIdempotence; private ProtocolVersion protocolVersion; private RequestIdGenerator requestIdGenerator; + private RequestThrottler requestThrottler; /** * Sets the given node as the next one in the query plan; an empty pool will be simulated when @@ -292,6 +309,11 @@ public Builder withRequestIdGenerator(RequestIdGenerator requestIdGenerator) { return this; } + public Builder withRequestThrottler(RequestThrottler requestThrottler) { + this.requestThrottler = requestThrottler; + return this; + } + /** * Sets the given node as the next one in the query plan; the test code is responsible of * calling the methods on the returned object to complete the write and the query. diff --git a/manual/core/retries/README.md b/manual/core/retries/README.md index e92f8e214aa..b9d6d7c81c4 100644 --- a/manual/core/retries/README.md +++ b/manual/core/retries/README.md @@ -166,6 +166,24 @@ nodes will likely have been detected as dead, and the retry has a high chance of * For `BATCH_LOG` write type: the policy will retry the same node, for the reasons explained above. * For other write types: the policy will always rethrow. +#### `onRequestTimeoutVerdict` + +The driver sent the request to a coordinator, but did not receive any response before the +client-side request timeout (`basic.request.timeout`) expired. The driver raises a +[DriverTimeoutException] and invokes this method so that the retry policy can decide whether to +retry. + +This is different from `onReadTimeoutVerdict` and `onWriteTimeoutVerdict`: those methods handle +server-side timeout responses, where the coordinator replied with `READ_TIMEOUT` or `WRITE_TIMEOUT`. +With a driver-side timeout, the coordinator might still have executed the request and only the +response was delayed or lost. + +This method is invoked with the resolved statement idempotence. Policies that retry this error should +reject non-idempotent requests unless duplicate execution is acceptable. + +The default policy rethrows client-side request timeouts. `ConsistencyDowngradingRetryPolicy` retries +them on the next node for idempotent requests only. + #### `onRequestAbortedVerdict` The request was aborted before we could get a response from the coordinator. This can happen in two