[BEAM-151] Migrate options relevant to pubsub to its own options interface This prevents moving DataflowPipelineOptions to the Dataflow runner maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c8cb55af Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c8cb55af Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c8cb55af Branch: refs/heads/master Commit: c8cb55af49832f39788e1936a9576760fd720eb1 Parents: a502baa Author: Luke Cwik <[email protected]> Authored: Mon Mar 28 10:02:01 2016 -0700 Committer: Luke Cwik <[email protected]> Committed: Thu Apr 7 11:19:49 2016 -0700 ---------------------------------------------------------------------- .../google/cloud/dataflow/sdk/io/PubsubIO.java | 6 ++-- .../options/DataflowPipelineDebugOptions.java | 8 ----- .../sdk/options/DataflowPipelineOptions.java | 2 +- .../dataflow/sdk/options/PubsubOptions.java | 35 ++++++++++++++++++++ .../cloud/dataflow/sdk/util/Transport.java | 3 +- 5 files changed, 41 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java index b7f2afe..68615c5 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java @@ -33,7 +33,7 @@ import com.google.api.services.pubsub.model.Subscription; import com.google.cloud.dataflow.sdk.coders.Coder; import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder; import com.google.cloud.dataflow.sdk.coders.VoidCoder; -import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions; +import com.google.cloud.dataflow.sdk.options.PubsubOptions; import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner; import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; import com.google.cloud.dataflow.sdk.runners.PipelineRunner; @@ -728,7 +728,7 @@ public class PubsubIO { @Override public void processElement(ProcessContext c) throws IOException { Pubsub pubsubClient = - Transport.newPubsubClient(c.getPipelineOptions().as(DataflowPipelineOptions.class)) + Transport.newPubsubClient(c.getPipelineOptions().as(PubsubOptions.class)) .build(); String subscription; @@ -1004,7 +1004,7 @@ public class PubsubIO { public void startBundle(Context c) { this.output = new ArrayList<>(); this.pubsubClient = - Transport.newPubsubClient(c.getPipelineOptions().as(DataflowPipelineOptions.class)) + Transport.newPubsubClient(c.getPipelineOptions().as(PubsubOptions.class)) .build(); } http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java index d4f6385..b55fa17 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java @@ -164,14 +164,6 @@ public interface DataflowPipelineDebugOptions extends PipelineOptions { } /** - * Root URL for use with the Pubsub API. - */ - @Description("Root URL for use with the Pubsub API") - @Default.String("https://pubsub.googleapis.com";) - String getPubsubRootUrl(); - void setPubsubRootUrl(String value); - - /** * Whether to update the currently running pipeline with the same name as this one. * * @deprecated This property is replaced by {@link DataflowPipelineOptions#getUpdate()} http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java index 23ca0e7..dbfafd1 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java @@ -33,7 +33,7 @@ public interface DataflowPipelineOptions extends PipelineOptions, GcpOptions, ApplicationNameOptions, DataflowPipelineDebugOptions, DataflowPipelineWorkerPoolOptions, BigQueryOptions, GcsOptions, StreamingOptions, CloudDebuggerOptions, DataflowWorkerLoggingOptions, - DataflowProfilingOptions { + DataflowProfilingOptions, PubsubOptions { @Description("Project id. Required when running a Dataflow in the cloud. " + "See https://cloud.google.com/storage/docs/projects for further details.") http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java new file mode 100644 index 0000000..51cf62c --- /dev/null +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java @@ -0,0 +1,35 @@ +/* + * Copyright (C) 2016 Google Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not + * use this file except in compliance with the License. You may obtain a copy of + * the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ + +package com.google.cloud.dataflow.sdk.options; + +/** + * Properties that can be set when using Pubsub with the Dataflow SDK. + */ +@Description("Options that are used to configure BigQuery. See " + + "https://cloud.google.com/bigquery/what-is-bigquery for details on BigQuery.") +public interface PubsubOptions extends ApplicationNameOptions, GcpOptions, + PipelineOptions, StreamingOptions { + + /** + * Root URL for use with the Pubsub API. + */ + @Description("Root URL for use with the Pubsub API") + @Default.String("https://pubsub.googleapis.com";) + @Hidden + String getPubsubRootUrl(); + void setPubsubRootUrl(String value); +} http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java ---------------------------------------------------------------------- diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java index 27f61fa..187d164 100644 --- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java +++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java @@ -32,6 +32,7 @@ import com.google.cloud.dataflow.sdk.options.BigQueryOptions; import com.google.cloud.dataflow.sdk.options.DataflowPipelineDebugOptions; import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions; import com.google.cloud.dataflow.sdk.options.GcsOptions; +import com.google.cloud.dataflow.sdk.options.PubsubOptions; import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer; import com.google.common.collect.ImmutableList; @@ -115,7 +116,7 @@ public class Transport { * {@link DataflowPipelineDebugOptions#getApiRootUrl()} option. */ public static Pubsub.Builder - newPubsubClient(DataflowPipelineOptions options) { + newPubsubClient(PubsubOptions options) { return new Pubsub.Builder(getTransport(), getJsonFactory(), chainHttpRequestInitializer( options.getGcpCredential(),
