From c640e7437871313303636def48d63acd678a3430 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 5 May 2017 10:36:40 -0700 Subject: [PATCH 1/3] Move PipelineRunner to toplevel sdk package (automated refactor) This allows excluding the runner-author-only sdk.runners package from the public javadoc. --- .../java/org/apache/beam/runners/apex/ApexRunner.java | 2 +- .../apache/beam/runners/apex/ApexRunnerRegistrar.java | 2 +- .../org/apache/beam/runners/apex/TestApexRunner.java | 2 +- .../main/java/org/apache/beam/runners/core/OldDoFn.java | 4 ++-- .../org/apache/beam/runners/direct/DirectRegistrar.java | 2 +- .../org/apache/beam/runners/direct/DirectRunner.java | 2 +- .../apache/beam/runners/direct/DoFnLifecycleManager.java | 2 +- .../org/apache/beam/runners/direct/DirectRunnerTest.java | 2 +- .../java/org/apache/beam/runners/flink/FlinkRunner.java | 2 +- .../apache/beam/runners/flink/FlinkRunnerRegistrar.java | 2 +- .../org/apache/beam/runners/flink/TestFlinkRunner.java | 2 +- .../beam/runners/dataflow/DataflowPipelineRegistrar.java | 2 +- .../org/apache/beam/runners/dataflow/DataflowRunner.java | 2 +- .../runners/dataflow/testing/TestDataflowRunner.java | 2 +- .../java/org/apache/beam/runners/spark/SparkRunner.java | 2 +- .../apache/beam/runners/spark/SparkRunnerDebugger.java | 2 +- .../apache/beam/runners/spark/SparkRunnerRegistrar.java | 2 +- .../org/apache/beam/runners/spark/TestSparkRunner.java | 2 +- .../core/src/main/java/org/apache/beam/sdk/Pipeline.java | 1 - .../apache/beam/sdk/{runners => }/PipelineRunner.java | 4 +--- .../src/main/java/org/apache/beam/sdk/coders/Coder.java | 9 +++++---- .../java/org/apache/beam/sdk/coders/package-info.java | 2 +- .../src/main/java/org/apache/beam/sdk/io/AvroSource.java | 2 +- .../org/apache/beam/sdk/options/PipelineOptions.java | 2 +- .../apache/beam/sdk/options/PipelineOptionsFactory.java | 2 +- .../src/main/java/org/apache/beam/sdk/package-info.java | 2 +- .../beam/sdk/runners/PTransformOverrideFactory.java | 1 + .../apache/beam/sdk/runners/PipelineRunnerRegistrar.java | 1 + .../java/org/apache/beam/sdk/state/StateContext.java | 3 ++- .../java/org/apache/beam/sdk/testing/CrashingRunner.java | 2 +- .../java/org/apache/beam/sdk/testing/NeedsRunner.java | 2 +- .../main/java/org/apache/beam/sdk/testing/PAssert.java | 2 +- .../java/org/apache/beam/sdk/testing/TestStream.java | 2 +- .../apache/beam/sdk/transforms/CombineWithContext.java | 3 ++- .../main/java/org/apache/beam/sdk/transforms/DoFn.java | 6 +++--- .../org/apache/beam/sdk/transforms/Materialization.java | 2 +- .../main/java/org/apache/beam/sdk/transforms/ParDo.java | 2 +- .../main/java/org/apache/beam/sdk/transforms/View.java | 2 +- .../java/org/apache/beam/sdk/values/PCollectionView.java | 3 ++- .../src/test/java/org/apache/beam/sdk/PipelineTest.java | 1 - .../beam/sdk/options/PipelineOptionsFactoryTest.java | 2 +- .../org/apache/beam/sdk/runners/PipelineRunnerTest.java | 1 + .../org/apache/beam/sdk/testing/CrashingRunnerTest.java | 2 +- .../sdk/transforms/display/DisplayDataEvaluator.java | 2 +- .../org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java | 2 +- .../org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java | 2 +- .../apache/beam/sdk/io/gcp/datastore/DatastoreV1.java | 3 ++- .../java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java | 2 +- .../src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java | 4 ++-- 49 files changed, 59 insertions(+), 55 deletions(-) rename sdks/java/core/src/main/java/org/apache/beam/sdk/{runners => }/PipelineRunner.java (94%) diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java index ede41bc4c351..a50e10ef53d8 100644 --- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java +++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java @@ -43,6 +43,7 @@ import org.apache.beam.runners.core.construction.PrimitiveCreate; import org.apache.beam.runners.core.construction.SingleInputOutputOverrideFactory; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; import org.apache.beam.sdk.coders.ListCoder; @@ -50,7 +51,6 @@ import org.apache.beam.sdk.options.PipelineOptionsValidator; import org.apache.beam.sdk.runners.AppliedPTransform; import org.apache.beam.sdk.runners.PTransformOverride; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.Combine.GloballyAsSingletonView; import org.apache.beam.sdk.transforms.Create; diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerRegistrar.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerRegistrar.java index 8cde692f8c15..2aa34ef15789 100644 --- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerRegistrar.java +++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerRegistrar.java @@ -20,9 +20,9 @@ import com.google.auto.service.AutoService; import com.google.common.collect.ImmutableList; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsRegistrar; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; /** diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/TestApexRunner.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/TestApexRunner.java index a64ac549ec70..e068db086442 100644 --- a/runners/apex/src/main/java/org/apache/beam/runners/apex/TestApexRunner.java +++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/TestApexRunner.java @@ -19,9 +19,9 @@ import java.io.IOException; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsValidator; -import org.apache.beam.sdk.runners.PipelineRunner; import org.joda.time.Duration; /** diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java index 419c83779057..41bb598795e6 100644 --- a/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java +++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/OldDoFn.java @@ -18,9 +18,9 @@ package org.apache.beam.runners.core; import java.io.Serializable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFnTester; import org.apache.beam.sdk.transforms.ParDo; @@ -71,7 +71,7 @@ public abstract class Context { /** * Returns the {@code PipelineOptions} specified with the - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * invoking this {@code OldDoFn}. The {@code PipelineOptions} will * be the default running via {@link DoFnTester}. */ diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRegistrar.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRegistrar.java index 9a34d3d9dfb4..3e69e2b527d2 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRegistrar.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRegistrar.java @@ -19,9 +19,9 @@ import com.google.auto.service.AutoService; import com.google.common.collect.ImmutableList; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsRegistrar; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; /** diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java index 984598a6faaf..b0ce5eb02845 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java @@ -34,12 +34,12 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.Pipeline.PipelineExecutionException; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.metrics.MetricResults; import org.apache.beam.sdk.metrics.MetricsEnvironment; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.runners.PTransformOverride; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.testing.TestStream; import org.apache.beam.sdk.transforms.GroupByKey; import org.apache.beam.sdk.transforms.PTransform; diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DoFnLifecycleManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DoFnLifecycleManager.java index cd644a66585b..0e30e5c07bb7 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DoFnLifecycleManager.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DoFnLifecycleManager.java @@ -26,7 +26,7 @@ import java.util.Collection; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import org.apache.beam.sdk.runners.PipelineRunner; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.DoFn.Setup; import org.apache.beam.sdk.transforms.DoFn.Teardown; diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java index 83881fc46f24..428c6fc43d71 100644 --- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java +++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java @@ -43,6 +43,7 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.PipelineResult.State; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderException; import org.apache.beam.sdk.coders.CustomCoder; @@ -55,7 +56,6 @@ import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Count; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java index a5972efa1a15..80ef7bb32acd 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java @@ -31,10 +31,10 @@ import java.util.TreeSet; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.metrics.MetricsEnvironment; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsValidator; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.View; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerRegistrar.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerRegistrar.java index 681459a819d7..ebc6adccf76c 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerRegistrar.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunnerRegistrar.java @@ -20,9 +20,9 @@ import com.google.auto.service.AutoService; import com.google.common.collect.ImmutableList; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsRegistrar; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java index 8f50105a55b9..01b67e517d8e 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java @@ -20,10 +20,10 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.Pipeline.PipelineExecutionException; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.PipelineOptionsValidator; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.util.UserCodeException; /** diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrar.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrar.java index 5bd3bcd6eb9a..b6802bb4b0aa 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrar.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrar.java @@ -20,9 +20,9 @@ import com.google.auto.service.AutoService; import com.google.common.collect.ImmutableList; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsRegistrar; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; /** diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java index 4198baea7517..5278a4a576d0 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java @@ -77,6 +77,7 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.Pipeline.PipelineVisitor; import org.apache.beam.sdk.PipelineResult.State; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.Coder.NonDeterministicException; @@ -102,7 +103,6 @@ import org.apache.beam.sdk.runners.AppliedPTransform; import org.apache.beam.sdk.runners.PTransformOverride; import org.apache.beam.sdk.runners.PTransformOverrideFactory; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.runners.TransformHierarchy.Node; import org.apache.beam.sdk.transforms.Combine; diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java index c238d80bc933..ce9191586d1f 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/testing/TestDataflowRunner.java @@ -38,8 +38,8 @@ import org.apache.beam.runners.dataflow.util.MonitoringUtil.JobMessagesHandler; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult.State; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestPipelineOptions; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java index 16d9fdcfbbe7..8c02f0f0e90d 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java @@ -41,13 +41,13 @@ import org.apache.beam.runners.spark.translation.streaming.SparkRunnerStreamingContextFactory; import org.apache.beam.runners.spark.util.GlobalWatermarkHolder.WatermarksListener; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.io.Read; import org.apache.beam.sdk.metrics.MetricsEnvironment; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.PipelineOptionsValidator; import org.apache.beam.sdk.runners.AppliedPTransform; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.transforms.Combine; import org.apache.beam.sdk.transforms.PTransform; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerDebugger.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerDebugger.java index 11c52c7a2ff3..6c74146bd61e 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerDebugger.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerDebugger.java @@ -25,9 +25,9 @@ import org.apache.beam.runners.spark.translation.TransformTranslator; import org.apache.beam.runners.spark.translation.streaming.StreamingTransformTranslator; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsValidator; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.streaming.api.java.JavaStreamingContext; import org.joda.time.Duration; diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerRegistrar.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerRegistrar.java index bf926dc54ecc..e2e5ceba0221 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerRegistrar.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunnerRegistrar.java @@ -20,9 +20,9 @@ import com.google.auto.service.AutoService; import com.google.common.collect.ImmutableList; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsRegistrar; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; /** diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/TestSparkRunner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/TestSparkRunner.java index ada7fb58fd7b..6d10b7588a42 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/TestSparkRunner.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/TestSparkRunner.java @@ -39,13 +39,13 @@ import org.apache.beam.runners.spark.util.GlobalWatermarkHolder; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.io.BoundedReadFromUnboundedSource; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsValidator; import org.apache.beam.sdk.runners.AppliedPTransform; import org.apache.beam.sdk.runners.PTransformOverride; import org.apache.beam.sdk.runners.PTransformOverrideFactory; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.values.PBegin; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java index 1769c0522c9c..f4da6ad9c1c2 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/Pipeline.java @@ -38,7 +38,6 @@ import org.apache.beam.sdk.runners.PTransformOverrideFactory; import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement; import org.apache.beam.sdk.runners.PTransformOverrideFactory.ReplacementOutput; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.runners.TransformHierarchy.Node; import org.apache.beam.sdk.transforms.Create; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunner.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/PipelineRunner.java similarity index 94% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunner.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/PipelineRunner.java index 229e04ffa12d..18e79eb7fa27 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunner.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/PipelineRunner.java @@ -15,12 +15,10 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.runners; +package org.apache.beam.sdk; import static com.google.common.base.Preconditions.checkNotNull; -import org.apache.beam.sdk.Pipeline; -import org.apache.beam.sdk.PipelineResult; import org.apache.beam.sdk.io.FileSystems; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsValidator; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/Coder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/Coder.java index 169e448623c7..061e9e535338 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/Coder.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/Coder.java @@ -29,6 +29,7 @@ import java.util.Arrays; import java.util.List; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.util.common.ElementByteSizeObserver; @@ -243,7 +244,7 @@ public abstract T decode(InputStream inStream, Context context) * constant time (or lazily). * *

Not intended to be called by user code, but instead by - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * implementations. */ public abstract boolean isRegisterByteSizeObserverCheap(T value); @@ -255,7 +256,7 @@ public abstract T decode(InputStream inStream, Context context) * constant time (or lazily). * *

