beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [2/3] beam git commit: fix TestFlinkRunner to throw PipelineExecutionException
Date Tue, 21 Feb 2017 10:05:07 GMT
fix TestFlinkRunner to throw PipelineExecutionException


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

Branch: refs/heads/master
Commit: 0e9173df7140f4a07086d83170c1ccdfd9412863
Parents: 8cfb3d1
Author: JingsongLi <lzljs3620320@aliyun.com>
Authored: Fri Feb 17 17:47:55 2017 +0800
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Tue Feb 21 09:53:05 2017 +0100

----------------------------------------------------------------------
 .../apache/beam/runners/flink/TestFlinkRunner.java   | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/0e9173df/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java
index 7ba98ab..30a94a1 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/TestFlinkRunner.java
@@ -18,12 +18,14 @@
 package org.apache.beam.runners.flink;
 
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
 import org.apache.beam.sdk.PipelineResult;
 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.transforms.PTransform;
+import org.apache.beam.sdk.util.UserCodeException;
 import org.apache.beam.sdk.values.PInput;
 import org.apache.beam.sdk.values.POutput;
 
@@ -68,7 +70,14 @@ public class TestFlinkRunner extends PipelineRunner<PipelineResult>
{
       // probably be a better story along the lines of UserCodeException.
       Throwable cause = e;
       Throwable oldCause = e;
+      PipelineExecutionException executionException = null;
       do {
+
+        // find UserCodeException and throw PipelineExecutionException
+        if (cause instanceof UserCodeException) {
+          executionException = new PipelineExecutionException(cause.getCause());
+        }
+
         if (cause.getCause() == null) {
           break;
         }
@@ -80,7 +89,11 @@ public class TestFlinkRunner extends PipelineRunner<PipelineResult>
{
       if (cause instanceof AssertionError) {
         throw (AssertionError) cause;
       } else {
-        throw e;
+        if (executionException != null) {
+          throw executionException;
+        } else {
+          throw e;
+        }
       }
     }
   }


Mime
View raw message