Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id 5189B200B35 for ; Tue, 21 Jun 2016 00:16:20 +0200 (CEST) Received: by cust-asf.ponee.io (Postfix) id 502C1160A69; Mon, 20 Jun 2016 22:16:20 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 83D7F160A68 for ; Tue, 21 Jun 2016 00:16:18 +0200 (CEST) Received: (qmail 12514 invoked by uid 500); 20 Jun 2016 22:16:17 -0000 Mailing-List: contact commits-help@beam.incubator.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@beam.incubator.apache.org Delivered-To: mailing list commits@beam.incubator.apache.org Received: (qmail 12422 invoked by uid 99); 20 Jun 2016 22:16:17 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd4-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 20 Jun 2016 22:16:17 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd4-us-west.apache.org (ASF Mail Server at spamd4-us-west.apache.org) with ESMTP id 39F22C0E1C for ; Mon, 20 Jun 2016 22:16:17 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd4-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: -4.646 X-Spam-Level: X-Spam-Status: No, score=-4.646 tagged_above=-999 required=6.31 tests=[KAM_ASCII_DIVIDERS=0.8, KAM_LAZY_DOMAIN_SECURITY=1, RCVD_IN_DNSWL_HI=-5, RCVD_IN_MSPIKE_H3=-0.01, RCVD_IN_MSPIKE_WL=-0.01, RP_MATCHES_RCVD=-1.426] autolearn=disabled Received: from mx2-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd4-us-west.apache.org [10.40.0.11]) (amavisd-new, port 10024) with ESMTP id CIXgyQLth3_E for ; Mon, 20 Jun 2016 22:16:04 +0000 (UTC) Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx2-lw-eu.apache.org (ASF Mail Server at mx2-lw-eu.apache.org) with SMTP id 020B760E49 for ; Mon, 20 Jun 2016 22:15:56 +0000 (UTC) Received: (qmail 8288 invoked by uid 99); 20 Jun 2016 22:15:55 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 20 Jun 2016 22:15:55 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 5F19FDFE65; Mon, 20 Jun 2016 22:15:55 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: davor@apache.org To: commits@beam.incubator.apache.org Date: Mon, 20 Jun 2016 22:16:37 -0000 Message-Id: <0d87568dfc994af6a505621cec651cb1@git.apache.org> In-Reply-To: <483b066d27bf4a4988cffaf4189af9f6@git.apache.org> References: <483b066d27bf4a4988cffaf4189af9f6@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [44/50] [abbrv] incubator-beam git commit: Rename DataflowPipelineRunner to DataflowRunner archived-at: Mon, 20 Jun 2016 22:16:20 -0000 Rename DataflowPipelineRunner to DataflowRunner Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/6d028ac6 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/6d028ac6 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/6d028ac6 Branch: refs/heads/python-sdk Commit: 6d028ac6118e31ee2415f77cf744a99c59b31d41 Parents: 340d098 Author: Thomas Groh Authored: Fri Jun 17 11:42:19 2016 -0700 Committer: Davor Bonaci Committed: Mon Jun 20 15:14:31 2016 -0700 ---------------------------------------------------------------------- .../beam/examples/DebuggingWordCount.java | 2 +- .../apache/beam/examples/MinimalWordCount.java | 4 +- .../apache/beam/examples/WindowedWordCount.java | 2 +- .../org/apache/beam/examples/WordCount.java | 2 +- .../examples/common/DataflowExampleUtils.java | 18 +- .../beam/examples/complete/AutoComplete.java | 4 +- .../apache/beam/examples/complete/TfIdf.java | 2 +- .../examples/complete/TopWikipediaSessions.java | 2 +- .../examples/cookbook/BigQueryTornadoes.java | 2 +- .../cookbook/CombinePerKeyExamples.java | 2 +- .../examples/cookbook/DatastoreWordCount.java | 4 +- .../beam/examples/cookbook/DeDupExample.java | 2 +- .../beam/examples/cookbook/FilterExamples.java | 2 +- .../beam/examples/cookbook/JoinExamples.java | 2 +- .../examples/cookbook/MaxPerKeyExamples.java | 2 +- .../beam/examples/MinimalWordCountJava8.java | 4 +- .../beam/examples/complete/game/GameStats.java | 2 +- .../examples/complete/game/HourlyTeamScore.java | 2 +- .../examples/complete/game/LeaderBoard.java | 2 +- .../beam/examples/complete/game/UserScore.java | 2 +- .../beam/runners/flink/FlinkPipelineRunner.java | 2 +- .../BlockingDataflowPipelineRunner.java | 186 - .../dataflow/BlockingDataflowRunner.java | 186 + .../dataflow/DataflowJobCancelledException.java | 2 +- .../dataflow/DataflowJobExecutionException.java | 2 +- .../dataflow/DataflowJobUpdatedException.java | 2 +- .../runners/dataflow/DataflowPipelineJob.java | 2 +- .../dataflow/DataflowPipelineRegistrar.java | 8 +- .../dataflow/DataflowPipelineRunner.java | 3229 ------------------ .../dataflow/DataflowPipelineRunnerHooks.java | 39 - .../dataflow/DataflowPipelineTranslator.java | 10 +- .../beam/runners/dataflow/DataflowRunner.java | 3229 ++++++++++++++++++ .../runners/dataflow/DataflowRunnerHooks.java | 39 + .../runners/dataflow/internal/package-info.java | 2 +- .../BlockingDataflowPipelineOptions.java | 6 +- .../options/DataflowPipelineOptions.java | 4 +- .../DataflowPipelineWorkerPoolOptions.java | 6 +- .../testing/TestDataflowPipelineRunner.java | 10 +- .../BlockingDataflowPipelineRunnerTest.java | 304 -- .../dataflow/BlockingDataflowRunnerTest.java | 304 ++ .../dataflow/DataflowPipelineRegistrarTest.java | 4 +- .../dataflow/DataflowPipelineRunnerTest.java | 1417 -------- .../DataflowPipelineTranslatorTest.java | 50 +- .../runners/dataflow/DataflowRunnerTest.java | 1417 ++++++++ .../runners/dataflow/io/DataflowAvroIOTest.java | 4 +- .../dataflow/io/DataflowPubsubIOTest.java | 4 +- .../runners/dataflow/io/DataflowTextIOTest.java | 4 +- .../testing/TestDataflowPipelineRunnerTest.java | 601 ---- .../testing/TestDataflowRunnerTest.java | 601 ++++ .../DataflowDisplayDataEvaluator.java | 10 +- .../transforms/DataflowGroupByKeyTest.java | 8 +- .../dataflow/transforms/DataflowViewTest.java | 8 +- .../util/DataflowPathValidatorTest.java | 8 +- .../beam/runners/spark/SimpleWordCountTest.java | 2 +- .../spark/translation/SerializationTest.java | 2 +- .../apache/beam/sdk/runners/package-info.java | 12 +- .../apache/beam/sdk/testing/TestPipeline.java | 2 +- .../apache/beam/sdk/transforms/Aggregator.java | 2 +- .../beam/sdk/util/BigQueryTableRowIterator.java | 2 +- .../beam/sdk/runners/PipelineRunnerTest.java | 2 +- .../src/main/java/DebuggingWordCount.java | 2 +- .../src/main/java/MinimalWordCount.java | 4 +- .../src/main/java/WindowedWordCount.java | 2 +- .../src/main/java/WordCount.java | 2 +- .../main/java/common/DataflowExampleUtils.java | 14 +- .../src/main/java/StarterPipeline.java | 2 +- .../src/main/java/it/pkg/StarterPipeline.java | 2 +- sdks/java/pom.xml | 2 +- 68 files changed, 5913 insertions(+), 5913 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java index 11c56d1..85823c2 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/DebuggingWordCount.java @@ -69,7 +69,7 @@ import java.util.regex.Pattern; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  *   --workerLogLevelOverrides={"org.apache.beam.examples":"DEBUG"}
  * }
  *
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java index 65d450d..6d4bfd4 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java @@ -17,7 +17,7 @@ */ package org.apache.beam.examples; -import org.apache.beam.runners.dataflow.BlockingDataflowPipelineRunner; +import org.apache.beam.runners.dataflow.BlockingDataflowRunner; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.io.TextIO; @@ -65,7 +65,7 @@ public class MinimalWordCount { // in Google Cloud Storage to stage files. DataflowPipelineOptions options = PipelineOptionsFactory.create() .as(DataflowPipelineOptions.class); - options.setRunner(BlockingDataflowPipelineRunner.class); + options.setRunner(BlockingDataflowRunner.class); // CHANGE 1/3: Your project ID is required in order to run your pipeline on the Google Cloud. options.setProject("SET_YOUR_PROJECT_ID_HERE"); // CHANGE 2/3: Your Google Cloud Storage path is required for staging local files. http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java index 9ca26bf..d4f1280 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java @@ -84,7 +84,7 @@ import java.util.List; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/WordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/WordCount.java b/examples/java/src/main/java/org/apache/beam/examples/WordCount.java index 03ed5d7..cf6c45a 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/WordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/WordCount.java @@ -82,7 +82,7 @@ import java.io.IOException; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and an output prefix on GCS: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java b/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java index 46b8af3..b436ed1 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java +++ b/examples/java/src/main/java/org/apache/beam/examples/common/DataflowExampleUtils.java @@ -17,9 +17,9 @@ */ package org.apache.beam.examples.common; -import org.apache.beam.runners.dataflow.BlockingDataflowPipelineRunner; +import org.apache.beam.runners.dataflow.BlockingDataflowRunner; import org.apache.beam.runners.dataflow.DataflowPipelineJob; -import org.apache.beam.runners.dataflow.DataflowPipelineRunner; +import org.apache.beam.runners.dataflow.DataflowRunner; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; import org.apache.beam.runners.dataflow.util.MonitoringUtil; import org.apache.beam.sdk.Pipeline; @@ -316,16 +316,16 @@ public class DataflowExampleUtils { /** * Do some runner setup: check that the DirectRunner is not used in conjunction with - * streaming, and if streaming is specified, use the DataflowPipelineRunner. + * streaming, and if streaming is specified, use the DataflowRunner. */ public void setupRunner() { Class> runner = options.getRunner(); if (options.isStreaming() - && (runner.equals(DataflowPipelineRunner.class) - || runner.equals(BlockingDataflowPipelineRunner.class))) { + && (runner.equals(DataflowRunner.class) + || runner.equals(BlockingDataflowRunner.class))) { // In order to cancel the pipelines automatically, - // {@literal DataflowPipelineRunner} is forced to be used. - options.setRunner(DataflowPipelineRunner.class); + // {@literal DataflowRunner} is forced to be used. + options.setRunner(DataflowRunner.class); } } @@ -363,7 +363,7 @@ public class DataflowExampleUtils { } copiedOptions.setStreaming(false); copiedOptions.setWorkerHarnessContainerImage( - DataflowPipelineRunner.BATCH_WORKER_HARNESS_CONTAINER_IMAGE); + DataflowRunner.BATCH_WORKER_HARNESS_CONTAINER_IMAGE); copiedOptions.setNumWorkers(options.as(DataflowExampleOptions.class).getInjectorNumWorkers()); copiedOptions.setJobName(options.getJobName() + "-injector"); Pipeline injectorPipeline = Pipeline.create(copiedOptions); @@ -396,7 +396,7 @@ public class DataflowExampleUtils { } /** - * If {@literal DataflowPipelineRunner} or {@literal BlockingDataflowPipelineRunner} is used, + * If {@literal DataflowRunner} or {@literal BlockingDataflowRunner} is used, * waits for the pipeline to finish and cancels it (and the injector) before the program exists. */ public void waitToFinish(PipelineResult result) { http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java index 3e4440c..ef47762 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java +++ b/examples/java/src/main/java/org/apache/beam/examples/complete/AutoComplete.java @@ -82,7 +82,7 @@ import java.util.regex.Pattern; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=DataflowPipelineRunner
+ *   --runner=DataflowRunner
  *   --inputFile=gs://path/to/input*.txt
  * }
