[BEAM-4652] Allow PubsubIO to read public data #5788

kennknowles · 2018-06-27T18:53:39Z

This adds an integration test run for GCP IOs and then adds an integration test for reading a read public data set.

It does not pass, but it also does not fail in the same way that my local testing fails. When I write a pipeline to read this dataset I get a 403 on creating a subscription. The logs that gradle captures are a bit different.

Follow this checklist to help us incorporate your contribution quickly and easily:

Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Spark
Go	---	---	---	---	---
Java
Python	---		---	---	---

kennknowles

@chamikaramj @lgajowy do you know anything about how I should set this up?

kennknowles · 2018-06-27T18:53:57Z

...a/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadIT.java

+    PCollection<PubsubMessage> messages =
+        pipeline.apply(
+            PubsubIO.readMessages()
+                .fromTopic("projects/pubsub-public-data/topics/taxirides-realtime"));


@rangadi do you know about this?

kennknowles · 2018-06-27T18:54:16Z

...a/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadIT.java

+        "waitForAnyMessage", signal.signalSuccessWhen(messages.getCoder(), anyMessage -> true));
+
+    Duration timeout =
+        pipeline.getOptions().getRunner().getSimpleName().endsWith("DataflowRunner")


Hack so we don't have to it forever waiting for a DirectRunner to fail.

lgajowy · 2018-06-28T19:11:25Z

...a/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadIT.java

+            ? Duration.standardMinutes(5)
+            : Duration.standardSeconds(30);
+    signal.waitForSuccess(timeout);
+  }


I don't know much about Pubsub and testing it (we tested batch scenarios only yet), but doesn't this test require pipeline.run() invocation (sorry if this is my ignorance)?

Lol. Yes. And I was sprinkling debugging throughout, asking myself "why isn't createSubscription ever begin called?!"

OTOH the TestPipeline is supposed to complain if run() is not called. But I guess it never reaches that error because of the signal timeout.

Yes, I also think this is the case. I'm glad this helped a little bit. :)

kennknowles · 2018-06-28T20:23:13Z

OK, I have now confirmed that if I run the pipeline the test fails before my fix and succeeds after. So it is a real problem.

kennknowles · 2018-06-28T20:24:01Z

...le-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java

