You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2020/08/19 15:59:26 UTC
[beam] branch master updated: [BEAM-10756] Fix empty pull response
to not ack and to not throw IndexOutOfBoundsException
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",