JAVA-2959: Don't throw NoNodeAvailableException when all connections busy (#1570)

akhaku · web-flow · commit 0a7b1c86d077 · 2021-08-24T10:13:50.000+02:00
For cases in which there are no connections available to
send requests to a Node in the query plan, collect the error
rather than silently skipping over the node. The error will
be thrown as part of an AllNodesFailedException if all nodes
fail.
This can happen when we've saturated the max in-flight
requests across all nodes or when the request is directed to
a particular node and it has no connections available (or
all its connections are saturated).
Note that in the latter case we used to throw a
NoNodeAvailableException but we now throw
AllNodesFailedException.
diff --git a/changelog/README.md b/changelog/README.md
@@ -2,6 +2,10 @@
 
 <!-- Note: contrary to 3.x, insert new entries *first* in their section -->
 
+### 4.14.0 (in progress)
+
+- [improvement] JAVA-2959: Don't throw NoNodeAvailableException when all connections busy
+
 ### 4.13.0
 
 - [improvement] JAVA-2940: Add GraalVM native image build configurations
diff --git a/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java b/core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java
@@ -24,6 +24,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.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
 import com.datastax.oss.driver.api.core.RequestThrottlingException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
@@ -348,6 +349,8 @@ private void sendRequest(
         channel = session.getChannel(node, logPrefix);
         if (channel != null) {
           break;
+        } else {
+          recordError(node, new NodeUnavailableException(node));
         }
       }
     }
@@ -455,6 +458,10 @@ CompletableFuture<ResultSetT> getPendingResult() {
     }
   }
 
+  private void recordError(@NonNull Node node, @NonNull Throwable error) {
+    errors.add(new AbstractMap.SimpleEntry<>(node, error));
+  }
+
   /**
    * Handles the interaction with a single node in the query plan.
    *
@@ -1433,10 +1440,6 @@ private void reenableAutoReadIfNeeded() {
 
     // ERROR HANDLING
 
-    private void recordError(@NonNull Node node, @NonNull Throwable error) {
-      errors.add(new AbstractMap.SimpleEntry<>(node, error));
-    }
-
     private void trackNodeError(@NonNull Node node, @NonNull Throwable error) {
       if (nodeErrorReported.compareAndSet(false, true)) {
         long latencyNanos = System.nanoTime() - this.messageStartTimeNanos;
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
@@ -24,6 +24,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.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.RequestThrottlingException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
@@ -253,6 +254,8 @@ private void sendRequest(
         channel = session.getChannel(node, logPrefix);
         if (channel != null) {
           break;
+        } else {
+          recordError(node, new NodeUnavailableException(node));
         }
       }
     }
diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/NodeUnavailableException.java b/core/src/main/java/com/datastax/oss/driver/api/core/NodeUnavailableException.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.api.core;
+
+import com.datastax.oss.driver.api.core.metadata.Node;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import java.util.Objects;
+
+/**
+ * Indicates that a {@link Node} was selected in a query plan, but it had no connection available.
+ *
+ * <p>A common reason to encounter this error is when the configured number of connections per node
+ * and requests per connection is not high enough to absorb the overall request rate. This can be
+ * mitigated by tuning the following options:
+ *
+ * <ul>
+ *   <li>{@code advanced.connection.pool.local.size};
+ *   <li>{@code advanced.connection.pool.remote.size};
+ *   <li>{@code advanced.connection.max-requests-per-connection}.
+ * </ul>
+ *
+ * See {@code reference.conf} for more details.
+ *
+ * <p>Another possibility is when you are trying to direct a request {@linkplain
+ * com.datastax.oss.driver.api.core.cql.Statement#setNode(Node) to a particular node}, but that node
+ * has no connections available.
+ */
+public class NodeUnavailableException extends DriverException {
+
+  private final Node node;
+
+  public NodeUnavailableException(Node node) {
+    super("No connection was available to " + node, null, null, true);
+    this.node = Objects.requireNonNull(node);
+  }
+
+  @NonNull
+  public Node getNode() {
+    return node;
+  }
+
+  @Override
+  @NonNull
+  public DriverException copy() {
+    return new NodeUnavailableException(node);
+  }
+}
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
@@ -18,6 +18,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.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.ProtocolVersion;
 import com.datastax.oss.driver.api.core.RequestThrottlingException;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
