fix race in TestReceiver (#1558)

pongad · web-flow · commit c68968b92233 · 2017-01-24T14:29:42.000+11:00
* fix race in TestReceiver

If TestReceiver is set so that messages must be explicitly acked,
test code acks a message by calling replyNextOutstandingMessage.
There is a race between calling the function and the message being
delivered.

Previously, if the function gets called before the message is delivered,
the test fails since polling an empty deque returns a null pointer.
This commit makes us wait until a message becomes available instead.
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/SubscriberImplTest.java
@@ -45,10 +45,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Deque;
 import java.util.List;
-import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
 import org.joda.time.Duration;
 import org.junit.After;
 import org.junit.Before;
@@ -87,8 +86,8 @@ public static Collection<Object[]> data() {
   private TestReceiver testReceiver;
 
   static class TestReceiver implements MessageReceiver {
-    private final Deque<SettableFuture<AckReply>> outstandingMessageReplies =
-        new ConcurrentLinkedDeque<>();
+    private final LinkedBlockingQueue<SettableFuture<AckReply>> outstandingMessageReplies =
+        new LinkedBlockingQueue<>();
     private AckReply ackReply = AckReply.ACK;
     private Optional<CountDownLatch> messageCountLatch = Optional.absent();
     private Optional<Throwable> error = Optional.absent();
@@ -124,34 +123,40 @@ public ListenableFuture<AckReply> receiveMessage(PubsubMessage message) {
       SettableFuture<AckReply> reply = SettableFuture.create();
 
       if (explicitAckReplies) {
-        outstandingMessageReplies.add(reply);
-      } else {
-        if (error.isPresent()) {
-          reply.setException(error.get());
-        } else {
-          reply.set(ackReply);
+        try {
+          outstandingMessageReplies.put(reply);
+        } catch (InterruptedException e) {
+          throw new IllegalStateException(e);
         }
+      } else {
+        replyTo(reply);
       }
 
       return reply;
     }
 
     public void replyNextOutstandingMessage() {
       Preconditions.checkState(explicitAckReplies);
-
-      SettableFuture<AckReply> reply = outstandingMessageReplies.poll();
-      if (error.isPresent()) {
-        reply.setException(error.get());
-      } else {
-        reply.set(ackReply);
+      try {
+        replyTo(outstandingMessageReplies.take());
+      } catch (InterruptedException e) {
+        throw new IllegalStateException(e);
       }
     }
 
     public void replyAllOutstandingMessage() {
       Preconditions.checkState(explicitAckReplies);
+      SettableFuture<AckReply> reply;
+      while ((reply = outstandingMessageReplies.poll()) != null) {
+        replyTo(reply);
+      }
+    }
 
-      while (!outstandingMessageReplies.isEmpty()) {
-        replyNextOutstandingMessage();
+    private void replyTo(SettableFuture<AckReply> reply) {
+      if (error.isPresent()) {
+        reply.setException(error.get());
+      } else {
+        reply.set(ackReply);
       }
     }
   }
