beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [2/3] beam git commit: Handle Errors in the DirectRunner
Date Tue, 09 May 2017 18:49:25 GMT
Handle Errors in the DirectRunner

When a worker dies because of an error, propagate that error and fail
the Pipeline.


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

Branch: refs/heads/master
Commit: f03c04a787343c3710355c84a105582cdc815469
Parents: 28180c4
Author: Thomas Groh <tgroh@google.com>
Authored: Tue May 9 09:46:38 2017 -0700
Committer: Dan Halperin <dhalperi@google.com>
Committed: Tue May 9 11:49:14 2017 -0700

----------------------------------------------------------------------
 .../beam/runners/direct/CompletionCallback.java |  6 +++++
 .../direct/ExecutorServiceParallelExecutor.java | 27 +++++++++++++++-----
 .../beam/runners/direct/TransformExecutor.java  | 12 +++++++--
 .../apache/beam/runners/direct/MockClock.java   |  2 +-
 .../runners/direct/TransformExecutorTest.java   |  5 ++++
 5 files changed, 43 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/f03c04a7/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CompletionCallback.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CompletionCallback.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CompletionCallback.java
index 0af22c8..417fa09 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CompletionCallback.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/CompletionCallback.java
@@ -40,4 +40,10 @@ interface CompletionCallback {
    * Handle a result that terminated abnormally due to the provided {@link Exception}.
    */
   void handleException(CommittedBundle<?> inputBundle, Exception t);
+
+  /**
+   * Handle a result that terminated abnormally due to the provided {@link Error}. The pipeline
+   * should be shut down, and the Error propagated.
+  */
+  void handleError(Error err);
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/f03c04a7/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
index b7f4732..71ab4cc 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java
@@ -285,8 +285,15 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
         // there are no updates to process and no updates will ever be published because
the
         // executor is shutdown
         return pipelineState.get();
-      } else if (update != null && update.exception.isPresent()) {
-        throw update.exception.get();
+      } else if (update != null && update.thrown.isPresent()) {
+        Throwable thrown = update.thrown.get();
+        if (thrown instanceof Exception) {
+          throw (Exception) thrown;
+        } else if (thrown instanceof Error) {
+          throw (Error) thrown;
+        } else {
+          throw new Exception("Unknown Type of Throwable", thrown);
+        }
       }
     }
     return pipelineState.get();
@@ -380,6 +387,11 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
       allUpdates.offer(ExecutorUpdate.fromException(e));
       outstandingWork.decrementAndGet();
     }
+
+    @Override
+    public void handleError(Error err) {
+      visibleUpdates.add(VisibleExecutorUpdate.fromError(err));
+    }
   }
 
   /**
@@ -424,7 +436,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
    * return normally or throw an exception.
    */
   private static class VisibleExecutorUpdate {
-    private final Optional<? extends Exception> exception;
+    private final Optional<? extends Throwable> thrown;
     @Nullable
     private final State newState;
 
@@ -432,6 +444,10 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
       return new VisibleExecutorUpdate(null, e);
     }
 
+    public static VisibleExecutorUpdate fromError(Error err) {
+      return new VisibleExecutorUpdate(State.FAILED, err);
+    }
+
     public static VisibleExecutorUpdate finished() {
       return new VisibleExecutorUpdate(State.DONE, null);
     }
@@ -440,15 +456,14 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
       return new VisibleExecutorUpdate(State.CANCELLED, null);
     }
 
-    private VisibleExecutorUpdate(State newState, @Nullable Exception exception) {
-      this.exception = Optional.fromNullable(exception);
+    private VisibleExecutorUpdate(State newState, @Nullable Throwable exception) {
+      this.thrown = Optional.fromNullable(exception);
       this.newState = newState;
     }
 
     public State getNewState() {
       return newState;
     }
-
   }
 
   private class MonitorRunnable implements Runnable {

http://git-wip-us.apache.org/repos/asf/beam/blob/f03c04a7/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
index 8e1515b..56f8650 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformExecutor.java
@@ -26,6 +26,8 @@ import org.apache.beam.sdk.metrics.MetricsContainer;
 import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.runners.AppliedPTransform;
 import org.apache.beam.sdk.util.WindowedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link Callable} responsible for constructing a {@link TransformEvaluator} from a
@@ -36,6 +38,8 @@ import org.apache.beam.sdk.util.WindowedValue;
  * that it is being executed on.
  */
 class TransformExecutor<T> implements Runnable {
+  private static final Logger LOG = LoggerFactory.getLogger(TransformExecutor.class);
+
   public static <T> TransformExecutor<T> create(
       EvaluationContext context,
       TransformEvaluatorFactory factory,
@@ -112,6 +116,10 @@ class TransformExecutor<T> implements Runnable {
         throw (RuntimeException) e;
       }
       throw new RuntimeException(e);
+    } catch (Error err) {
+      LOG.error("Error occurred within {}", this, err);
+      onComplete.handleError(err);
+      throw err;
     } finally {
       // Report the physical metrics from the end of this step.
       context.getMetrics().commitPhysical(inputBundle, metricsContainer.getCumulative());
@@ -162,8 +170,8 @@ class TransformExecutor<T> implements Runnable {
       TransformEvaluator<T> evaluator, MetricsContainer metricsContainer,
       Collection<ModelEnforcement<T>> enforcements)
       throws Exception {
-    TransformResult<T> result = evaluator.finishBundle()
-        .withLogicalMetricUpdates(metricsContainer.getCumulative());
+    TransformResult<T> result =
+        evaluator.finishBundle().withLogicalMetricUpdates(metricsContainer.getCumulative());
     CommittedResult outputs = onComplete.handleResult(inputBundle, result);
     for (ModelEnforcement<T> enforcement : enforcements) {
       enforcement.afterFinish(inputBundle, result, outputs.getOutputs());

http://git-wip-us.apache.org/repos/asf/beam/blob/f03c04a7/runners/direct-java/src/test/java/org/apache/beam/runners/direct/MockClock.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/MockClock.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/MockClock.java
index 11ecbff..9275e3c 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/MockClock.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/MockClock.java
@@ -28,7 +28,7 @@ import org.joda.time.Instant;
  *
  * <p>For uses of the {@link Clock} interface in unit tests.
  */
-public class MockClock implements Clock {
+class MockClock implements Clock {
 
   private Instant now;
 

http://git-wip-us.apache.org/repos/asf/beam/blob/f03c04a7/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
index dc0ef7c..86412a0 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/TransformExecutorTest.java
@@ -434,6 +434,11 @@ public class TransformExecutorTest {
       handledException = e;
       onMethod.countDown();
     }
+
+    @Override
+    public void handleError(Error err) {
+      throw err;
+    }
   }
 
   private static class TestEnforcementFactory implements ModelEnforcementFactory {


Mime
View raw message