@@ -188,6 +189,8 @@ private void sendRequest(PrepareRequest request, Node node, int retryCount) {
         channel = session.getChannel(node, logPrefix);
         if (channel != null) {
           break;
+        } else {
+          recordError(node, new NodeUnavailableException(node));
         }
       }
     }
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
@@ -19,6 +19,7 @@
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.DriverException;
 import com.datastax.oss.driver.api.core.DriverTimeoutException;
+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;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
@@ -251,6 +252,8 @@ private void sendRequest(
         channel = session.getChannel(node, logPrefix);
         if (channel != null) {
           break;
+        } else {
+          recordError(node, new NodeUnavailableException(node));
         }
       }
     }
diff --git a/core/src/test/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousCqlRequestHandlerNodeTargetingTest.java b/core/src/test/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousCqlRequestHandlerNodeTargetingTest.java
@@ -25,13 +25,17 @@
 import com.datastax.dse.driver.DseTestFixtures;
 import com.datastax.dse.driver.api.core.DseProtocolVersion;
 import com.datastax.dse.driver.api.core.cql.continuous.ContinuousAsyncResultSet;
-import com.datastax.oss.driver.api.core.NoNodeAvailableException;
+import com.datastax.oss.driver.api.core.AllNodesFailedException;
+import com.datastax.oss.driver.api.core.NodeUnavailableException;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.session.Request;
 import com.datastax.oss.driver.api.core.session.Session;
 import com.datastax.oss.driver.internal.core.cql.RequestHandlerTestHarness;
 import com.datastax.oss.driver.internal.core.metadata.LoadBalancingPolicyWrapper;
 import com.tngtech.java.junit.dataprovider.UseDataProvider;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletionStage;
 import org.junit.Test;
 import org.mockito.InOrder;
