fix deadlock in testStreamAckDeadlineUpdate

Fixes #1577. The detailed description of the bug is available in the
issue.

This commit fixes the bug by ensuring that the Future received by
MessageReceiver cannot be completed before a callback is added to it.

Author: pongad

google-cloud-examples/src/main/java/com/google/cloud/examples/pubsub/snippets/CreateSubscriptionAndPullMessages.java

@@ -19,9 +19,8 @@
 import com.google.cloud.pubsub.spi.v1.MessageReceiver;
 import com.google.cloud.pubsub.spi.v1.Subscriber;
 import com.google.cloud.pubsub.spi.v1.SubscriberClient;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.PushConfig;
 import com.google.pubsub.v1.SubscriptionName;
@@ -44,9 +43,10 @@ public static void main(String... args) throws Exception {
     MessageReceiver receiver =
         new MessageReceiver() {
           @Override
-          public ListenableFuture<MessageReceiver.AckReply> receiveMessage(PubsubMessage message) {
+          public void receiveMessage(
+              PubsubMessage message, SettableFuture<MessageReceiver.AckReply> response) {
             System.out.println("got message: " + message.getData().toStringUtf8());
-            return Futures.immediateFuture(MessageReceiver.AckReply.ACK);
+            response.set(MessageReceiver.AckReply.ACK);
           }
         };
     Subscriber subscriber = null;

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java

@@ -25,6 +25,7 @@
 import com.google.common.primitives.Ints;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.ReceivedMessage;
 import java.util.ArrayList;
@@ -278,11 +279,13 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r
     for (ReceivedMessage userMessage : responseMessages) {
       final PubsubMessage message = userMessage.getMessage();
       final AckHandler ackHandler = acksIterator.next();
+      final SettableFuture<AckReply> response = SettableFuture.create();
+      Futures.addCallback(response, ackHandler);
       executor.submit(
           new Runnable() {
             @Override
             public void run() {
-              Futures.addCallback(receiver.receiveMessage(message), ackHandler);
+              receiver.receiveMessage(message, response);
             }
           });
     }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageReceiver.java

@@ -16,7 +16,7 @@
 
 package com.google.cloud.pubsub.spi.v1;
 
-import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 
 /** This interface can be implemented by users of {@link Subscriber} to receive messages. */
@@ -34,11 +34,10 @@ enum AckReply {
      */
     NACK
   }
-  
+
   /**
-   * Called when a message is received by the subscriber.
-   *
-   * @return A future that signals when a message has been processed.
+   * Called when a message is received by the subscriber. The implementation must arrange for {@code
+   * reponse} to be set after processing the {@code message}.
    */
-  ListenableFuture<AckReply> receiveMessage(PubsubMessage message);
+  void receiveMessage(PubsubMessage message, SettableFuture<AckReply> response);
 }

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/Subscriber.java

@@ -81,9 +81,9 @@
  * <pre><code>
  * MessageReceiver receiver = new MessageReceiver() {
  *   &#64;Override
- *   public ListenableFuture&lt;AckReply&gt; receiveMessage(PubsubMessage message) {
+ *   public void receiveMessage(PubsubMessage message, SettableFuture&lt;AckReply&gt; response) {
  *     // ... process message ...
- *     return Futures.immediateFuture(AckReply.ACK);
+ *     return response.set(AckReply.ACK);
  *   }
  * }
  *

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java

@@ -28,7 +28,6 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.PullResponse;
@@ -115,23 +114,20 @@ void waitForExpectedMessages() throws InterruptedException {
     }
 
     @Override
-    public ListenableFuture<AckReply> receiveMessage(PubsubMessage message) {
+    public void receiveMessage(PubsubMessage message, SettableFuture<AckReply> response) {
       if (messageCountLatch.isPresent()) {
         messageCountLatch.get().countDown();
       }
-      SettableFuture<AckReply> reply = SettableFuture.create();
 
       if (explicitAckReplies) {
         try {
-          outstandingMessageReplies.put(reply);
+          outstandingMessageReplies.put(response);
         } catch (InterruptedException e) {
           throw new IllegalStateException(e);
         }
       } else {
-        replyTo(reply);
+        replyTo(response);
       }
-
-      return reply;
     }
 
     public void replyNextOutstandingMessage() {