fix: More robust STREAM_RST logic

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/util/Errors.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * 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
+ *
+ *     https://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.google.cloud.bigquery.storage.util;
+
+import io.grpc.Status;
+
+/** Static utility methods for working with Errors returned from the service. */
+public class Errors {
+  private Errors() {};
+
+  /**
+   * Returns true iff the Status indicates and internal error that is retryable.
+   *
+   * <p>Generally, internal errors are not considered retryable, however there are certain transient
+   * network issues that appear as internal but are in fact retryable.
+   */
+  public static boolean isRetryableInternalStatus(Status status) {
+    String description = status.getDescription();
+    return status.getCode() == Status.Code.INTERNAL
+        && description != null
+        && (description.contains("Received unexpected EOS on DATA frame from server")
+            || description.contains(" Rst ")
+            || description.contains("RST_STREAM")
+            || description.contains("Connection closed with unknown cause")
+            || description.contains("HTTP/2 error code: INTERNAL_ERROR"));
+  }
+}

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/stub/readrows/ApiResultRetryAlgorithm.java

@@ -20,6 +20,7 @@
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.TimedAttemptSettings;
 import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.bigquery.storage.util.Errors;
 import io.grpc.Status;
 import org.threeten.bp.Duration;
 
@@ -29,19 +30,12 @@ public class ApiResultRetryAlgorithm<ResponseT> implements ResultRetryAlgorithm<
   // Duration to sleep on if the error is DEADLINE_EXCEEDED.
   public static final Duration DEADLINE_SLEEP_DURATION = Duration.ofMillis(1);
 
-  private boolean isRetryableStatus(Status status) {
-    return status.getCode() == Status.Code.INTERNAL
-        && status.getDescription() != null
-        && (status.getDescription().contains("Received unexpected EOS on DATA frame from server")
-            || status.getDescription().contains("Received Rst Stream"));
-  }
-
   @Override
   public TimedAttemptSettings createNextAttempt(
       Throwable prevThrowable, ResponseT prevResponse, TimedAttemptSettings prevSettings) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return TimedAttemptSettings.newBuilder()
             .setGlobalSettings(prevSettings.getGlobalSettings())
             .setRetryDelay(prevSettings.getRetryDelay())
@@ -59,7 +53,7 @@ public TimedAttemptSettings createNextAttempt(
   public boolean shouldRetry(Throwable prevThrowable, ResponseT prevResponse) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return true;
       }
     }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta1/stub/readrows/ApiResultRetryAlgorithm.java

@@ -20,6 +20,7 @@
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.TimedAttemptSettings;
 import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.bigquery.storage.util.Errors;
 import io.grpc.Status;
 import org.threeten.bp.Duration;
 
@@ -29,19 +30,12 @@ public class ApiResultRetryAlgorithm<ResponseT> implements ResultRetryAlgorithm<
   // Duration to sleep on if the error is DEADLINE_EXCEEDED.
   public static final Duration DEADLINE_SLEEP_DURATION = Duration.ofMillis(1);
 
-  private boolean isRetryableStatus(Status status) {
-    return status.getCode() == Status.Code.INTERNAL
-        && status.getDescription() != null
-        && (status.getDescription().contains("Received unexpected EOS on DATA frame from server")
-            || status.getDescription().contains("Received Rst Stream"));
-  }
-
   @Override
   public TimedAttemptSettings createNextAttempt(
       Throwable prevThrowable, ResponseT prevResponse, TimedAttemptSettings prevSettings) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return TimedAttemptSettings.newBuilder()
             .setGlobalSettings(prevSettings.getGlobalSettings())
             .setRetryDelay(prevSettings.getRetryDelay())
@@ -59,7 +53,7 @@ public TimedAttemptSettings createNextAttempt(
   public boolean shouldRetry(Throwable prevThrowable, ResponseT prevResponse) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return true;
       }
     }

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1beta2/stub/readrows/ApiResultRetryAlgorithm.java

@@ -20,6 +20,7 @@
 import com.google.api.gax.retrying.ResultRetryAlgorithm;
 import com.google.api.gax.retrying.TimedAttemptSettings;
 import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.bigquery.storage.util.Errors;
 import io.grpc.Status;
 import org.threeten.bp.Duration;
 
@@ -29,19 +30,12 @@ public class ApiResultRetryAlgorithm<ResponseT> implements ResultRetryAlgorithm<
   // Duration to sleep on if the error is DEADLINE_EXCEEDED.
   public static final Duration DEADLINE_SLEEP_DURATION = Duration.ofMillis(1);
 
-  private boolean isRetryableStatus(Status status) {
-    return status.getCode() == Status.Code.INTERNAL
-        && status.getDescription() != null
-        && (status.getDescription().contains("Received unexpected EOS on DATA frame from server")
-            || status.getDescription().contains("Received Rst Stream"));
-  }
-
   @Override
   public TimedAttemptSettings createNextAttempt(
       Throwable prevThrowable, ResponseT prevResponse, TimedAttemptSettings prevSettings) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return TimedAttemptSettings.newBuilder()
             .setGlobalSettings(prevSettings.getGlobalSettings())
             .setRetryDelay(prevSettings.getRetryDelay())
@@ -59,7 +53,7 @@ public TimedAttemptSettings createNextAttempt(
   public boolean shouldRetry(Throwable prevThrowable, ResponseT prevResponse) {
     if (prevThrowable != null) {
       Status status = Status.fromThrowable(prevThrowable);
-      if (isRetryableStatus(status)) {
+      if (Errors.isRetryableInternalStatus(status)) {
         return true;
       }
     }

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/util/ErrorsTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * 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
+ *
+ *     https://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.google.cloud.bigquery.storage.util;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import io.grpc.Status;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class ErrorsTest {
+
+  @Test
+  public void testRetryableInternalForRstErrors() {
+    assertTrue(
+        Errors.isRetryableInternalStatus(
+            Status.INTERNAL.withDescription(
+                "HTTP/2 error code: INTERNAL_ERROR\nReceived Rst stream")));
+    assertTrue(
+        Errors.isRetryableInternalStatus(
+            Status.INTERNAL.withDescription(
+                "RST_STREAM closed stream. HTTP/2 error code: INTERNAL_ERROR")));
+  }
+
+  @Test
+  public void testNonRetryableInternalError() {
+    assertFalse(Errors.isRetryableInternalStatus(Status.INTERNAL));
+    assertFalse(Errors.isRetryableInternalStatus(Status.INTERNAL.withDescription("Server error.")));
+  }
+
+  @Test
+  public void testNonRetryableOtherError() {
+    assertFalse(
+        Errors.isRetryableInternalStatus(
+            Status.DATA_LOSS.withDescription(
+                "RST_STREAM closed stream. HTTP/2 error code: INTERNAL_ERROR")));
+  }
+}