* @@ -91,7 +91,7 @@ import java.util.regex.Pattern; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=DataflowPipelineRunner
+ *   --runner=DataflowRunner
  *   --inputFile=gs://YOUR_INPUT_DIRECTORY/*.txt
  *   --streaming
  * }
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java index 1f1885f..73f3323 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java +++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TfIdf.java @@ -80,7 +80,7 @@ import java.util.Set; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * and an output prefix on GCS:
  *   --output=gs://YOUR_OUTPUT_PREFIX
  * }
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java b/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java index c483608..2c0d0eb 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java +++ b/examples/java/src/main/java/org/apache/beam/examples/complete/TopWikipediaSessions.java @@ -60,7 +60,7 @@ import java.util.List; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and an output prefix on GCS: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java index 4c69efb..a0ac6c5 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/BigQueryTornadoes.java @@ -61,7 +61,7 @@ import java.util.List; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and the BigQuery table for the output: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java index 505989c..dbaa1f8 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/CombinePerKeyExamples.java @@ -68,7 +68,7 @@ import java.util.List; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and the BigQuery table for the output: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/DatastoreWordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/DatastoreWordCount.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/DatastoreWordCount.java index 2d1f88c..b0192c9 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/DatastoreWordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/DatastoreWordCount.java @@ -71,7 +71,7 @@ import javax.annotation.Nullable; *

