scylladb · arnabnandy7 · Jul 19, 2026
diff --git a/changelog/README.md b/changelog/README.md
@@ -23,6 +23,7 @@ under the License.
 
 ### 4.19.2
 
+- [bug] PR 947: Release pre-acquired stream IDs when requests fail before being written
 - [bug] CASSJAVA-116: Retry or Speculative Execution with RequestIdGenerator throws "Duplicate Key"
 
 ### 4.19.1

diff --git a/...va/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java b/...va/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java
@@ -367,23 +367,33 @@ private void sendRequest(
         abortGlobalRequestOrChosenCallback(AllNodesFailedException.fromErrors(errors));
       }
     } else if (!chosenCallback.isDone()) {
-      NodeResponseCallback nodeResponseCallback =
-          new NodeResponseCallback(
-              statement,
-              node,
-              channel,
-              currentExecutionIndex,
-              retryCount,
-              scheduleSpeculativeExecution,
-              logPrefix);
-      inFlightCallbacks.add(nodeResponseCallback);
-      channel
-          .write(
-              getMessage(statement),
-              isTracingEnabled(statement),
-              createPayload(statement),
-              nodeResponseCallback)
-          .addListener(nodeResponseCallback);
+      boolean writeSubmitted = false;
+      try {
+        NodeResponseCallback nodeResponseCallback =
+            new NodeResponseCallback(
+                statement,
+                node,
+                channel,
+                currentExecutionIndex,
+                retryCount,
+                scheduleSpeculativeExecution,
+                logPrefix);
+        inFlightCallbacks.add(nodeResponseCallback);
+        channel
+            .write(
+                getMessage(statement),
+                isTracingEnabled(statement),
+                createPayload(statement),
+                nodeResponseCallback)
+            .addListener(nodeResponseCallback);
+        writeSubmitted = true;
+      } finally {
+        if (!writeSubmitted) {
+          channel.cancelPreAcquireId();
+        }
+      }
+    } else {
+      channel.cancelPreAcquireId();
     }
   }
 

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
@@ -302,29 +302,37 @@ private void sendRequest(
             NO_SUCCESSFUL_EXECUTION);
       }
     } else {
-      NodeResponseCallback nodeResponseCallback =
-          new NodeResponseCallback(
-              statement,
-              node,
-              queryPlan,
-              channel,
-              currentExecutionIndex,
-              retryCount,
-              scheduleNextExecution,
-              logPrefix);
-      DriverExecutionProfile executionProfile =
-          Conversions.resolveExecutionProfile(statement, context);
-      GraphProtocol graphSubProtocol =
-          GraphConversions.resolveGraphSubProtocol(statement, graphSupportChecker, context);
-      Message message =
-          GraphConversions.createMessageFromGraphStatement(
-              statement, graphSubProtocol, executionProfile, context, graphBinaryModule);
-      Map<String, ByteBuffer> customPayload =
-          GraphConversions.createCustomPayload(
-              statement, graphSubProtocol, executionProfile, context, graphBinaryModule);
-      channel
-          .write(message, statement.isTracing(), customPayload, nodeResponseCallback)
-          .addListener(nodeResponseCallback);
+      boolean writeSubmitted = false;
+      try {
+        NodeResponseCallback nodeResponseCallback =
+            new NodeResponseCallback(
+                statement,
+                node,
+                queryPlan,
+                channel,
+                currentExecutionIndex,
+                retryCount,
+                scheduleNextExecution,
+                logPrefix);
+        DriverExecutionProfile executionProfile =
+            Conversions.resolveExecutionProfile(statement, context);
+        GraphProtocol graphSubProtocol =
+            GraphConversions.resolveGraphSubProtocol(statement, graphSupportChecker, context);
+        Message message =
+            GraphConversions.createMessageFromGraphStatement(
+                statement, graphSubProtocol, executionProfile, context, graphBinaryModule);
+        Map<String, ByteBuffer> customPayload =
+            GraphConversions.createCustomPayload(
+                statement, graphSubProtocol, executionProfile, context, graphBinaryModule);
+        channel
+            .write(message, statement.isTracing(), customPayload, nodeResponseCallback)
+            .addListener(nodeResponseCallback);
+        writeSubmitted = true;
+      } finally {
+        if (!writeSubmitted) {
+          channel.cancelPreAcquireId();
+        }
+      }
     }
   }
 

diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/DriverChannel.java
@@ -100,10 +100,16 @@ public Future<Void> write(
       Map<String, ByteBuffer> customPayload,
       ResponseCallback responseCallback) {
     if (closing.get()) {
+      inFlightHandler.cancelPreAcquireId();
       return channel.newFailedFuture(new IllegalStateException("Driver channel is closing"));
     }
     RequestMessage message = new RequestMessage(request, tracing, customPayload, responseCallback);
-    return writeCoalescer.writeAndFlush(channel, message);
+    try {
+      return writeCoalescer.writeAndFlush(channel, message);
+    } catch (Throwable t) {
+      inFlightHandler.cancelPreAcquireId();
+      return channel.newFailedFuture(t);
+    }
   }
 
   /**
@@ -220,6 +226,15 @@ public boolean preAcquireId() {
     return inFlightHandler.preAcquireId();
   }
 
+  /**
+   * Cancels the reservation made by a successful {@link #preAcquireId()} call when the
+   * corresponding request cannot be submitted to {@link #write(Message, boolean, Map,
+   * ResponseCallback)}.
+   */
+  public void cancelPreAcquireId() {
+    inFlightHandler.cancelPreAcquireId();
+  }
+
   /**
    * @return the number of requests currently executing on this channel (including {@link
    *     #getOrphanedIds() orphaned ids}).

diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/InFlightHandler.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/InFlightHandler.java
@@ -396,6 +396,10 @@ boolean preAcquireId() {
     return streamIds.preAcquire();
   }
 
+  void cancelPreAcquireId() {
+    streamIds.cancelPreAcquire();
+  }
+
   int getInFlight() {
     return streamIds.getMaxAvailableIds() - streamIds.getAvailableIds();
   }

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
@@ -229,14 +229,22 @@ private void sendRequest(PrepareRequest request, Node node, int retryCount) {
     if (channel == null) {
       setFinalError(AllNodesFailedException.fromErrors(this.errors));
     } else {
-      InitialPrepareCallback initialPrepareCallback =
-          new InitialPrepareCallback(request, node, channel, retryCount);
-
-      Prepare message = toPrepareMessage(request);
-
-      channel
-          .write(message, false, request.getCustomPayload(), initialPrepareCallback)
-          .addListener(initialPrepareCallback);
+      boolean writeSubmitted = false;
+      try {
+        InitialPrepareCallback initialPrepareCallback =
+            new InitialPrepareCallback(request, node, channel, retryCount);
+
+        Prepare message = toPrepareMessage(request);
+
+        channel
+            .write(message, false, request.getCustomPayload(), initialPrepareCallback)
+            .addListener(initialPrepareCallback);
+        writeSubmitted = true;
+      } finally {
+        if (!writeSubmitted) {
+          channel.cancelPreAcquireId();
+        }
+      }
     }
   }
 
@@ -316,28 +324,38 @@ private CompletionStage<Void> prepareOnOtherNode(PrepareRequest request, Node no
       LOG.trace("[{}] Could not get a channel to reprepare on {}, skipping", logPrefix, node);
       return CompletableFuture.completedFuture(null);
     } else {
-      ThrottledAdminRequestHandler<ByteBuffer> handler =
-          ThrottledAdminRequestHandler.prepare(
-              channel,
-              false,
-              toPrepareMessage(request),
-              request.getCustomPayload(),
-              Conversions.resolveRequestTimeout(request, executionProfile),
-              throttler,
-              session.getMetricUpdater(),
-              logPrefix);
-      return handler
-          .start()
-          .handle(
-              (result, error) -> {
-                if (error == null) {
-                  LOG.trace("[{}] Successfully reprepared on {}", logPrefix, node);
-                } else {
-                  Loggers.warnWithException(
-                      LOG, "[{}] Error while repreparing on {}", node, logPrefix, error);
-                }
-                return null;
-              });
+      boolean requestStarted = false;
+      try {
+        ThrottledAdminRequestHandler<ByteBuffer> handler =
+            ThrottledAdminRequestHandler.prepare(
+                channel,
+                false,
+                toPrepareMessage(request),
+                request.getCustomPayload(),
+                Conversions.resolveRequestTimeout(request, executionProfile),
+                throttler,
+                session.getMetricUpdater(),
+                logPrefix);
+        CompletionStage<Void> result =
+            handler
+                .start()
+                .handle(
+                    (preparedId, error) -> {
+                      if (error == null) {
+                        LOG.trace("[{}] Successfully reprepared on {}", logPrefix, node);
+                      } else {
+                        Loggers.warnWithException(
+                            LOG, "[{}] Error while repreparing on {}", node, logPrefix, error);
+                      }
+                      return null;
+                    });
+        requestStarted = true;
+        return result;
+      } finally {
+        if (!requestStarted) {
+          channel.cancelPreAcquireId();
+        }
+      }
     }
   }
 

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
@@ -409,30 +409,39 @@ private void sendRequest(
         setFinalError(statement, AllNodesFailedException.fromErrors(this.errors), null, -1);
       }
     } else {
-      Statement finalStatement = statement;
-      String nodeRequestId =
-          this.requestIdGenerator
-              .map((g) -> g.getNodeRequestId(finalStatement, sessionRequestId))
-              .orElse(Integer.toString(this.hashCode()));
-      statement =
-          this.requestIdGenerator
-              .map((g) -> g.getDecoratedStatement(finalStatement, nodeRequestId))
-              .orElse(finalStatement);
+      boolean writeSubmitted = false;
+      try {
+        Statement finalStatement = statement;
+        String nodeRequestId =
+            this.requestIdGenerator
+                .map((g) -> g.getNodeRequestId(finalStatement, sessionRequestId))
+                .orElse(Integer.toString(this.hashCode()));
+        statement =
+            this.requestIdGenerator
+                .map((g) -> g.getDecoratedStatement(finalStatement, nodeRequestId))
+                .orElse(finalStatement);
 
-      NodeResponseCallback nodeResponseCallback =
-          new NodeResponseCallback(
-              statement,
-              node,
-              queryPlan,
-              channel,
-              currentExecutionIndex,
-              retryCount,
-              scheduleNextExecution,
-              logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex));
-      Message message = Conversions.toMessage(statement, executionProfile, context);
-      channel
-          .write(message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback)
-          .addListener(nodeResponseCallback);
+        NodeResponseCallback nodeResponseCallback =
+            new NodeResponseCallback(
+                statement,
+                node,
+                queryPlan,
+                channel,
+                currentExecutionIndex,
+                retryCount,
+                scheduleNextExecution,
+                logPrefixJoiner.join(this.sessionName, nodeRequestId, currentExecutionIndex));
+        Message message = Conversions.toMessage(statement, executionProfile, context);
+        channel
+            .write(
+                message, statement.isTracing(), statement.getCustomPayload(), nodeResponseCallback)
+            .addListener(nodeResponseCallback);
+        writeSubmitted = true;
+      } finally {
+        if (!writeSubmitted) {
+          channel.cancelPreAcquireId();
+        }
+      }
     }
   }
 

diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/DriverChannelTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/DriverChannelTest.java
@@ -18,6 +18,7 @@
 package com.datastax.oss.driver.internal.core.channel;
 
 import static com.datastax.oss.driver.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
 
 import com.datastax.oss.driver.api.core.DefaultProtocolVersion;
 import com.datastax.oss.driver.api.core.connection.ClosedConnectionException;
@@ -143,6 +144,17 @@ public void should_wait_for_coalesced_writes_when_closing_forcefully() {
         .hasMessageContaining("Channel was force-closed");
   }
 
+  @Test
+  public void should_cancel_pre_acquired_id_when_write_is_rejected_before_submission() {
+    driverChannel.close();
+
+    Future<java.lang.Void> writeFuture =
+        driverChannel.write(new Query("test"), false, Frame.NO_PAYLOAD, new MockResponseCallback());
+
+    assertThat(writeFuture).isFailed();
+    verify(streamIds).cancelPreAcquire();
+  }
+
   // Simple implementation that holds all the writes, and flushes them when it's explicitly
   // triggered.
   private class MockWriteCoalescer implements WriteCoalescer {