fix: Split retryable codes for streams, and retry RESOURCE_EXHAUSTED errors.

Author: manuelmenzella-google

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/ErrorCodes.java

@@ -25,9 +25,16 @@ public final class ErrorCodes {
       ImmutableSet.of(
           Code.DEADLINE_EXCEEDED, Code.ABORTED, Code.INTERNAL, Code.UNAVAILABLE, Code.UNKNOWN);
 
+  public static final ImmutableSet<Code> STREAM_RETRYABLE_CODES =
+      ImmutableSet.<Code>builder().addAll(RETRYABLE_CODES).add(Code.RESOURCE_EXHAUSTED).build();
+
   public static boolean IsRetryable(Code code) {
     return RETRYABLE_CODES.contains(code);
   }
 
+  public static boolean IsRetryableForStreams(Code code) {
+    return STREAM_RETRYABLE_CODES.contains(code);
+  }
+
   private ErrorCodes() {}
 }

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/RetryingConnectionImpl.java

@@ -153,7 +153,7 @@ public final void onError(Throwable t) {
       setPermanentError(t);
       return;
     }
-    if (!ErrorCodes.IsRetryable(statusOr.get().getCode())) {
+    if (!ErrorCodes.IsRetryableForStreams(statusOr.get().getCode())) {
       setPermanentError(statusOr.get().asRuntimeException());
       return;
     }