fix: retry INTERNAL retriable auth errors

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/ConvertExceptionCallable.java

@@ -77,12 +77,20 @@ protected void onCompleteImpl() {
   private Throwable convertException(Throwable t) {
     // Long lived connections sometimes are disconnected via an RST frame or a goaway. These errors
     // are transient and should be retried.
-    if (isRstStreamError(t) || isGoAway(t)) {
+    if (isRstStreamError(t) || isGoAway(t) || isRetriableAuthError(t)) {
       return new InternalException(t, ((InternalException) t).getStatusCode(), true);
     }
     return t;
   }
 
+  private boolean isRetriableAuthError(Throwable t) {
+    if (t instanceof InternalException && t.getMessage() != null) {
+      String error = t.getMessage();
+      return error.contains("Authentication backend internal server error. Please retry");
+    }
+    return false;
+  }
+
   private boolean isRstStreamError(Throwable t) {
     if (t instanceof InternalException && t.getMessage() != null) {
       String error = t.getMessage().toLowerCase();

google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/ConvertExceptionCallableTest.java

@@ -59,6 +59,32 @@ public void rstStreamExceptionConvertedToRetryableTest() {
     assertTrue(actualException.isRetryable());
   }
 
+  @Test
+  public void retriableAuthExceptionConvertedToRetryableTest() {
+    ApiException originalException =
+        new InternalException(
+            new StatusRuntimeException(
+                Status.INTERNAL.withDescription(
+                    "Authentication backend internal server error. Please retry")),
+            GrpcStatusCode.of(Status.Code.INTERNAL),
+            false);
+    assertFalse(originalException.isRetryable());
+    SettableExceptionCallable<String, String> settableExceptionCallable =
+        new SettableExceptionCallable<>(originalException);
+    ConvertExceptionCallable<String, String> convertStreamExceptionCallable =
+        new ConvertExceptionCallable<>(settableExceptionCallable);
+
+    Throwable actualError = null;
+    try {
+      convertStreamExceptionCallable.all().call("fake-request");
+    } catch (Throwable t) {
+      actualError = t;
+    }
+    assert actualError instanceof InternalException;
+    InternalException actualException = (InternalException) actualError;
+    assertTrue(actualException.isRetryable());
+  }
+
   private static final class SettableExceptionCallable<RequestT, ResponseT>
       extends ServerStreamingCallable<RequestT, ResponseT> {
     private final Throwable throwable;