JAVA-1944: Surface Read and WriteFailureException to RetryPolicy (#1084)

Author: tolbertam

changelog/README.md

@@ -19,6 +19,7 @@
 - [improvement] JAVA-1771: Send driver name and version in STARTUP message.
 - [improvement] JAVA-1388: Add dynamic port discovery for system.peers\_v2.
 - [documentation] JAVA-1810: Note which setters are not propagated to PreparedStatement.
+- [bug] JAVA-1944: Surface Read and WriteFailureException to RetryPolicy.
 
 Merged from 3.5.x:
 

driver-core/src/main/java/com/datastax/driver/core/RequestHandler.java

@@ -25,9 +25,11 @@
 import com.datastax.driver.core.exceptions.NoHostAvailableException;
 import com.datastax.driver.core.exceptions.OperationTimedOutException;
 import com.datastax.driver.core.exceptions.OverloadedException;
+import com.datastax.driver.core.exceptions.ReadFailureException;
 import com.datastax.driver.core.exceptions.ReadTimeoutException;
 import com.datastax.driver.core.exceptions.ServerError;
 import com.datastax.driver.core.exceptions.UnavailableException;
+import com.datastax.driver.core.exceptions.WriteFailureException;
 import com.datastax.driver.core.exceptions.WriteTimeoutException;
 import com.datastax.driver.core.policies.RetryPolicy;
 import com.datastax.driver.core.policies.RetryPolicy.RetryDecision.Type;
@@ -761,6 +763,23 @@ public void onSet(
                 write(connection, prepareAndRetry(toPrepare.getQueryString()));
                 // we're done for now, the prepareAndRetry callback will handle the rest
                 return;
+              case READ_FAILURE:
+                assert exceptionToReport instanceof ReadFailureException;
+                connection.release();
+                retry =
+                    computeRetryDecisionOnRequestError((ReadFailureException) exceptionToReport);
+                break;
+              case WRITE_FAILURE:
+                assert exceptionToReport instanceof WriteFailureException;
+                connection.release();
+                if (statement.isIdempotentWithDefault(
+                    manager.cluster.getConfiguration().getQueryOptions())) {
+                  retry =
+                      computeRetryDecisionOnRequestError((WriteFailureException) exceptionToReport);
+                } else {
+                  retry = RetryPolicy.RetryDecision.rethrow();
+                }
+                break;
               default:
                 connection.release();
                 if (metricsEnabled()) metrics().getErrorMetrics().getOthers().inc();

driver-core/src/main/java/com/datastax/driver/core/policies/DefaultRetryPolicy.java

@@ -20,6 +20,8 @@
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.WriteType;
 import com.datastax.driver.core.exceptions.DriverException;
+import com.datastax.driver.core.exceptions.ReadFailureException;
+import com.datastax.driver.core.exceptions.WriteFailureException;
 
 /**
  * The default retry policy.
@@ -32,7 +34,8 @@
  *   <li>On a write timeout, retries once on the same host if we timeout while writing the
  *       distributed log used by batch statements.
  *   <li>On an unavailable exception, retries once on the next host.
- *   <li>On a request error, such as a client timeout, the query is retried on the next host.
+ *   <li>On a request error, such as a client timeout, the query is retried on the next host. Do not
+ *       retry on read or write failures.
  * </ul>
  *
  * <p>This retry policy is conservative in that it will never retry with a different consistency
@@ -137,6 +140,11 @@ public RetryDecision onUnavailable(
   @Override
   public RetryDecision onRequestError(
       Statement statement, ConsistencyLevel cl, DriverException e, int nbRetry) {
+    // do not retry these by default as they generally indicate a data problem or
+    // other issue that is unlikely to be resolved by a retry.
+    if (e instanceof WriteFailureException || e instanceof ReadFailureException) {
+      return RetryDecision.rethrow();
+    }
     return RetryDecision.tryNextHost(cl);
   }
 

driver-core/src/main/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicy.java

@@ -20,6 +20,8 @@
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.WriteType;
 import com.datastax.driver.core.exceptions.DriverException;
+import com.datastax.driver.core.exceptions.ReadFailureException;
+import com.datastax.driver.core.exceptions.WriteFailureException;
 
 /**
  * A retry policy that sometimes retries with a lower consistency level than the one initially
@@ -215,6 +217,11 @@ public RetryDecision onUnavailable(
   @Override
   public RetryDecision onRequestError(
       Statement statement, ConsistencyLevel cl, DriverException e, int nbRetry) {
+    // do not retry these by default as they generally indicate a data problem or
+    // other issue that is unlikely to be resolved by a retry.
+    if (e instanceof WriteFailureException || e instanceof ReadFailureException) {
+      return RetryDecision.rethrow();
+    }
     return RetryDecision.tryNextHost(cl);
   }
 

driver-core/src/main/java/com/datastax/driver/core/policies/RetryPolicy.java

@@ -267,8 +267,8 @@ RetryDecision onUnavailable(
    *   <li>On a client timeout, while waiting for the server response (see {@link
    *       SocketOptions#getReadTimeoutMillis()});
    *   <li>On a connection error (socket closed, etc.);
-   *   <li>When the contacted host replies with an {@code OVERLOADED} error or a {@code
-   *       SERVER_ERROR}.
+   *   <li>When the contacted host replies with an {@code OVERLOADED} error, {@code SERVER_ERROR},
+   *       {@code READ_FAILURE} or {@code WRITE_FAILURE}.
    * </ol>
    *
    * <p>Note that when such an error occurs, there is no guarantee that the mutation has been

driver-core/src/test/java/com/datastax/driver/core/policies/DefaultRetryPolicyIntegrationTest.java

@@ -20,8 +20,10 @@
 import static org.scassandra.http.client.Consistency.LOCAL_SERIAL;
 import static org.scassandra.http.client.PrimingRequest.then;
 import static org.scassandra.http.client.Result.closed_connection;
+import static org.scassandra.http.client.Result.read_failure;
 import static org.scassandra.http.client.Result.read_request_timeout;
 import static org.scassandra.http.client.Result.unavailable;
+import static org.scassandra.http.client.Result.write_failure;
 import static org.scassandra.http.client.Result.write_request_timeout;
 
 import com.datastax.driver.core.Cluster;
@@ -32,9 +34,11 @@
 import com.datastax.driver.core.exceptions.DriverException;
 import com.datastax.driver.core.exceptions.NoHostAvailableException;
 import com.datastax.driver.core.exceptions.OperationTimedOutException;
+import com.datastax.driver.core.exceptions.ReadFailureException;
 import com.datastax.driver.core.exceptions.ReadTimeoutException;
 import com.datastax.driver.core.exceptions.TransportException;
 import com.datastax.driver.core.exceptions.UnavailableException;
+import com.datastax.driver.core.exceptions.WriteFailureException;
 import com.datastax.driver.core.exceptions.WriteTimeoutException;
 import java.util.Collections;
 import org.assertj.core.api.Fail;
@@ -261,6 +265,46 @@ public void should_try_next_host_on_server_side_error(
     assertQueried(3, 1);
   }
 
+  @Test(groups = "short")
+  public void should_rethrow_on_read_failure() {
+    simulateError(1, read_failure);
+
+    try {
+      query();
+      fail("expected a ReadFailureException");
+    } catch (DriverException e) {
+      assertThat(e).isInstanceOf(ReadFailureException.class);
+    }
+
+    assertOnRequestErrorWasCalled(1, ReadFailureException.class);
+    assertThat(errors.getOthers().getCount()).isEqualTo(1);
+    assertThat(errors.getRetries().getCount()).isEqualTo(0);
+    assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(0);
+    assertQueried(1, 1);
+    assertQueried(2, 0);
+    assertQueried(3, 0);
+  }
+
+  @Test(groups = "short")
+  public void should_rethrow_on_write_failure() {
+    simulateError(1, write_failure);
+
+    try {
+      query();
+      fail("expected a WriteFailureException");
+    } catch (DriverException e) {
+      assertThat(e).isInstanceOf(WriteFailureException.class);
+    }
+
+    assertOnRequestErrorWasCalled(1, WriteFailureException.class);
+    assertThat(errors.getOthers().getCount()).isEqualTo(1);
+    assertThat(errors.getRetries().getCount()).isEqualTo(0);
+    assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(0);
+    assertQueried(1, 1);
+    assertQueried(2, 0);
+    assertQueried(3, 0);
+  }
+
   @Test(groups = "short", dataProvider = "connectionErrors")
   public void should_try_next_host_on_connection_error(ClosedConnectionConfig.CloseType closeType) {
     simulateError(1, closed_connection, new ClosedConnectionConfig(closeType));

driver-core/src/test/java/com/datastax/driver/core/policies/DowngradingConsistencyRetryPolicyIntegrationTest.java

@@ -19,8 +19,10 @@
 import static org.scassandra.http.client.Consistency.SERIAL;
 import static org.scassandra.http.client.PrimingRequest.then;
 import static org.scassandra.http.client.Result.closed_connection;
+import static org.scassandra.http.client.Result.read_failure;
 import static org.scassandra.http.client.Result.read_request_timeout;
 import static org.scassandra.http.client.Result.unavailable;
+import static org.scassandra.http.client.Result.write_failure;
 import static org.scassandra.http.client.Result.write_request_timeout;
 import static org.scassandra.http.client.WriteTypePrime.UNLOGGED_BATCH;
 import static org.testng.Assert.fail;
@@ -31,9 +33,11 @@
 import com.datastax.driver.core.exceptions.DriverException;
 import com.datastax.driver.core.exceptions.NoHostAvailableException;
 import com.datastax.driver.core.exceptions.OperationTimedOutException;
+import com.datastax.driver.core.exceptions.ReadFailureException;
 import com.datastax.driver.core.exceptions.ReadTimeoutException;
 import com.datastax.driver.core.exceptions.TransportException;
 import com.datastax.driver.core.exceptions.UnavailableException;
+import com.datastax.driver.core.exceptions.WriteFailureException;
 import com.datastax.driver.core.exceptions.WriteTimeoutException;
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Fail;
@@ -519,4 +523,44 @@ public void should_rethrow_on_unavailable_if_CAS() {
     assertQueried(2, 1);
     assertQueried(3, 0);
   }
+
+  @Test(groups = "short")
+  public void should_rethrow_on_read_failure() {
+    simulateError(1, read_failure);
+
+    try {
+      query();
+      fail("expected a ReadFailureException");
+    } catch (DriverException e) {
+      assertThat(e).isInstanceOf(ReadFailureException.class);
+    }
+
+    assertOnRequestErrorWasCalled(1, ReadFailureException.class);
+    assertThat(errors.getOthers().getCount()).isEqualTo(1);
+    assertThat(errors.getRetries().getCount()).isEqualTo(0);
+    assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(0);
+    assertQueried(1, 1);
+    assertQueried(2, 0);
+    assertQueried(3, 0);
+  }
+
+  @Test(groups = "short")
+  public void should_rethrow_on_write_failure() {
+    simulateError(1, write_failure);
+
+    try {
+      query();
+      fail("expected a WriteFailureException");
+    } catch (DriverException e) {
+      assertThat(e).isInstanceOf(WriteFailureException.class);
+    }
+
+    assertOnRequestErrorWasCalled(1, WriteFailureException.class);
+    assertThat(errors.getOthers().getCount()).isEqualTo(1);
+    assertThat(errors.getRetries().getCount()).isEqualTo(0);
+    assertThat(errors.getRetriesOnOtherErrors().getCount()).isEqualTo(0);
+    assertQueried(1, 1);
+    assertQueried(2, 0);
+    assertQueried(3, 0);
+  }
 }

upgrade_guide/README.md

@@ -10,6 +10,14 @@ versions of the Java driver.
     ConsistencyLevel (which does not include Serial CLs), but as a matter of
     correctness this was updated.
 
+2.  `ReadFailureException` and `WriteFailureException` are now surfaced to
+    `RetryPolicy.onRequestError`. Consider updating custom `RetryPolicy`
+    implementations to account for this. In the general case, we recommend
+    using `RetryDecision.rethrow()`, see [JAVA-1944].
+
+[JAVA-1944]: https://datastax-oss.atlassian.net/browse/JAVA-1944
+
+
 ### 3.5.0
 
 1.  The `DowngradingConsistencyRetryPolicy` is now deprecated, see [JAVA-1752].