To run this example using Dataflow service, you must additionally * provide either {@literal --tempLocation} or {@literal --tempLocation}, and * select one of the Dataflow pipeline runners, eg - * {@literal --runner=BlockingDataflowPipelineRunner}. + * {@literal --runner=BlockingDataflowRunner}. * *

Note: this example creates entities with Ancestor keys to ensure that all * entities created are in the same entity group. Similarly, the query used to read from the Cloud @@ -248,7 +248,7 @@ public class DatastoreWordCount { /** * An example to demo how to use {@link DatastoreIO}. The runner here is * customizable, which means users could pass either {@code DirectRunner} - * or {@code DataflowPipelineRunner} in the pipeline options. + * or {@code DataflowRunner} in the pipeline options. */ public static void main(String args[]) { // The options are used in two places, for Dataflow service, and http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/DeDupExample.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/DeDupExample.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/DeDupExample.java index 37ddb4b..db65543 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/DeDupExample.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/DeDupExample.java @@ -43,7 +43,7 @@ import org.apache.beam.sdk.util.gcsfs.GcsPath; *

To execute this pipeline using the Dataflow service, specify pipeline configuration: * --project=YOUR_PROJECT_ID * --tempLocation=gs://YOUR_TEMP_DIRECTORY - * --runner=BlockingDataflowPipelineRunner + * --runner=BlockingDataflowRunner * and an output prefix on GCS: * --output=gs://YOUR_OUTPUT_PREFIX * http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java index 1c6a6e1..a669b0c 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/FilterExamples.java @@ -73,7 +73,7 @@ import java.util.logging.Logger; * 

{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and the BigQuery table for the output: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java index 8e642f8..e8f1f01 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/JoinExamples.java @@ -56,7 +56,7 @@ import com.google.api.services.bigquery.model.TableRow; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and an output prefix on GCS: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java index 2480967..6630f3d 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java +++ b/examples/java/src/main/java/org/apache/beam/examples/cookbook/MaxPerKeyExamples.java @@ -62,7 +62,7 @@ import java.util.List; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  * }
  *
* and the BigQuery table for the output: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java ---------------------------------------------------------------------- diff --git a/examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java b/examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java index 0ad1a04..cf3bbf9 100644 --- a/examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java +++ b/examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java @@ -17,7 +17,7 @@ */ package org.apache.beam.examples; -import org.apache.beam.runners.dataflow.BlockingDataflowPipelineRunner; +import org.apache.beam.runners.dataflow.BlockingDataflowRunner; import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.io.TextIO; @@ -42,7 +42,7 @@ public class MinimalWordCountJava8 { DataflowPipelineOptions options = PipelineOptionsFactory.create() .as(DataflowPipelineOptions.class); - options.setRunner(BlockingDataflowPipelineRunner.class); + options.setRunner(BlockingDataflowRunner.class); // CHANGE 1 of 3: Your project ID is required in order to run your pipeline on the Google Cloud. options.setProject("SET_YOUR_PROJECT_ID_HERE"); http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java ---------------------------------------------------------------------- diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java index c557913..c8bcc8c 100644 --- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java +++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java @@ -83,7 +83,7 @@ import java.util.TimeZone; * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  *   --dataset=YOUR-DATASET
  *   --topic=projects/YOUR-PROJECT/topics/YOUR-TOPIC
  * }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java8/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
----------------------------------------------------------------------
diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
index ba3983d..c5c2fb5 100644
--- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
+++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/HourlyTeamScore.java
@@ -60,7 +60,7 @@ import java.util.TimeZone;
  * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  *   --dataset=YOUR-DATASET
  * }
  * 


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java8/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
----------------------------------------------------------------------
diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
index d79403a..12d2729 100644
--- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
+++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/LeaderBoard.java
@@ -80,7 +80,7 @@ import java.util.TimeZone;
  * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  *   --dataset=YOUR-DATASET
  *   --topic=projects/YOUR-PROJECT/topics/YOUR-TOPIC
  * }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