@@ -1297,7 +1298,9 @@ private SubscriptionPath createRandomSubscription(PipelineOptions options) {
              timestampAttribute, idAttribute, options.as(PubsubOptions.class))) {
        SubscriptionPath subscriptionPath =
            pubsubClient.createRandomSubscription(
-                project.get(), topic.get(), DEAULT_ACK_TIMEOUT_SEC);
+                PubsubClient.projectPathFromId(options.as(GcpOptions.class).getProject()),


But this is not quite right - it should respect the project ValueProvider if one was provided, yet default it to providing this.

kennknowles · 2018-06-28T21:01:58Z

run java postcommit

kennknowles · 2018-06-28T21:18:06Z

run java postcommit

rangadi · 2018-06-28T21:24:29Z

...le-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java

@@ -1235,7 +1235,7 @@ public PubsubUnboundedSource(
  /** Get the project path. */
  @Nullable
  public ProjectPath getProject() {
-    return project == null ? null : project.get();
+    return project != null ? null : project.get();


rangadi · 2018-06-28T21:45:41Z

.../java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java

@@ -693,9 +693,9 @@ public String toString() {

      @Nullable
      ValueProvider<ProjectPath> projectPath =
-          getTopicProvider() == null
+          getSubscriptionProvider() == null


Does this require user to provide subscription rather than topic for read? The integration test test added here specifies topic.

Here is what I can tell:

PubsubUnboundedSource is made so that you can customize the project for the subscription.

But PubsubIO is not made to customize the project.

Before my change:

PubsubIO always uses the project off the topic if it is read().fromTopic() and leaves it null if it is read().fromSubscription()

PubsubUnboundedSource always requires a project if it is given a topic because it needs it to create the random subscription

After my change:

PubsubIO always uses the project from the subscription and leaves it null for read().fromTopic()

PubsubUnboundedSource never requires a project, because it defaults to getting it from PipelineOptions

So I guess actually since PubsubIO never actually provides a useful project, it could always be left null. Or probably better to refactor PubsubUnboundedSource to have two variants with some shared internals so there are zero nullable fields.

Thanks. That makes sense. I was also wondering why createReader() also creates a random subscription if required (that does not sound correct). As you mentioned, may be best to always require project to be present in GcpOptions.

kennknowles · 2018-06-28T23:04:17Z

.../google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/TestPubsubSignal.java


    StatefulPredicateCheck(Coder<T> coder, SerializableFunction<Set<T>, Boolean> successPredicate) {
-      this.seenEvents = StateSpecs.set(coder);
+      this.seenEvents = StateSpecs.bag(coder);


FYI @akedin the only reason to use SetState is if you want to do efficient membership checking. Since the code here reads the whole thing, it doesn't add any functionality. And more runners support BagState so I switched it.

kennknowles · 2018-06-28T23:04:45Z

Precommit failed due to formatting and Postcommit failed because Dataflow doesn't support SetState. Should both be fixed. I'll run them both again.

kennknowles · 2018-06-28T23:04:50Z

run java postcommit

rangadi · 2018-06-28T23:26:44Z

PubsubIO/PubsubUnboundedSource changes LGTM. We can even just require --project option in GcpOptions to be always present, which is quite normal for many GCP components.

kennknowles · 2018-06-29T03:20:25Z

run java postcommit

kennknowles · 2018-06-29T11:50:41Z

run java postcommit

kennknowles · 2018-06-29T15:38:18Z

run java postcommit

kennknowles · 2018-06-29T21:21:26Z

run java postcommit

kennknowles · 2018-06-29T22:21:05Z

run java postcommit

kennknowles · 2018-06-29T23:47:30Z

run java postcommit

…yhow

This test does not pass because of idiosyncrasies of the TestDataflowRunner. It is clearly observable that the condition under test. Getting a DirectRunner integration test in place provides a regression test for our ability to read public data.

kennknowles · 2018-06-30T02:48:15Z

run java postcommit

kennknowles · 2018-06-30T03:34:02Z

It passed! The Python failure is on master, and fixed separately.

kennknowles · 2018-06-30T03:35:12Z

.../sql/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/pubsub/PubsubJsonIT.java

-            }
-          }
-        });
+    Future<List<String>> queryResult =


FYI @amaliujia

kennknowles commented Jun 27, 2018

View reviewed changes

lgajowy reviewed Jun 28, 2018

View reviewed changes

kennknowles force-pushed the pubsub-public-data branch from 18d0ef5 to daecb5c Compare June 28, 2018 20:22

kennknowles commented Jun 28, 2018

View reviewed changes

kennknowles force-pushed the pubsub-public-data branch from daecb5c to e451161 Compare June 28, 2018 21:00

kennknowles force-pushed the pubsub-public-data branch from e451161 to 5eca626 Compare June 28, 2018 21:13

kennknowles changed the title ~~[BEAM-4652] Pubsub public data~~ [BEAM-4652] Allow PubsubIO to read public data Jun 28, 2018

rangadi reviewed Jun 28, 2018

View reviewed changes

kennknowles force-pushed the pubsub-public-data branch from 5eca626 to c6bc82c Compare June 28, 2018 23:02

kennknowles commented Jun 28, 2018

View reviewed changes

kennknowles force-pushed the pubsub-public-data branch from c6bc82c to 5112419 Compare June 29, 2018 03:20

kennknowles force-pushed the pubsub-public-data branch 3 times, most recently from de5f515 to 93ea4b4 Compare June 29, 2018 15:17

kennknowles force-pushed the pubsub-public-data branch 5 times, most recently from 3de0308 to 6ff715e Compare June 29, 2018 21:20

kennknowles force-pushed the pubsub-public-data branch from 6ff715e to 85a683e Compare June 29, 2018 22:20

kennknowles force-pushed the pubsub-public-data branch from 85a683e to 5b9f073 Compare June 29, 2018 23:47

kennknowles and others added 10 commits June 29, 2018 19:47

Add integrationTest task for GCP IO module

2fa9f00

PubsubIO: Always use GCP project from PipelineOptions for subscriptions

69971af

Use BagState in PubsubSignal since it doesn't use SetState methods an…

aff6387

…yhow

Replace println with logging

edfaed3

More logging when PubsubJsonIT fails

3e293fc

Add start signal to TestPubsubSignal

6db089d

Use futures in PubsubJsonIT LIMIT test

8e3e0ca

Integration test of Pubsub public dataset

08e892f

Use test PipelineOptions in PubsubJsonIT LIMIT test

bdce4a0

Sickbay PubsubReadIT on Dataflow

b606d8a

This test does not pass because of idiosyncrasies of the TestDataflowRunner. It is clearly observable that the condition under test. Getting a DirectRunner integration test in place provides a regression test for our ability to read public data.

kennknowles force-pushed the pubsub-public-data branch from 5b9f073 to b606d8a Compare June 30, 2018 02:47

kennknowles merged commit d6c5bf9 into apache:master Jun 30, 2018

kennknowles commented Jun 30, 2018

View reviewed changes

kennknowles deleted the pubsub-public-data branch July 3, 2018 21:55

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-4652] Allow PubsubIO to read public data #5788

[BEAM-4652] Allow PubsubIO to read public data #5788

kennknowles commented Jun 27, 2018

kennknowles left a comment

kennknowles Jun 27, 2018

kennknowles Jun 27, 2018

lgajowy Jun 28, 2018 •

edited

kennknowles Jun 28, 2018

kennknowles Jun 28, 2018

lgajowy Jun 28, 2018

kennknowles commented Jun 28, 2018

kennknowles Jun 28, 2018

kennknowles Jun 28, 2018

kennknowles commented Jun 28, 2018

kennknowles commented Jun 28, 2018

rangadi Jun 28, 2018

kennknowles Jun 28, 2018

rangadi Jun 28, 2018

kennknowles Jun 28, 2018

rangadi Jun 28, 2018

kennknowles Jun 28, 2018

kennknowles commented Jun 28, 2018

kennknowles commented Jun 28, 2018

rangadi commented Jun 28, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 30, 2018

kennknowles commented Jun 30, 2018

kennknowles Jun 30, 2018

[BEAM-4652] Allow PubsubIO to read public data #5788

[BEAM-4652] Allow PubsubIO to read public data #5788

Conversation

kennknowles commented Jun 27, 2018

Post-Commit Tests Status (on master branch)

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lgajowy Jun 28, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Jun 28, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Jun 28, 2018

kennknowles commented Jun 28, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Jun 28, 2018

kennknowles commented Jun 28, 2018

rangadi commented Jun 28, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 29, 2018

kennknowles commented Jun 30, 2018

kennknowles commented Jun 30, 2018

Choose a reason for hiding this comment

lgajowy Jun 28, 2018 •

edited