Disable use of GcsIOChannelFactory on query10 Make NexmarkOptions depend on GcpOptions instead of PubsubOptions
issue #21 Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/902050b0 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/902050b0 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/902050b0 Branch: refs/heads/master Commit: 902050b0b276e22ab002e8efb390bbaa01e18e99 Parents: 1541fad Author: Ismaël MejÃa <ieme...@apache.org> Authored: Fri May 5 10:47:46 2017 +0200 Committer: Ismaël MejÃa <ieme...@gmail.com> Committed: Wed Aug 23 19:07:28 2017 +0200 ---------------------------------------------------------------------- integration/java/nexmark/pom.xml | 5 ----- .../beam/integration/nexmark/NexmarkOptions.java | 7 ++++++- .../beam/integration/nexmark/queries/Query10.java | 13 +++++++------ 3 files changed, 13 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/902050b0/integration/java/nexmark/pom.xml ---------------------------------------------------------------------- diff --git a/integration/java/nexmark/pom.xml b/integration/java/nexmark/pom.xml index 8a65c0f..35fe0f3 100644 --- a/integration/java/nexmark/pom.xml +++ b/integration/java/nexmark/pom.xml @@ -218,11 +218,6 @@ <artifactId>google-api-services-bigquery</artifactId> </dependency> - <dependency> - <groupId>com.google.cloud.bigdataoss</groupId> - <artifactId>gcsio</artifactId> - </dependency> - <!-- Extra libraries --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> http://git-wip-us.apache.org/repos/asf/beam/blob/902050b0/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/NexmarkOptions.java ---------------------------------------------------------------------- diff --git a/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/NexmarkOptions.java b/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/NexmarkOptions.java index e1c1af2..f162fd6 100644 --- a/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/NexmarkOptions.java +++ b/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/NexmarkOptions.java @@ -18,14 +18,19 @@ package org.apache.beam.integration.nexmark; import javax.annotation.Nullable; +import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; import org.apache.beam.sdk.io.gcp.pubsub.PubsubOptions; +import org.apache.beam.sdk.options.ApplicationNameOptions; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description; +import org.apache.beam.sdk.options.PipelineOptions; +import org.apache.beam.sdk.options.StreamingOptions; /** * Command line flags. */ -public interface NexmarkOptions extends PubsubOptions { +public interface NexmarkOptions + extends ApplicationNameOptions, GcpOptions, PipelineOptions, StreamingOptions { @Description("Which suite to run. Default is to use command line arguments for one job.") @Default.Enum("DEFAULT") NexmarkSuite getSuite(); http://git-wip-us.apache.org/repos/asf/beam/blob/902050b0/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/queries/Query10.java ---------------------------------------------------------------------- diff --git a/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/queries/Query10.java b/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/queries/Query10.java index 516dab1..c868666 100644 --- a/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/queries/Query10.java +++ b/integration/java/nexmark/src/main/java/org/apache/beam/integration/nexmark/queries/Query10.java @@ -50,7 +50,6 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing; import org.apache.beam.sdk.transforms.windowing.Repeatedly; import org.apache.beam.sdk.transforms.windowing.Window; -import org.apache.beam.sdk.util.GcsIOChannelFactory; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.joda.time.Duration; @@ -133,11 +132,13 @@ public class Query10 extends NexmarkQuery { */ private WritableByteChannel openWritableGcsFile(GcsOptions options, String filename) throws IOException { - WritableByteChannel channel = - GcsIOChannelFactory.fromOptions(options).create(filename, "text/plain"); - checkState(channel instanceof GoogleCloudStorageWriteChannel); - ((GoogleCloudStorageWriteChannel) channel).setUploadBufferSize(CHANNEL_BUFFER); - return channel; + //TODO Decide what to do about this one +// WritableByteChannel channel = +// GcsIOChannelFactory.fromOptions(options).create(filename, "text/plain"); +// checkState(channel instanceof GoogleCloudStorageWriteChannel); +// ((GoogleCloudStorageWriteChannel) channel).setUploadBufferSize(CHANNEL_BUFFER); +// return channel; + throw new UnsupportedOperationException("Disabled after removal of GcsIOChannelFactory"); } /** Return a short string to describe {@code timing}. */