The Publisher's callback should happen second

sduskis · sduskis · commit 92a7bf4760df · 2019-05-01T14:11:44.000-04:00
The `SequentialExecutor`'s future callback needs to happen before the `Publisher`'s callback, in case the `SequentialExecutor` cancels.
diff --git a/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java b/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
@@ -220,7 +220,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
 
     final String orderingKey = message.getOrderingKey();
     Preconditions.checkState(
-        orderingKey != null && !orderingKey.isEmpty() && !enableMessageOrdering,
+        orderingKey.isEmpty() || enableMessageOrdering,
         "Cannot publish a message with an ordering key when message ordering is not enabled.");
 
     final OutstandingPublish outstandingPublish =
@@ -357,7 +357,7 @@ private void publishOutstandingBatch(final OutstandingBatch outstandingBatch) {
           @Override
           public void onSuccess(PublishResponse result) {
             try {
-              if (result.getMessageIdsCount() != outstandingBatch.size()) {
+              if (result == null || result.getMessageIdsCount() != outstandingBatch.size()) {
                 outstandingBatch.onFailure(
                     new IllegalStateException(
                         String.format(
@@ -387,15 +387,15 @@ public void onFailure(Throwable t) {
       ApiFutures.addCallback(publishCall(outstandingBatch), futureCallback, directExecutor());
     } else {
       // If ordering key is specified, publish the batch using the sequential executor.
-      sequentialExecutor.submit(
-          outstandingBatch.orderingKey,
-          new Callable<ApiFuture<PublishResponse>>() {
-            public ApiFuture<PublishResponse> call() {
-              ApiFuture<PublishResponse> future = publishCall(outstandingBatch);
-              ApiFutures.addCallback(future, futureCallback, directExecutor());
-              return future;
-            }
-          });
+      ApiFuture<PublishResponse> future =
+          sequentialExecutor.submit(
+              outstandingBatch.orderingKey,
+              new Callable<ApiFuture<PublishResponse>>() {
+                public ApiFuture<PublishResponse> call() {
+                  return publishCall(outstandingBatch);
+                }
+              });
+      ApiFutures.addCallback(future, futureCallback, directExecutor());
     }
   }
 
