beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [28/36] beam git commit: mr-runner: fail early in the runner when MapReduce job fails.
Date Thu, 07 Sep 2017 18:39:37 GMT
mr-runner: fail early in the runner when MapReduce job fails.


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/989d7d8e
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/989d7d8e
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/989d7d8e

Branch: refs/heads/mr-runner
Commit: 989d7d8e4eabe6aafd2008eed16f533ec75a43d1
Parents: 9f312c5
Author: Pei He <pei@apache.org>
Authored: Fri Sep 1 13:04:54 2017 +0800
Committer: Pei He <pei@apache.org>
Committed: Fri Sep 1 17:13:52 2017 +0800

----------------------------------------------------------------------
 .../java/org/apache/beam/runners/mapreduce/MapReduceRunner.java  | 4 ++++
 1 file changed, 4 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/989d7d8e/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReduceRunner.java
----------------------------------------------------------------------
diff --git a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReduceRunner.java
b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReduceRunner.java
index 71edf1a..8198848 100644
--- a/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReduceRunner.java
+++ b/runners/map-reduce/src/main/java/org/apache/beam/runners/mapreduce/MapReduceRunner.java
@@ -35,6 +35,7 @@ import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobStatus;
 import org.apache.log4j.BasicConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -95,6 +96,9 @@ public class MapReduceRunner extends PipelineRunner<PipelineResult>
{
       try {
         Job job = jobPrototype.build(options.getJarClass(), config);
         job.waitForCompletion(true);
+        if (!job.getStatus().getState().equals(JobStatus.State.SUCCEEDED)) {
+          throw new RuntimeException("MapReduce job failed: " + job.getJobID());
+        }
         jobs.add(job);
       } catch (Exception e) {
         Throwables.throwIfUnchecked(e);


Mime
View raw message