----------------------------------------------------------------------
diff --git a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
index de049e8..97b6929 100644
--- a/examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
+++ b/examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java
@@ -63,7 +63,7 @@ import java.util.Map;
  * 
{@code
  *   --project=YOUR_PROJECT_ID
  *   --tempLocation=gs://YOUR_TEMP_DIRECTORY
- *   --runner=BlockingDataflowPipelineRunner
+ *   --runner=BlockingDataflowRunner
  *   --dataset=YOUR-DATASET
  * }
  * 


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java
index 220e8e8..46a4fce 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java
@@ -145,7 +145,7 @@ public class FlinkPipelineRunner extends PipelineRunner {
 
   @Override
   public String toString() {
-    return "DataflowPipelineRunner#" + hashCode();
+    return "DataflowRunner#" + hashCode();
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowPipelineRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowPipelineRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowPipelineRunner.java
deleted file mode 100644
index d8ee16a..0000000
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowPipelineRunner.java
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you 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 org.apache.beam.runners.dataflow;
-
-import org.apache.beam.runners.dataflow.options.BlockingDataflowPipelineOptions;
-import org.apache.beam.runners.dataflow.util.MonitoringUtil;
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.PipelineResult.State;
-import org.apache.beam.sdk.annotations.Experimental;
-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.transforms.PTransform;
-import org.apache.beam.sdk.values.PInput;
-import org.apache.beam.sdk.values.POutput;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.concurrent.TimeUnit;
-
-import javax.annotation.Nullable;
-
-/**
- * A {@link PipelineRunner} that's like {@link DataflowPipelineRunner}
- * but that waits for the launched job to finish.
- *
- * 
Prints out job status updates and console messages while it waits.
- *
- * 
Returns the final job state, or throws an exception if the job
- * fails or cannot be monitored.
- *
- * 
Permissions

- * When reading from a Dataflow source or writing to a Dataflow sink using
- * {@code BlockingDataflowPipelineRunner}, the Google cloud services account and the Google compute
- * engine service account of the GCP project running the Dataflow Job will need access to the
- * corresponding source/sink.
- *
- * 
Please see Google Cloud
- * Dataflow Security and Permissions for more details.
- */
-public class BlockingDataflowPipelineRunner extends
-    PipelineRunner {
-  private static final Logger LOG = LoggerFactory.getLogger(BlockingDataflowPipelineRunner.class);
-
-  // Defaults to an infinite wait period.
-  // TODO: make this configurable after removal of option map.
-  private static final long BUILTIN_JOB_TIMEOUT_SEC = -1L;
-
-  private final DataflowPipelineRunner dataflowPipelineRunner;
-  private final BlockingDataflowPipelineOptions options;
-
-  protected BlockingDataflowPipelineRunner(
-      DataflowPipelineRunner internalRunner,
-      BlockingDataflowPipelineOptions options) {
-    this.dataflowPipelineRunner = internalRunner;
-    this.options = options;
-  }
-
-  /**
-   * Constructs a runner from the provided options.
-   */
-  public static BlockingDataflowPipelineRunner fromOptions(
-      PipelineOptions options) {
-    BlockingDataflowPipelineOptions dataflowOptions =
-        PipelineOptionsValidator.validate(BlockingDataflowPipelineOptions.class, options);
-    DataflowPipelineRunner dataflowPipelineRunner =
-        DataflowPipelineRunner.fromOptions(dataflowOptions);
-
-    return new BlockingDataflowPipelineRunner(dataflowPipelineRunner, dataflowOptions);
-  }
-
-  /**
-   * {@inheritDoc}
-   *
-   * @throws DataflowJobExecutionException if there is an exception during job execution.
-   * @throws DataflowServiceException if there is an exception retrieving information about the job.
-   */
-  @Override
-  public DataflowPipelineJob run(Pipeline p) {
-    final DataflowPipelineJob job = dataflowPipelineRunner.run(p);
-
-    // We ignore the potential race condition here (Ctrl-C after job submission but before the
-    // shutdown hook is registered). Even if we tried to do something smarter (eg., SettableFuture)
-    // the run method (which produces the job) could fail or be Ctrl-C'd before it had returned a
-    // job. The display of the command to cancel the job is best-effort anyways -- RPC's could fail,
-    // etc. If the user wants to verify the job was cancelled they should look at the job status.
-    Thread shutdownHook = new Thread() {
-      @Override
-      public void run() {
-        LOG.warn("Job is already running in Google Cloud Platform, Ctrl-C will not cancel it.\n"
-            + "To cancel the job in the cloud, run:\n> {}",
-            MonitoringUtil.getGcloudCancelCommand(options, job.getJobId()));
-      }
-    };
-
-    try {
-      Runtime.getRuntime().addShutdownHook(shutdownHook);
-
-      @Nullable
-      State result;
-      try {
-        result = job.waitToFinish(
-            BUILTIN_JOB_TIMEOUT_SEC, TimeUnit.SECONDS,
-            new MonitoringUtil.PrintHandler(options.getJobMessageOutput()));
-      } catch (IOException | InterruptedException ex) {
-        if (ex instanceof InterruptedException) {
-          Thread.currentThread().interrupt();
-        }
-        LOG.debug("Exception caught while retrieving status for job {}", job.getJobId(), ex);
-        throw new DataflowServiceException(
-            job, "Exception caught while retrieving status for job " + job.getJobId(), ex);
-      }
-
-      if (result == null) {
-        throw new DataflowServiceException(
-            job, "Timed out while retrieving status for job " + job.getJobId());
-      }
-
-      LOG.info("Job finished with status {}", result);
-      if (!result.isTerminal()) {
-        throw new IllegalStateException("Expected terminal state for job " + job.getJobId()
-            + ", got " + result);
-      }
-
-      if (result == State.DONE) {
-        return job;
-      } else if (result == State.UPDATED) {
-        DataflowPipelineJob newJob = job.getReplacedByJob();
-        LOG.info("Job {} has been updated and is running as the new job with id {}."
-            + "To access the updated job on the Dataflow monitoring console, please navigate to {}",
-            job.getJobId(),
-            newJob.getJobId(),
-            MonitoringUtil.getJobMonitoringPageURL(newJob.getProjectId(), newJob.getJobId()));
-        throw new DataflowJobUpdatedException(
-            job,
-            String.format("Job %s updated; new job is %s.", job.getJobId(), newJob.getJobId()),
-            newJob);
-      } else if (result == State.CANCELLED) {
-        String message = String.format("Job %s cancelled by user", job.getJobId());
-        LOG.info(message);
-        throw new DataflowJobCancelledException(job, message);
-      } else {
-        throw new DataflowJobExecutionException(job, "Job " + job.getJobId()
-            + " failed with status " + result);
-      }
-    } finally {
-      Runtime.getRuntime().removeShutdownHook(shutdownHook);
-    }
-  }
-
-  @Override
-  public  OutputT apply(
-      PTransform transform, InputT input) {
-    return dataflowPipelineRunner.apply(transform, input);
-  }
-
-  /**
-   * Sets callbacks to invoke during execution. See {@link DataflowPipelineRunnerHooks}.
-   */
-  @Experimental
-  public void setHooks(DataflowPipelineRunnerHooks hooks) {
-    this.dataflowPipelineRunner.setHooks(hooks);
-  }
-
-  @Override
-  public String toString() {
-    return "BlockingDataflowPipelineRunner#" + options.getJobName();
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowRunner.java
new file mode 100644
index 0000000..5c59bc2
--- /dev/null
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BlockingDataflowRunner.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.beam.runners.dataflow;
+
+import org.apache.beam.runners.dataflow.options.BlockingDataflowPipelineOptions;
+import org.apache.beam.runners.dataflow.util.MonitoringUtil;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult.State;
+import org.apache.beam.sdk.annotations.Experimental;
+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.transforms.PTransform;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import javax.annotation.Nullable;
+
+/**
+ * A {@link PipelineRunner} that's like {@link DataflowRunner}
+ * but that waits for the launched job to finish.
+ *
+ * 
Prints out job status updates and console messages while it waits.
+ *
+ * 
Returns the final job state, or throws an exception if the job
+ * fails or cannot be monitored.
+ *
+ * 
Permissions

+ * When reading from a Dataflow source or writing to a Dataflow sink using
+ * {@code BlockingDataflowRunner}, the Google cloud services account and the Google compute
+ * engine service account of the GCP project running the Dataflow Job will need access to the
+ * corresponding source/sink.
+ *
+ * 
Please see Google Cloud
+ * Dataflow Security and Permissions for more details.
+ */
+public class BlockingDataflowRunner extends
+    PipelineRunner {
+  private static final Logger LOG = LoggerFactory.getLogger(BlockingDataflowRunner.class);
+
+  // Defaults to an infinite wait period.
+  // TODO: make this configurable after removal of option map.
+  private static final long BUILTIN_JOB_TIMEOUT_SEC = -1L;
+
+  private final DataflowRunner dataflowRunner;
+  private final BlockingDataflowPipelineOptions options;
+
+  protected BlockingDataflowRunner(
+      DataflowRunner internalRunner,
+      BlockingDataflowPipelineOptions options) {
+    this.dataflowRunner = internalRunner;
+    this.options = options;
+  }
+
+  /**
+   * Constructs a runner from the provided options.
+   */
+  public static BlockingDataflowRunner fromOptions(
+      PipelineOptions options) {
+    BlockingDataflowPipelineOptions dataflowOptions =
+        PipelineOptionsValidator.validate(BlockingDataflowPipelineOptions.class, options);
+    DataflowRunner dataflowRunner =
+        DataflowRunner.fromOptions(dataflowOptions);
+
+    return new BlockingDataflowRunner(dataflowRunner, dataflowOptions);
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @throws DataflowJobExecutionException if there is an exception during job execution.
+   * @throws DataflowServiceException if there is an exception retrieving information about the job.
+   */
+  @Override
+  public DataflowPipelineJob run(Pipeline p) {
+    final DataflowPipelineJob job = dataflowRunner.run(p);
+
+    // We ignore the potential race condition here (Ctrl-C after job submission but before the
+    // shutdown hook is registered). Even if we tried to do something smarter (eg., SettableFuture)
+    // the run method (which produces the job) could fail or be Ctrl-C'd before it had returned a
+    // job. The display of the command to cancel the job is best-effort anyways -- RPC's could fail,
+    // etc. If the user wants to verify the job was cancelled they should look at the job status.
+    Thread shutdownHook = new Thread() {
+      @Override
+      public void run() {
+        LOG.warn("Job is already running in Google Cloud Platform, Ctrl-C will not cancel it.\n"
+            + "To cancel the job in the cloud, run:\n> {}",
+            MonitoringUtil.getGcloudCancelCommand(options, job.getJobId()));
+      }
+    };
+
+    try {
+      Runtime.getRuntime().addShutdownHook(shutdownHook);
+
+      @Nullable
+      State result;
+      try {
+        result = job.waitToFinish(
+            BUILTIN_JOB_TIMEOUT_SEC, TimeUnit.SECONDS,
+            new MonitoringUtil.PrintHandler(options.getJobMessageOutput()));
+      } catch (IOException | InterruptedException ex) {
+        if (ex instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+        }
+        LOG.debug("Exception caught while retrieving status for job {}", job.getJobId(), ex);
+        throw new DataflowServiceException(
+            job, "Exception caught while retrieving status for job " + job.getJobId(), ex);
+      }
+
+      if (result == null) {
+        throw new DataflowServiceException(
+            job, "Timed out while retrieving status for job " + job.getJobId());
+      }
+
+      LOG.info("Job finished with status {}", result);
+      if (!result.isTerminal()) {
+        throw new IllegalStateException("Expected terminal state for job " + job.getJobId()
+            + ", got " + result);
+      }
+
+      if (result == State.DONE) {
+        return job;
+      } else if (result == State.UPDATED) {
+        DataflowPipelineJob newJob = job.getReplacedByJob();
+        LOG.info("Job {} has been updated and is running as the new job with id {}."
+            + "To access the updated job on the Dataflow monitoring console, please navigate to {}",
+            job.getJobId(),
+            newJob.getJobId(),
+            MonitoringUtil.getJobMonitoringPageURL(newJob.getProjectId(), newJob.getJobId()));
+        throw new DataflowJobUpdatedException(
+            job,
+            String.format("Job %s updated; new job is %s.", job.getJobId(), newJob.getJobId()),
+            newJob);
+      } else if (result == State.CANCELLED) {
+        String message = String.format("Job %s cancelled by user", job.getJobId());
+        LOG.info(message);
+        throw new DataflowJobCancelledException(job, message);
+      } else {
+        throw new DataflowJobExecutionException(job, "Job " + job.getJobId()
+            + " failed with status " + result);
+      }
+    } finally {
+      Runtime.getRuntime().removeShutdownHook(shutdownHook);
+    }
+  }
+
+  @Override
+  public  OutputT apply(
+      PTransform transform, InputT input) {
+    return dataflowRunner.apply(transform, input);
+  }
+
+  /**
+   * Sets callbacks to invoke during execution. See {@link DataflowRunnerHooks}.
+   */
+  @Experimental
+  public void setHooks(DataflowRunnerHooks hooks) {
+    this.dataflowRunner.setHooks(hooks);
+  }
+
+  @Override
+  public String toString() {
+    return "BlockingDataflowRunner#" + options.getJobName();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobCancelledException.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobCancelledException.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobCancelledException.java
index de44a19..e2edb6a 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobCancelledException.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobCancelledException.java
@@ -18,7 +18,7 @@
 package org.apache.beam.runners.dataflow;
 
 /**
- * Signals that a job run by a {@link BlockingDataflowPipelineRunner} was updated during execution.
+ * Signals that a job run by a {@link BlockingDataflowRunner} was updated during execution.
  */
 public class DataflowJobCancelledException extends DataflowJobException {
   /**

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobExecutionException.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobExecutionException.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobExecutionException.java
index 11b8723..ccf8057 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobExecutionException.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobExecutionException.java
@@ -20,7 +20,7 @@ package org.apache.beam.runners.dataflow;
 import javax.annotation.Nullable;
 
 /**
- * Signals that a job run by a {@link BlockingDataflowPipelineRunner} fails during execution, and
+ * Signals that a job run by a {@link BlockingDataflowRunner} fails during execution, and
  * provides access to the failed job.
  */
 public class DataflowJobExecutionException extends DataflowJobException {

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobUpdatedException.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobUpdatedException.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobUpdatedException.java
index e30a0e7..39d1d47 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobUpdatedException.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowJobUpdatedException.java
@@ -18,7 +18,7 @@
 package org.apache.beam.runners.dataflow;
 
 /**
- * Signals that a job run by a {@link BlockingDataflowPipelineRunner} was updated during execution.
+ * Signals that a job run by a {@link BlockingDataflowRunner} was updated during execution.
  */
 public class DataflowJobUpdatedException extends DataflowJobException {
   private DataflowPipelineJob replacedByJob;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
index 0f42148..548d2b9 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java
@@ -55,7 +55,7 @@ import javax.annotation.Nullable;
 
 /**
  * A DataflowPipelineJob represents a job submitted to Dataflow using
- * {@link DataflowPipelineRunner}.
+ * {@link DataflowRunner}.
  */
 public class DataflowPipelineJob implements PipelineResult {
   private static final Logger LOG = LoggerFactory.getLogger(DataflowPipelineJob.class);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/6d028ac6/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineRegistrar.java
----------------------------------------------------------------------
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 d7d243f..4c88c4e 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
@@ -29,7 +29,7 @@ import com.google.common.collect.ImmutableList;
 
 /**
  * Contains the {@link PipelineOptionsRegistrar} and {@link PipelineRunnerRegistrar} for the
- * {@link DataflowPipelineRunner}.
+ * {@link DataflowRunner}.
  */
 public class DataflowPipelineRegistrar {
   private DataflowPipelineRegistrar() { }
@@ -48,15 +48,15 @@ public class DataflowPipelineRegistrar {
   }
 
   /**
-   * Register the {@link DataflowPipelineRunner} and {@link BlockingDataflowPipelineRunner}.
+   * Register the {@link DataflowRunner} and {@link BlockingDataflowRunner}.
    */
   @AutoService(PipelineRunnerRegistrar.class)
   public static class Runner implements PipelineRunnerRegistrar {
     @Override
     public Iterable>> getPipelineRunners() {
       return ImmutableList.>>of(
-          DataflowPipelineRunner.class,
-          BlockingDataflowPipelineRunner.class);
+          DataflowRunner.class,
+          BlockingDataflowRunner.class);
     }
   }
 }