---

pongad · web-flow · commit 630296359d84 · 2017-10-18T12:16:43.000+11:00
yaml --- r: 8803 b: refs/heads/lesv-patch-1 c: 87dd418 h: refs/heads/master i: 8801: 3b922b0 8799: cd8fd8e
diff --git a/[refs] b/[refs]
@@ -66,7 +66,7 @@ refs/tags/v0.22.0: 18b298fe4bfe8ec2f20b0e0bf7ffdcce5cc3c5fe
 refs/heads/vam-google-patch-1: d0c8fee3a4074d0bf7360ce8c4f7f7223d0ee7b9
 refs/heads/vam-google-patch-CODEOWNERS: 2ac1616e25229e51d08a984708ef1918f91a35ee
 refs/heads/danoscarmike-patch-1: 7342a9916bce4ed00002c7202e2a16c5d46afaea
-refs/heads/lesv-patch-1: 23bb30ea382d237c9eebf9758f89194f1cb0eb75
+refs/heads/lesv-patch-1: 87dd4186a9cd58fc405caf1135330d2e4c24b7c1
 refs/heads/ml-update-branch: 079dd6610017f5c51b9d1938c12d6d55b61513cf
 refs/heads/vkedia-patch-2: 7d8241388a9769a5c069334761b06c7012c878e7
 refs/heads/vkedia-patch-3: 4d128043acaa7db9160faf439d2ca6104e8a88cb
diff --git a/branches/lesv-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/branches/lesv-patch-1/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());
diff --git a/branches/lesv-patch-1/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java b/branches/lesv-patch-1/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);
diff --git a/branches/lesv-patch-1/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java b/branches/lesv-patch-1/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());
     }
   }
 