@@ -67,7 +71,12 @@ public void should_fail_if_targeted_node_not_available(DseProtocolVersion versio
       assertThatStage(resultSetFuture)
           .isFailed(
               error -> {
-                assertThat(error).isInstanceOf(NoNodeAvailableException.class);
+                assertThat(error).isInstanceOf(AllNodesFailedException.class);
+                Map<Node, List<Throwable>> errors =
+                    ((AllNodesFailedException) error).getAllErrors();
+                assertThat(errors).hasSize(1);
+                List<Throwable> nodeErrors = errors.values().iterator().next();
+                assertThat(nodeErrors).singleElement().isInstanceOf(NodeUnavailableException.class);
                 invocations
                     .verify(loadBalancingPolicy, never())
                     .newQueryPlan(any(Request.class), anyString(), any(Session.class));
diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandlerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/cql/CqlPrepareHandlerTest.java
@@ -25,6 +25,8 @@
 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.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.ColumnDefinitions;
@@ -47,6 +49,7 @@
 import com.datastax.oss.protocol.internal.response.result.RowsMetadata;
 import com.datastax.oss.protocol.internal.util.Bytes;
 import java.nio.ByteBuffer;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletionStage;
 import org.junit.Before;
@@ -228,6 +231,39 @@ public void should_not_retry_initial_prepare_if_unrecoverable_error() {
     }
   }
 
+  @Test
+  public void should_fail_if_nodes_unavailable() {
+    RequestHandlerTestHarness.Builder harnessBuilder = RequestHandlerTestHarness.builder();
+    try (RequestHandlerTestHarness harness =
+        harnessBuilder.withEmptyPool(node1).withEmptyPool(node2).build()) {
+      CompletionStage<PreparedStatement> prepareFuture =
+          new CqlPrepareHandler(PREPARE_REQUEST, harness.getSession(), harness.getContext(), "test")
+              .handle();
+      assertThatStage(prepareFuture)
+          .isFailed(
+              error -> {
+                assertThat(error).isInstanceOf(AllNodesFailedException.class);
+                Map<Node, List<Throwable>> allErrors =
+                    ((AllNodesFailedException) error).getAllErrors();
+                assertThat(allErrors).hasSize(2);
+                assertThat(allErrors)
+                    .hasEntrySatisfying(
+                        node1,
+                        nodeErrors ->
+                            assertThat(nodeErrors)
+                                .singleElement()
+                                .isInstanceOf(NodeUnavailableException.class));
+                assertThat(allErrors)
+                    .hasEntrySatisfying(
+                        node2,
+                        nodeErrors ->
+                            assertThat(nodeErrors)
+                                .singleElement()
+                                .isInstanceOf(NodeUnavailableException.class));
+              });
+    }
+  }
+
   @Test
   public void should_fail_if_retry_policy_ignores_error() {
     RequestHandlerTestHarness.Builder harnessBuilder =
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
@@ -21,9 +21,11 @@
 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.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.cql.AsyncResultSet;
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
@@ -32,6 +34,7 @@
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 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.internal.core.session.RepreparePayload;
 import com.datastax.oss.driver.internal.core.util.concurrent.CapturingTimer.CapturedTimeout;
 import com.datastax.oss.protocol.internal.request.Prepare;
@@ -43,6 +46,8 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -105,6 +110,43 @@ public void should_fail_if_no_node_available() {
     }
   }
 
+  @Test
+  public void should_fail_if_nodes_unavailable() {
+    RequestHandlerTestHarness.Builder harnessBuilder = RequestHandlerTestHarness.builder();
+    try (RequestHandlerTestHarness harness =
+        harnessBuilder.withEmptyPool(node1).withEmptyPool(node2).build()) {
+      CompletionStage<AsyncResultSet> resultSetFuture =
+          new CqlRequestHandler(
+                  UNDEFINED_IDEMPOTENCE_STATEMENT,
+                  harness.getSession(),
+                  harness.getContext(),
+                  "test")
+              .handle();
+      assertThatStage(resultSetFuture)
+          .isFailed(
+              error -> {
+                assertThat(error).isInstanceOf(AllNodesFailedException.class);
+                Map<Node, List<Throwable>> allErrors =
+                    ((AllNodesFailedException) error).getAllErrors();
+                assertThat(allErrors).hasSize(2);
+                assertThat(allErrors)
+                    .hasEntrySatisfying(
+                        node1,
+                        nodeErrors ->
+                            assertThat(nodeErrors)
+                                .singleElement()
+                                .isInstanceOf(NodeUnavailableException.class));
+                assertThat(allErrors)
+                    .hasEntrySatisfying(
+                        node2,
+                        nodeErrors ->
+                            assertThat(nodeErrors)
+                                .singleElement()
+                                .isInstanceOf(NodeUnavailableException.class));
+              });
+    }
+  }
+
   @Test
   public void should_time_out_if_first_node_takes_too_long_to_respond() throws Exception {
     RequestHandlerTestHarness.Builder harnessBuilder = RequestHandlerTestHarness.builder();
diff --git a/upgrade_guide/README.md b/upgrade_guide/README.md
@@ -1,5 +1,14 @@
 ## Upgrade guide
 
+### 4.14.0
+
+#### AllNodesFailedException instead of NoNodeAvailableException in certain cases
+
+[JAVA-2959](https://datastax-oss.atlassian.net/browse/JAVA-2959) changed the behavior for when a
+request cannot be executed because all nodes tried were busy. Previously you would get back a
+`NoNodeAvailableException` but you will now get back an `AllNodesFailedException` where the
+`getAllErrors` map contains a `NodeUnavailableException` for that node.
+
 ### 4.13.0
 
 #### Enhanced support for GraalVM native images 

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,7 @@`
`24`	`24`	`import com.datastax.oss.driver.api.core.AllNodesFailedException;`
`25`	`25`	`import com.datastax.oss.driver.api.core.CqlIdentifier;`
`26`	`26`	`import com.datastax.oss.driver.api.core.DriverTimeoutException;`
	`27`	`+import com.datastax.oss.driver.api.core.NodeUnavailableException;`
`27`	`28`	`import com.datastax.oss.driver.api.core.ProtocolVersion;`
`28`	`29`	`import com.datastax.oss.driver.api.core.RequestThrottlingException;`
`29`	`30`	`import com.datastax.oss.driver.api.core.config.DefaultDriverOption;`
`@@ -348,6 +349,8 @@ private void sendRequest(`
`348`	`349`	`channel = session.getChannel(node, logPrefix);`
`349`	`350`	`if (channel != null) {`
`350`	`351`	`break;`
	`352`	`+ } else {`
	`353`	`+ recordError(node, new NodeUnavailableException(node));`
`351`	`354`	`}`
`352`	`355`	`}`
`353`	`356`	`}`
`@@ -455,6 +458,10 @@ CompletableFuture<ResultSetT> getPendingResult() {`
`455`	`458`	`}`
`456`	`459`	`}`
`457`	`460`
	`461`	`+ private void recordError(@NonNull Node node, @NonNull Throwable error) {`
	`462`	`+ errors.add(new AbstractMap.SimpleEntry<>(node, error));`
	`463`	`+ }`
	`464`	`+`
`458`	`465`	`/**`
`459`	`466`	`* Handles the interaction with a single node in the query plan.`
`460`	`467`	`*`
`@@ -1433,10 +1440,6 @@ private void reenableAutoReadIfNeeded() {`
`1433`	`1440`
`1434`	`1441`	`// ERROR HANDLING`
`1435`	`1442`
`1436`		`- private void recordError(@NonNull Node node, @NonNull Throwable error) {`
`1437`		`- errors.add(new AbstractMap.SimpleEntry<>(node, error));`
`1438`		`- }`
`1439`		`-`
`1440`	`1443`	`private void trackNodeError(@NonNull Node node, @NonNull Throwable error) {`
`1441`	`1444`	`if (nodeErrorReported.compareAndSet(false, true)) {`
`1442`	`1445`	`long latencyNanos = System.nanoTime() - this.messageStartTimeNanos;`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`import com.datastax.oss.driver.api.core.AllNodesFailedException;`
`19`	`19`	`import com.datastax.oss.driver.api.core.CqlIdentifier;`
`20`	`20`	`import com.datastax.oss.driver.api.core.DriverTimeoutException;`
	`21`	`+import com.datastax.oss.driver.api.core.NodeUnavailableException;`
`21`	`22`	`import com.datastax.oss.driver.api.core.ProtocolVersion;`
`22`	`23`	`import com.datastax.oss.driver.api.core.RequestThrottlingException;`
`23`	`24`	`import com.datastax.oss.driver.api.core.config.DefaultDriverOption;`
`@@ -188,6 +189,8 @@ private void sendRequest(PrepareRequest request, Node node, int retryCount) {`
`188`	`189`	`channel = session.getChannel(node, logPrefix);`
`189`	`190`	`if (channel != null) {`
`190`	`191`	`break;`
	`192`	`+ } else {`
	`193`	`+ recordError(node, new NodeUnavailableException(node));`
`191`	`194`	`}`
`192`	`195`	`}`
`193`	`196`	`}`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@`
`19`	`19`	`import com.datastax.oss.driver.api.core.CqlIdentifier;`
`20`	`20`	`import com.datastax.oss.driver.api.core.DriverException;`
`21`	`21`	`import com.datastax.oss.driver.api.core.DriverTimeoutException;`
	`22`	`+import com.datastax.oss.driver.api.core.NodeUnavailableException;`
`22`	`23`	`import com.datastax.oss.driver.api.core.RequestThrottlingException;`
`23`	`24`	`import com.datastax.oss.driver.api.core.config.DefaultDriverOption;`
`24`	`25`	`import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;`
`@@ -251,6 +252,8 @@ private void sendRequest(`
`251`	`252`	`channel = session.getChannel(node, logPrefix);`
`252`	`253`	`if (channel != null) {`
`253`	`254`	`break;`
	`255`	`+ } else {`
	`256`	`+ recordError(node, new NodeUnavailableException(node));`
`254`	`257`	`}`
`255`	`258`	`}`
`256`	`259`	`}`