Not intended to be called by user code, but instead by - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * implementations. */ @Deprecated @@ -266,7 +267,7 @@ public abstract T decode(InputStream inStream, Context context) * of the encoded value using this {@code Coder}. * *

Not intended to be called by user code, but instead by - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * implementations. */ public abstract void registerByteSizeObserver( @@ -278,7 +279,7 @@ public abstract void registerByteSizeObserver( * of the encoded value using this {@code Coder}. * *

Not intended to be called by user code, but instead by - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * implementations. */ @Deprecated diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/package-info.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/package-info.java index 84e59e018b45..5693077759c4 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/package-info.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/package-info.java @@ -27,7 +27,7 @@ * machines. * *

Exactly when PCollection elements are encoded during execution depends on which - * {@link org.apache.beam.sdk.runners.PipelineRunner} is being used and how that runner + * {@link org.apache.beam.sdk.PipelineRunner} is being used and how that runner * chooses to execute the pipeline. As such, Beam requires that all PCollections have an * appropriate Coder in case it becomes necessary. In many cases, the Coder can be inferred from * the available Java type diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java index 96d21c674658..61bc4a4f1563 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSource.java @@ -48,12 +48,12 @@ import org.apache.avro.io.DecoderFactory; import org.apache.avro.reflect.ReflectData; import org.apache.avro.reflect.ReflectDatumReader; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.coders.AvroCoder; import org.apache.beam.sdk.io.fs.MatchResult.Metadata; import org.apache.beam.sdk.io.fs.ResourceId; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.values.PCollection; import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream; import org.apache.commons.compress.compressors.snappy.SnappyCompressorInputStream; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java index 37081fde1a79..9a4d25ae3797 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java @@ -30,9 +30,9 @@ import java.util.concurrent.atomic.AtomicLong; import javax.annotation.concurrent.ThreadSafe; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.ProxyInvocationHandler.Deserializer; import org.apache.beam.sdk.options.ProxyInvocationHandler.Serializer; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.display.HasDisplayData; import org.joda.time.DateTimeUtils; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java index 0f6bedfc8b9d..c0990cb108a8 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java @@ -73,8 +73,8 @@ import java.util.TreeMap; import java.util.TreeSet; import javax.annotation.Nonnull; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.Validation.Required; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.util.StringUtils; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/package-info.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/package-info.java index 166c27651dc2..995bcb900854 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/package-info.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/package-info.java @@ -27,7 +27,7 @@ * the PTransforms consume and produce. * *

Each Pipeline has a - * {@link org.apache.beam.sdk.runners.PipelineRunner} to specify + * {@link org.apache.beam.sdk.PipelineRunner} to specify * where and how it should run after pipeline construction is complete. * */ diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformOverrideFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformOverrideFactory.java index 5f012dedca06..58314c40058c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformOverrideFactory.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PTransformOverrideFactory.java @@ -21,6 +21,7 @@ import com.google.auto.value.AutoValue; import java.util.Map; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.annotations.Internal; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunnerRegistrar.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunnerRegistrar.java index 41fd6f0029ba..c6303dded4ec 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunnerRegistrar.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/PipelineRunnerRegistrar.java @@ -19,6 +19,7 @@ import com.google.auto.service.AutoService; import java.util.ServiceLoader; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Internal; /** diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java index 52177cc0a41c..b7dd8e2681cf 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/state/StateContext.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.state; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; @@ -30,7 +31,7 @@ public interface StateContext { /** * Returns the {@code PipelineOptions} specified with the - * {@link org.apache.beam.sdk.runners.PipelineRunner}. + * {@link PipelineRunner}. */ PipelineOptions getPipelineOptions(); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/CrashingRunner.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/CrashingRunner.java index b3360ae6583e..0dea27460470 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/CrashingRunner.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/CrashingRunner.java @@ -20,8 +20,8 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; /** * A {@link PipelineRunner} that applies no overrides and throws an exception on calls to diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/NeedsRunner.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/NeedsRunner.java index 5f22bee631e8..8ebc76c66d58 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/NeedsRunner.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/NeedsRunner.java @@ -18,7 +18,7 @@ package org.apache.beam.sdk.testing; -import org.apache.beam.sdk.runners.PipelineRunner; +import org.apache.beam.sdk.PipelineRunner; /** * Category tag for validation tests which utilize {@link TestPipeline} for execution and expect diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java index b5d7db5fc204..6e2b8c66fa41 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/PAssert.java @@ -34,6 +34,7 @@ import java.util.NoSuchElementException; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.Pipeline.PipelineVisitor; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderException; import org.apache.beam.sdk.coders.IterableCoder; @@ -42,7 +43,6 @@ import org.apache.beam.sdk.coders.VarIntCoder; import org.apache.beam.sdk.metrics.Counter; import org.apache.beam.sdk.metrics.Metrics; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy.Node; import org.apache.beam.sdk.transforms.Create; import org.apache.beam.sdk.transforms.DoFn; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestStream.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestStream.java index 446c8a5d1974..d41b9ef5dc23 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestStream.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestStream.java @@ -25,8 +25,8 @@ import com.google.common.collect.ImmutableList; import java.util.List; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.Coder; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.windowing.BoundedWindow; import org.apache.beam.sdk.values.PBegin; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineWithContext.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineWithContext.java index 9ae19f8bfdd1..a7601b3cfc2c 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineWithContext.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineWithContext.java @@ -17,6 +17,7 @@ */ package org.apache.beam.sdk.transforms; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.transforms.Combine.CombineFn; import org.apache.beam.sdk.values.PCollectionView; @@ -36,7 +37,7 @@ public class CombineWithContext { public abstract static class Context { /** * Returns the {@code PipelineOptions} specified with the - * {@link org.apache.beam.sdk.runners.PipelineRunner} + * {@link PipelineRunner} * invoking this {@code KeyedCombineFn}. */ public abstract PipelineOptions getPipelineOptions(); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java index befba1dd71d7..e711ac2f297a 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java @@ -88,7 +88,7 @@ public abstract class DoFn implements Serializable, HasDisplayD public abstract class StartBundleContext { /** * Returns the {@code PipelineOptions} specified with the {@link - * org.apache.beam.sdk.runners.PipelineRunner} invoking this {@code DoFn}. The {@code + * org.apache.beam.sdk.PipelineRunner} invoking this {@code DoFn}. The {@code * PipelineOptions} will be the default running via {@link DoFnTester}. */ public abstract PipelineOptions getPipelineOptions(); @@ -100,7 +100,7 @@ public abstract class StartBundleContext { public abstract class FinishBundleContext { /** * Returns the {@code PipelineOptions} specified with the {@link - * org.apache.beam.sdk.runners.PipelineRunner} invoking this {@code DoFn}. The {@code + * org.apache.beam.sdk.PipelineRunner} invoking this {@code DoFn}. The {@code * PipelineOptions} will be the default running via {@link DoFnTester}. */ public abstract PipelineOptions getPipelineOptions(); @@ -136,7 +136,7 @@ public abstract void output( public abstract class WindowedContext { /** * Returns the {@code PipelineOptions} specified with the {@link - * org.apache.beam.sdk.runners.PipelineRunner} invoking this {@code DoFn}. The {@code + * org.apache.beam.sdk.PipelineRunner} invoking this {@code DoFn}. The {@code * PipelineOptions} will be the default running via {@link DoFnTester}. */ public abstract PipelineOptions getPipelineOptions(); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java index 6fb8c297f547..0d02b32d3cc3 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Materialization.java @@ -18,8 +18,8 @@ package org.apache.beam.sdk.transforms; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Internal; -import org.apache.beam.sdk.runners.PipelineRunner; /** * For internal use only; no backwards-compatibility guarantees. diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java index c45311a7b0ab..e67dbe153a5e 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java @@ -29,10 +29,10 @@ import java.util.List; import java.util.Map; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.CannotProvideCoderException; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderRegistry; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.state.StateSpec; import org.apache.beam.sdk.transforms.DoFn.WindowedContext; import org.apache.beam.sdk.transforms.display.DisplayData; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/View.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/View.java index d7b81455b63f..bcbdb24d5290 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/View.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/View.java @@ -19,10 +19,10 @@ import java.util.List; import java.util.Map; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Internal; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderException; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy.Node; import org.apache.beam.sdk.util.CoderUtils; import org.apache.beam.sdk.values.KV; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionView.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionView.java index f89041a6124e..7d874124a9bd 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionView.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionView.java @@ -19,6 +19,7 @@ import java.io.Serializable; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.annotations.Internal; @@ -36,7 +37,7 @@ * *

A {@link PCollectionView} should always be the output of a * {@link org.apache.beam.sdk.transforms.PTransform}. It is the joint responsibility of - * this transform and each {@link org.apache.beam.sdk.runners.PipelineRunner} to implement + * this transform and each {@link PipelineRunner} to implement * the view in a runner-specific manner. * *

The most common case is using the {@link View} transforms to prepare a {@link PCollection} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java index d6b527ccf6ee..2cc3f04203d1 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/PipelineTest.java @@ -44,7 +44,6 @@ import org.apache.beam.sdk.runners.PTransformMatcher; import org.apache.beam.sdk.runners.PTransformOverride; import org.apache.beam.sdk.runners.PTransformOverrideFactory; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy.Node; import org.apache.beam.sdk.testing.CrashingRunner; import org.apache.beam.sdk.testing.ExpectedLogs; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java index 76a5f18c0397..d40b5fcc0dc6 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java @@ -53,7 +53,7 @@ import java.util.Set; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; -import org.apache.beam.sdk.runners.PipelineRunner; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.runners.PipelineRunnerRegistrar; import org.apache.beam.sdk.testing.CrashingRunner; import org.apache.beam.sdk.testing.ExpectedLogs; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java index 76d8627b19ed..c5d7fbf7cf23 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/runners/PipelineRunnerTest.java @@ -19,6 +19,7 @@ import static org.junit.Assert.assertTrue; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.testing.CrashingRunner; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/CrashingRunnerTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/CrashingRunnerTest.java index c66aa50cb57f..62c5134ce67a 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/CrashingRunnerTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/testing/CrashingRunnerTest.java @@ -22,9 +22,9 @@ import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.Create; import org.junit.Rule; import org.junit.Test; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java index 8f57f45b0ac4..6e8ca5e5db2f 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/display/DisplayDataEvaluator.java @@ -21,10 +21,10 @@ import java.util.Objects; import java.util.Set; import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.VoidCoder; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.runners.TransformHierarchy; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Create; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java index 6e410c7b6c28..304864a7ca85 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java @@ -43,6 +43,7 @@ import java.util.Map; import java.util.regex.Pattern; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.CannotProvideCoderException; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.KvCoder; @@ -66,7 +67,6 @@ import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.SerializableFunction; import org.apache.beam.sdk.transforms.display.DisplayData; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java index 0e97c1254d93..846d07bb78a2 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java @@ -42,6 +42,7 @@ import java.util.NoSuchElementException; import java.util.concurrent.ConcurrentLinkedQueue; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.extensions.gcp.options.GcpOptions; @@ -52,7 +53,6 @@ import org.apache.beam.sdk.io.range.ByteKeyRange; import org.apache.beam.sdk.io.range.ByteKeyRangeTracker; import org.apache.beam.sdk.options.PipelineOptions; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/datastore/DatastoreV1.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/datastore/DatastoreV1.java index fd4fccf0ba1d..d6464dd95855 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/datastore/DatastoreV1.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/datastore/DatastoreV1.java @@ -69,6 +69,7 @@ import java.util.List; import java.util.NoSuchElementException; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.annotations.Experimental.Kind; import org.apache.beam.sdk.coders.SerializableCoder; @@ -192,7 +193,7 @@ * by providing the host port information through {@code withLocalhost("host:port"} for all the * above transforms. In such a case, all the Cloud Datastore API calls are directed to the Emulator. * - * @see org.apache.beam.sdk.runners.PipelineRunner + * @see PipelineRunner */ public class DatastoreV1 { diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java index 51da111ccd24..ac6cb44be291 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java @@ -31,6 +31,7 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import javax.annotation.Nullable; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.coders.AvroCoder; import org.apache.beam.sdk.coders.Coder; import org.apache.beam.sdk.coders.CoderException; @@ -44,7 +45,6 @@ import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.options.ValueProvider.NestedValueProvider; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.PTransform; diff --git a/sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java b/sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java index a34fb0f3cda0..7255a94357eb 100644 --- a/sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java +++ b/sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java @@ -26,6 +26,7 @@ import javax.xml.bind.JAXBContext; import javax.xml.bind.JAXBException; import javax.xml.bind.ValidationEventHandler; +import org.apache.beam.sdk.PipelineRunner; import org.apache.beam.sdk.io.BoundedSource; import org.apache.beam.sdk.io.CompressedSource; import org.apache.beam.sdk.io.FileBasedSink; @@ -34,7 +35,6 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.ValueProvider; import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider; -import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.values.PBegin; @@ -102,7 +102,7 @@ public class XmlIO { * *

Permissions

* - *

Permission requirements depend on the {@link org.apache.beam.sdk.runners.PipelineRunner + *

Permission requirements depend on the {@link PipelineRunner * PipelineRunner} that is used to execute the Beam pipeline. Please refer to the documentation of * corresponding {@link PipelineRunner PipelineRunners} for more details. * From b64f9acb116e92fb9ad35304a770aae92338a404 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 5 May 2017 10:33:11 -0700 Subject: [PATCH 2/3] Exclude sdk.runners from javadoc --- sdks/java/javadoc/ant.xml | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/java/javadoc/ant.xml b/sdks/java/javadoc/ant.xml index 288f48bf9baa..9a1880acaa17 100644 --- a/sdks/java/javadoc/ant.xml +++ b/sdks/java/javadoc/ant.xml @@ -90,6 +90,7 @@ + From 15be5df5d83dd6d151e4b2ace0df8286a378b008 Mon Sep 17 00:00:00 2001 From: Kenneth Knowles Date: Fri, 5 May 2017 10:39:26 -0700 Subject: [PATCH 3/3] Javadoc that the sdk.runners package is internal --- .../main/java/org/apache/beam/sdk/runners/package-info.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/package-info.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/package-info.java index bb6cce6079cb..272693643d60 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/package-info.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/runners/package-info.java @@ -16,8 +16,8 @@ * limitations under the License. */ /** - * Defines different data processing backends (aka, runners) for executing Beam Pipelines. + * For internal use only; no backwards compatibility guarantees. - *

The runner is specified as part of the {@link org.apache.beam.sdk.options.PipelineOptions}. + *

Internals for use by runners. */ package org.apache.beam.sdk.runners;