You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2017/05/09 18:49:25 UTC
[2/3] beam git commit: Handle Errors in the DirectRunner
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 <tg...@google.com>
Authored: Tue May 9 09:46:38 2017 -0700
Committer: Dan Halperin <dh...@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 {