fix: enrich DriverTimeoutException with per-node pool/channel diagnostics (DRIVER-540)

Add NodeDiagnostics public inner class to DriverTimeoutException with fields
for in-flight counts and pool capacity, and generate a diagnostic suffix in
the exception message at timeout time. Refactor all four request-handler
timeout paths (CqlRequestHandler, CqlPrepareHandler, GraphRequestHandler,
ContinuousRequestHandlerBase) to build List<NodeDiagnostics> instead of a
raw message string. Update IT assertions that matched exact message strings
to use hasMessageStartingWith() to accommodate the new suffix.

Author: nikagra

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

@@ -17,6 +17,8 @@
  */
 package com.datastax.dse.driver.internal.core.cql.continuous;
 
+import static com.datastax.oss.driver.api.core.DriverTimeoutException.UNAVAILABLE;
+
 import com.datastax.dse.driver.api.core.DseProtocolVersion;
 import com.datastax.dse.driver.api.core.cql.continuous.ContinuousAsyncResultSet;
 import com.datastax.dse.driver.internal.core.DseProtocolFeature;
@@ -26,6 +28,7 @@
 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;
 import com.datastax.oss.driver.api.core.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
 import com.datastax.oss.driver.api.core.RequestThrottlingException;
@@ -62,6 +65,7 @@
 import com.datastax.oss.driver.internal.core.metadata.DefaultNode;
 import com.datastax.oss.driver.internal.core.metrics.NodeMetricUpdater;
 import com.datastax.oss.driver.internal.core.metrics.SessionMetricUpdater;
+import com.datastax.oss.driver.internal.core.pool.ChannelPool;
 import com.datastax.oss.driver.internal.core.session.DefaultSession;
 import com.datastax.oss.driver.internal.core.session.RepreparePayload;
 import com.datastax.oss.driver.internal.core.util.Loggers;
@@ -89,6 +93,8 @@
 import java.time.Duration;
 import java.util.AbstractMap;
 import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -390,13 +396,32 @@ private Timeout scheduleGlobalTimeout() {
     }
     LOG.trace("[{}] Scheduling global timeout for pages in {}", logPrefix, globalTimeout);
     return timer.newTimeout(
-        timeout ->
-            abortGlobalRequestOrChosenCallback(
-                new DriverTimeoutException("Query timed out after " + globalTimeout)),
+        timeout -> {
+          List<NodeDiagnostics> diagnostics = buildNodeDiagnostics();
+          abortGlobalRequestOrChosenCallback(
+              new DriverTimeoutException("Query timed out after " + globalTimeout, diagnostics));
+        },
         globalTimeout.toNanos(),
         TimeUnit.NANOSECONDS);
   }
 
+  private List<NodeDiagnostics> buildNodeDiagnostics() {
+    List<NodeResponseCallback> callbacks = inFlightCallbacks;
+    List<NodeDiagnostics> result = new ArrayList<>(callbacks.size());
+    for (NodeResponseCallback cb : callbacks) {
+      int channelInFlight = cb.channel.getInFlight();
+      ChannelPool pool = session.getPools().get(cb.node);
+      result.add(
+          NodeDiagnostics.of(
+              cb.node.getEndPoint(),
+              channelInFlight,
+              pool != null ? pool.getInFlight() : UNAVAILABLE,
+              pool != null ? pool.getAvailableIds() : UNAVAILABLE,
+              pool != null ? pool.getOrphanedIds() : UNAVAILABLE));
+    }
+    return result;
+  }
+
   /**
    * Cancels the continuous paging request.
    *
@@ -718,9 +743,18 @@ private void onPageTimeout(int expectedPage) {
       lock.lock();
       try {
         if (state == expectedPage) {
+          int channelInFlight = channel.getInFlight();
+          ChannelPool pool = session.getPools().get(node);
           abort(
               new DriverTimeoutException(
-                  String.format("Timed out waiting for page %d", expectedPage)),
+                  "Timed out waiting for page " + expectedPage,
+                  Collections.singletonList(
+                      NodeDiagnostics.of(
+                          node.getEndPoint(),
+                          channelInFlight,
+                          pool != null ? pool.getInFlight() : UNAVAILABLE,
+                          pool != null ? pool.getAvailableIds() : UNAVAILABLE,
+                          pool != null ? pool.getOrphanedIds() : UNAVAILABLE))),
               false);
         } else {
           // Ignore timeout if the request has moved on in the interim.

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

@@ -17,6 +17,8 @@
  */
 package com.datastax.dse.driver.internal.core.graph;
 
+import static com.datastax.oss.driver.api.core.DriverTimeoutException.UNAVAILABLE;
+
 import com.datastax.dse.driver.api.core.graph.AsyncGraphResultSet;
 import com.datastax.dse.driver.api.core.graph.GraphNode;
 import com.datastax.dse.driver.api.core.graph.GraphStatement;
@@ -26,6 +28,7 @@
 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;
 import com.datastax.oss.driver.api.core.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.RequestThrottlingException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
@@ -56,6 +59,7 @@
 import com.datastax.oss.driver.internal.core.metadata.DefaultNode;
 import com.datastax.oss.driver.internal.core.metrics.NodeMetricUpdater;
 import com.datastax.oss.driver.internal.core.metrics.SessionMetricUpdater;
+import com.datastax.oss.driver.internal.core.pool.ChannelPool;
 import com.datastax.oss.driver.internal.core.session.DefaultSession;
 import com.datastax.oss.driver.internal.core.tracker.NoopRequestTracker;
 import com.datastax.oss.driver.internal.core.tracker.RequestLogger;
@@ -77,6 +81,7 @@
 import java.time.Duration;
 import java.util.AbstractMap;
 import java.util.ArrayDeque;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -209,12 +214,15 @@ private Timeout scheduleTimeout(Duration timeoutDuration) {
     if (timeoutDuration != null && timeoutDuration.toNanos() > 0) {
       try {
         return this.timer.newTimeout(
-            (Timeout timeout1) ->
-                setFinalError(
-                    initialStatement,
-                    new DriverTimeoutException("Query timed out after " + timeoutDuration),
-                    null,
-                    NO_SUCCESSFUL_EXECUTION),
+            (Timeout timeout1) -> {
+              List<NodeDiagnostics> diagnostics = buildNodeDiagnostics();
+              setFinalError(
+                  initialStatement,
+                  new DriverTimeoutException(
+                      "Query timed out after " + timeoutDuration, diagnostics),
+                  null,
+                  NO_SUCCESSFUL_EXECUTION);
+            },
             timeoutDuration.toNanos(),
             TimeUnit.NANOSECONDS);
       } catch (IllegalStateException e) {
@@ -229,6 +237,23 @@ private Timeout scheduleTimeout(Duration timeoutDuration) {
     return null;
   }
 
+  private List<NodeDiagnostics> buildNodeDiagnostics() {
+    List<NodeResponseCallback> callbacks = inFlightCallbacks;
+    List<NodeDiagnostics> result = new ArrayList<>(callbacks.size());
+    for (NodeResponseCallback cb : callbacks) {
+      int channelInFlight = cb.channel.getInFlight();
+      ChannelPool pool = session.getPools().get(cb.node);
+      result.add(
+          NodeDiagnostics.of(
+              cb.node.getEndPoint(),
+              channelInFlight,
+              pool != null ? pool.getInFlight() : UNAVAILABLE,
+              pool != null ? pool.getAvailableIds() : UNAVAILABLE,
+              pool != null ? pool.getOrphanedIds() : UNAVAILABLE));
+    }
+    return result;
+  }
+
   /**
    * Sends the request to the next available node.
    *