4.x: onThrottleReady exceptions can escape or leave request futures incomplete

[dkropachev]

Problem

Request handlers do not have a top-level error boundary around onThrottleReady().

The throttler calls onThrottleReady() directly. If query-plan creation or sendRequest() throws synchronously, behavior depends on throttler timing:

• immediate throttler path: exception can escape from executeAsync() instead of returning a failed CompletionStage
• queued throttler path: exception happens later on the throttler/event-loop thread and the request future may not complete cleanly

Proof

Throttlers call onThrottleReady() directly:

• pass-through throttler:

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/PassThroughRequestThrottler.java

  Lines 52 to 55 in 3cafb2a

  	@Override
  	public void register(@NonNull Throttled request) {
  	request.onThrottleReady(false);
  	}

• concurrency throttler immediate path:

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java

  Lines 96 to 103 in 3cafb2a

  	// If no backlog exists AND we get capacity, we can execute immediately
  	if (queueSize.get() == 0) {
  	// Take a claim first, and then check if we are OK to proceed
  	int newConcurrent = concurrentRequests.incrementAndGet();
  	if (newConcurrent <= maxConcurrentRequests) {
  	LOG.trace("[{}] Starting newly registered request", logPrefix);
  	request.onThrottleReady(false);
  	return;

• concurrency throttler queued path:

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/ConcurrencyLimitingRequestThrottler.java

  Lines 138 to 143 in 3cafb2a

  	@Override
  	public void signalSuccess(@NonNull Throttled request) {
  	Throttled nextRequest = onRequestDoneAndDequeNext();
  	if (nextRequest != null) {
  	nextRequest.onThrottleReady(true);
  	}

• rate throttler immediate/queued paths:

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/RateLimitingRequestThrottler.java

  Lines 127 to 130 in 3cafb2a

  	} else if (queue.isEmpty() && acquire(now, 1) == 1) {
  	LOG.trace("[{}] Starting newly registered request", logPrefix);
  	request.onThrottleReady(false);
  	} else if (queue.size() < maxQueueSize) {

  and

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/session/throttling/RateLimitingRequestThrottler.java

  Lines 163 to 166 in 3cafb2a

  	for (int i = 0; i < toDequeue; i++) {
  	LOG.trace("[{}] Starting dequeued request", logPrefix);
  	queue.poll().onThrottleReady(true);
  	}

CQL handler does query-plan creation and request scheduling inside onThrottleReady() without a catch-all boundary:

java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlRequestHandler.java

Lines 212 to 235 in 3cafb2a

	@Override
	public void onThrottleReady(boolean wasDelayed) {
	if (wasDelayed
	// avoid call to nanoTime() if metric is disabled:
	&& sessionMetricUpdater.isEnabled(
	DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
	sessionMetricUpdater.updateTimer(
	DefaultSessionMetric.THROTTLING_DELAY,
	executionProfile.getName(),
	System.nanoTime() - startTimeNanos,
	TimeUnit.NANOSECONDS);
	}
	Queue<Node> queryPlan;
	if (this.initialStatement.getNode() != null) {
	queryPlan = new SimpleQueryPlan(this.initialStatement.getNode());
	} else {
	queryPlan =
	context
	.getLoadBalancingPolicyWrapper()
	.newQueryPlan(initialStatement, executionProfile.getName(), session);
	}
	sendRequest(initialStatement, null, queryPlan, 0, 0, true);

LoadBalancingPolicyWrapper.newQueryPlan() calls the policy directly in RUNNING state:

java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java

Lines 144 to 160 in 3cafb2a

	@NonNull
	public Queue<Node> newQueryPlan(
	@Nullable Request request, @NonNull String executionProfileName, @Nullable Session session) {
	switch (stateRef.get()) {
	case BEFORE_INIT:
	case DURING_INIT:
	// The contact points are not stored in the metadata yet:
	List<Node> nodes = new ArrayList<>(context.getMetadataManager().getContactPoints());
	Collections.shuffle(nodes);
	return new ConcurrentLinkedQueue<>(nodes);
	case RUNNING:
	LoadBalancingPolicy policy = policiesPerProfile.get(executionProfileName);
	if (policy == null) {
	policy = policiesPerProfile.get(DriverExecutionProfile.DEFAULT_NAME);
	}
	return policy.newQueryPlan(request, session);
	default:

Similar unguarded onThrottleReady() scheduling exists in:

• prepare:

  java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandler.java

  Lines 149 to 163 in 3cafb2a

  	@Override
  	public void onThrottleReady(boolean wasDelayed) {
  	DriverExecutionProfile executionProfile =
  	Conversions.resolveExecutionProfile(initialRequest, context);
  	if (wasDelayed) {
  	session
  	.getMetricUpdater()
  	.updateTimer(
  	DefaultSessionMetric.THROTTLING_DELAY,
  	executionProfile.getName(),
  	System.nanoTime() - startTimeNanos,
  	TimeUnit.NANOSECONDS);
  	}
  	sendRequest(initialRequest, null, 0);
  	}

• graph:

  java-driver/core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphRequestHandler.java

  Lines 186 to 206 in 3cafb2a

  	@Override
  	public void onThrottleReady(boolean wasDelayed) {
  	DriverExecutionProfile executionProfile =
  	Conversions.resolveExecutionProfile(initialStatement, context);
  	if (wasDelayed
  	// avoid call to nanoTime() if metric is disabled:
  	&& sessionMetricUpdater.isEnabled(
  	DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
  	sessionMetricUpdater.updateTimer(
  	DefaultSessionMetric.THROTTLING_DELAY,
  	executionProfile.getName(),
  	System.nanoTime() - startTimeNanos,
  	TimeUnit.NANOSECONDS);
  	}
  	Queue<Node> queryPlan =
  	initialStatement.getNode() != null
  	? new SimpleQueryPlan(initialStatement.getNode())
  	: context
  	.getLoadBalancingPolicyWrapper()
  	.newQueryPlan(initialStatement, executionProfile.getName(), session);
  	sendRequest(initialStatement, null, queryPlan, 0, 0, true);

• continuous CQL:

  java-driver/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java

  Lines 255 to 272 in 3cafb2a

  	@Override
  	public void onThrottleReady(boolean wasDelayed) {
  	DriverExecutionProfile executionProfile =
  	Conversions.resolveExecutionProfile(initialStatement, context);
  	if (wasDelayed
  	// avoid call to nanoTime() if metric is disabled:
  	&& sessionMetricUpdater.isEnabled(
  	DefaultSessionMetric.THROTTLING_DELAY, executionProfile.getName())) {
  	session
  	.getMetricUpdater()
  	.updateTimer(
  	DefaultSessionMetric.THROTTLING_DELAY,
  	executionProfile.getName(),
  	System.nanoTime() - startTimeNanos,
  	TimeUnit.NANOSECONDS);
  	}
  	activeExecutionsCount.incrementAndGet();
  	sendRequest(initialStatement, null, 0, 0, specExecEnabled);

Expected

Any exception thrown from onThrottleReady() scheduling should complete the request with a failed future consistently, regardless of throttler implementation or whether the request was delayed.

Suggested fix

Add a top-level try/catch around handler onThrottleReady() scheduling and route failures through existing final-error paths.