[beam] branch master updated: [BEAM-10756] Fix empty pull response to not ack and to not throw IndexOutOfBoundsException

[lcwik]

This is an automated email from the ASF dual-hosted git repository.

lcwik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 85efc38  [BEAM-10756] Fix empty pull response to not ack and to not 
throw IndexOutOfBoundsException
     new ecfc389  Merge pull request #12623 from lukecwik/beam10756
85efc38 is described below

commit 85efc3864960fb9f0553fb4671c71e2058169288
Author: Luke Cwik <lc...@google.com>
AuthorDate: Tue Aug 18 16:05:32 2020 -0700

    [BEAM-10756] Fix empty pull response to not ack and to not throw 
IndexOutOfBoundsException
---
 .../java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java     | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git 
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java
 
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java
index 76d4734..ba52a2e 100644
--- 
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java
+++ 
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java
@@ -253,6 +253,9 @@ public class TestPubsubSignal implements TestRule {
     do {
       try {
         signal = pubsub.pull(DateTime.now().getMillis(), 
signalSubscriptionPath, 1, false);
+        if (signal.isEmpty()) {
+          continue;
+        }
         pubsub.acknowledge(
             signalSubscriptionPath, 
signal.stream().map(IncomingMessage::ackId).collect(toList()));
         break;
@@ -267,7 +270,7 @@ public class TestPubsubSignal implements TestRule {
       }
     } while (DateTime.now().isBefore(endPolling));
 
-    if (signal == null) {
+    if (signal == null || signal.isEmpty()) {
       throw new AssertionError(
           String.format(
               "Did not receive signal on %s in %ss",