4.x: throttler concurrency slot and request timeout leak when onThrottleReady() throws

[nikagra]

Problem

ConcurrencyLimitingRequestThrottler.register() claims a concurrency slot and then invokes the
request synchronously:

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

Lines 97 to 103 in bff77b2

	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;

If onThrottleReady() throws, the exception propagates out through register() and
concurrentRequests is never decremented. The slot is lost for the lifetime of the session.

The same shape exists on the dequeue side — signalSuccess() calls onThrottleReady(true) on the
next request while the current one is completing:

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

Lines 138 to 149 in bff77b2

	@Override
	public void signalSuccess(@NonNull Throttled request) {
	Throttled nextRequest = onRequestDoneAndDequeNext();
	if (nextRequest != null) {
	nextRequest.onThrottleReady(true);
	}
	}
	@Override
	public void signalError(@NonNull Throttled request, @NonNull Throwable error) {
	signalSuccess(request); // not treated differently
	}

so a throw there escapes into an unrelated request's completion path as well.

Affected paths

1. CqlRequestHandler's constructor. It registers with the throttler as its last act, and with
the default PassThroughRequestThrottler — or ConcurrencyLimitingRequestThrottler under the
limit — onThrottleReady() runs synchronously, on the caller's thread, inside the constructor:

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

Lines 204 to 211 in bff77b2

	this.timer = context.getNettyOptions().getTimer();
	this.executionProfile = Conversions.resolveExecutionProfile(initialStatement, context);
	Duration timeout = Conversions.resolveRequestTimeout(statement, executionProfile);
	this.scheduledTimeout = scheduleTimeout(timeout);
	this.throttler = context.getRequestThrottler();
	this.throttler.register(this);
	}

onThrottleReady() → sendRequest(), which can throw from RequestIdGenerator statement
decoration or from Conversions.toMessage(). That exception escapes session.execute() synchronously
rather than completing the returned future, and the slot is gone.

2. ThrottledAdminRequestHandler.onThrottleReady() deliberately rethrows after releasing the
stream id (added in #965):

java-driver/core/src/main/java/com/datastax/oss/driver/internal/core/adminrequest/ThrottledAdminRequestHandler.java

Lines 148 to 164 in bff77b2

	@Override
	public void onThrottleReady(boolean wasDelayed) {
	try {
	if (wasDelayed) {
	metricUpdater.updateTimer(
	DefaultSessionMetric.THROTTLING_DELAY,
	null,
	System.nanoTime() - startTimeNanos,
	TimeUnit.NANOSECONDS);
	}
	} catch (Throwable t) {
	cancelExternallyPreAcquiredId();
	throw t;
	}
	externallyPreAcquiredId.set(false);
	super.start();
	}

Second leak on the same path

CqlRequestHandler.scheduledTimeout is assigned at L207, before throttler.register(this) at
L210. When the constructor throws, nothing cancels it: the timer entry survives for the full request
timeout and then fires setFinalError on a result future no caller holds.

Impact

Strictly worse than #947. That one was per-channel stream-id drift, self-healing when the channel is
recycled. This is per-session and permanent: a session on
ConcurrencyLimitingRequestThrottler monotonically loses request concurrency and eventually wedges
at advanced.throttler.max-concurrent-requests, with every subsequent request queued and then
rejected. With PassThroughRequestThrottler (the default) there is no slot to leak, so only the
scheduledTimeout half applies.

Relationship to #965

#965 fixes the stream-id half of the accounting on exactly these paths, and its new test
should_cancel_pre_acquired_id_if_request_decoration_fails_before_write exercises path 1 — asserting
that the exception escapes the constructor. Worth reading that test as "the stream id is released
here", not "this path is handled".

Expected

A throw out of onThrottleReady() should leave the throttler's accounting unchanged, and should not
leave scheduled work behind.

Possible fixes

• Have sendRequest() route failures through setFinalError(...) instead of propagating, so the
  normal completion path (which already signals the throttler and cancels scheduled tasks) runs. This
  also makes session.execute() fail through its future rather than synchronously, which is the more
  consistent contract.
• Or make register() / signalSuccess() release their claim if onThrottleReady() throws.
• Either way, move scheduleTimeout() after registration, or cancel it on the failure path.

[nikagra]

Third affected path

Found while re-reviewing #965 at ef698f9.

ContinuousRequestHandlerBase.sendRequest() leaks activeExecutionsCount on the same class of failure. The write-setup finally unwinds the stream id and the in-flight callback, but not the execution count:

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

Lines 391 to 398 in ef698f9

	} finally {
	if (!writeSubmitted) {
	if (nodeResponseCallback != null) {
	inFlightCallbacks.remove(nodeResponseCallback);
	}
	channel.cancelPreAcquireId();
	}
	}

whereas the channel == null branch a few lines above does, and can abort the session as a result:

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

Lines 363 to 368 in ef698f9

	if (channel == null) {
	// 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));
	}

So a throw from getMessage(), createPayload() or the NodeResponseCallback constructor — after the channel is acquired, before the write is submitted — decrements nothing.

Different counter and a different handler from paths 1 and 2, but the same shape: a synchronous throw between channel acquisition and InFlightHandler leaves per-request bookkeeping inflated with no owner left to unwind it.

Distinct consequence

This one doesn't lose a slot, it hangs. If the throwing execution was the last live one, nothing calls abortGlobalRequestOrChosenCallback, so the caller's CompletionStage is never completed and only the global timeout ends the continuous paging session. That's independent of which throttler is configured — unlike the concurrency-slot half, which needs ConcurrencyLimitingRequestThrottler.

Relationship to #965

#965 is right not to touch this. Its finally deliberately scopes to the stream id, which is #947's subject, and the else branch it adds at L400 is genuinely fine — chosenCallback is already done there, so the result is set and the count no longer matters. Recording it here so it isn't lost when someone picks this issue up.