pubsub: translate exception to ApiException (#2531)

Fixes #2512.

Author: pongad

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java

@@ -23,7 +23,9 @@
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.core.InstantiatingExecutorProvider;
 import com.google.api.gax.grpc.ChannelProvider;
+import com.google.api.gax.grpc.GrpcApiExceptionFactory;
 import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.rpc.ApiException;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.annotations.VisibleForTesting;
@@ -367,9 +369,12 @@ public void onFailure(Throwable t) {
                 || System.currentTimeMillis() + nextBackoffDelay
                     > outstandingBatch.creationTime + retrySettings.getTotalTimeout().toMillis()) {
               try {
+                ApiException gaxException =
+                    GrpcApiExceptionFactory.createException(
+                        t, Status.fromThrowable(t).getCode(), false);
                 for (OutstandingPublish outstandingPublish :
                     outstandingBatch.outstandingPublishes) {
-                  outstandingPublish.publishResult.setException(t);
+                  outstandingPublish.publishResult.setException(gaxException);
                 }
               } finally {
                 messagesWaiter.incrementPendingMessages(-outstandingBatch.size());

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

@@ -20,6 +20,8 @@
 import com.google.api.core.ApiClock;
 import com.google.api.gax.batching.FlowController;
 import com.google.api.gax.core.Distribution;
+import com.google.api.gax.grpc.GrpcApiExceptionFactory;
+import com.google.api.gax.rpc.ApiException;
 import com.google.cloud.pubsub.v1.MessageDispatcher.AckProcessor;
 import com.google.cloud.pubsub.v1.MessageDispatcher.PendingModifyAckDeadline;
 import com.google.common.annotations.VisibleForTesting;
@@ -230,8 +232,11 @@ public void onFailure(Throwable cause) {
               return;
             }
             if (!StatusUtil.isRetryable(cause)) {
-              logger.log(Level.SEVERE, "terminated streaming with exception", cause);
-              notifyFailed(cause);
+              ApiException gaxException =
+                  GrpcApiExceptionFactory.createException(
+                      cause, Status.fromThrowable(cause).getCode(), false);
+              logger.log(Level.SEVERE, "terminated streaming with exception", gaxException);
+              notifyFailed(gaxException);
               return;
             }
             logger.log(Level.FINE, "stream closed with retryable exception; will reconnect", cause);

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java

@@ -23,6 +23,8 @@
 import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.core.InstantiatingExecutorProvider;
 import com.google.api.gax.grpc.FixedChannelProvider;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
 import com.google.cloud.pubsub.v1.FakeSubscriberServiceImpl.ModifyAckDeadline;
 import com.google.cloud.pubsub.v1.Subscriber.Builder;
 import com.google.common.base.Function;
@@ -38,7 +40,6 @@
 import io.grpc.Server;
 import io.grpc.Status;
 import io.grpc.StatusException;
-import io.grpc.StatusRuntimeException;
 import io.grpc.inprocess.InProcessChannelBuilder;
 import io.grpc.inprocess.InProcessServerBuilder;
 import java.util.ArrayList;
@@ -517,9 +518,15 @@ public void testFailedChannel_fatalError_subscriberFails() throws Exception {
     } finally {
       // The subscriber must finish with an state error because its FAILED status.
       assertEquals(Subscriber.State.FAILED, subscriber.state());
-      assertEquals(
-          Status.INVALID_ARGUMENT,
-          ((StatusRuntimeException) subscriber.failureCause()).getStatus());
+
+      Throwable t = subscriber.failureCause();
+      assertTrue(t instanceof ApiException);
+
+      ApiException ex = (ApiException) (t);
+      assertTrue(ex.getStatusCode() instanceof GrpcStatusCode);
+
+      GrpcStatusCode grpcCode = (GrpcStatusCode) ex.getStatusCode();
+      assertEquals(Status.Code.INVALID_ARGUMENT, grpcCode.getCode());
     }
   }