beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lc...@apache.org
Subject [1/3] incubator-beam git commit: Replace DirectResult#awaitCompletion with waitUntilFinish
Date Sat, 15 Oct 2016 01:08:35 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master c2c650a63 -> b8e6eea69


Replace DirectResult#awaitCompletion with waitUntilFinish

This is the PipelineResult interface method, rather than the
DirectRunner-specific method.


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

Branch: refs/heads/master
Commit: 06bd0747ab35c2c10e44638dac60279ed870136a
Parents: 5e51c84
Author: Thomas Groh <tgroh@google.com>
Authored: Thu Oct 13 15:07:51 2016 -0700
Committer: Luke Cwik <lcwik@google.com>
Committed: Fri Oct 14 17:21:18 2016 -0700

----------------------------------------------------------------------
 .../beam/runners/direct/DirectOptions.java      |  4 +--
 .../beam/runners/direct/DirectRunner.java       | 27 ++++++++++++--------
 .../beam/runners/direct/DirectRunnerTest.java   |  6 ++---
 3 files changed, 22 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/06bd0747/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java
index b2c4f47..32ef352 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectOptions.java
@@ -42,8 +42,8 @@ public interface DirectOptions extends PipelineOptions, ApplicationNameOptions
{
   @Description(
       "If the pipeline should block awaiting completion of the pipeline. If set to true,
"
           + "a call to Pipeline#run() will block until all PTransforms are complete. Otherwise,
"
-          + "the Pipeline will execute asynchronously. If set to false, the completion of
the "
-          + "pipeline can be awaited on by use of DirectPipelineResult#awaitCompletion().")
+          + "the Pipeline will execute asynchronously. If set to false, use "
+          + "PipelineResult#waitUntilFinish() to block until the Pipeline is complete.")
   boolean isBlockOnRun();
 
   void setBlockOnRun(boolean b);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/06bd0747/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
index 664a915..36d19cf 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java
@@ -274,7 +274,7 @@ public class DirectRunner
     DirectPipelineResult result = new DirectPipelineResult(executor, context, aggregatorSteps);
     if (options.isBlockOnRun()) {
       try {
-        result.awaitCompletion();
+        result.waitUntilFinish();
       } catch (UserCodeException userException) {
         throw new PipelineExecutionException(userException.getCause());
       } catch (Throwable t) {
@@ -403,10 +403,21 @@ public class DirectRunner
      *
      * <p>See also {@link PipelineExecutor#awaitCompletion()}.
      */
-    public State awaitCompletion() throws Exception {
+    @Override
+    public State waitUntilFinish() {
       if (!state.isTerminal()) {
-        executor.awaitCompletion();
-        state = State.DONE;
+        try {
+          executor.awaitCompletion();
+          state = State.DONE;
+        } catch (Exception e) {
+          if (e instanceof InterruptedException) {
+            Thread.currentThread().interrupt();
+          }
+          if (e instanceof RuntimeException) {
+            throw (RuntimeException) e;
+          }
+          throw new RuntimeException(e);
+        }
       }
       return state;
     }
@@ -417,14 +428,10 @@ public class DirectRunner
     }
 
     @Override
-    public State waitUntilFinish() throws IOException {
-      return waitUntilFinish(Duration.millis(-1));
-    }
-
-    @Override
     public State waitUntilFinish(Duration duration) throws IOException {
       throw new UnsupportedOperationException(
-          "DirectPipelineResult does not support waitUntilFinish.");
+          "DirectPipelineResult does not support waitUntilFinish with a Duration parameter.
See"
+              + " BEAM-596.");
     }
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/06bd0747/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
index d93dd7a..37af90c 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/DirectRunnerTest.java
@@ -126,7 +126,7 @@ public class DirectRunnerTest implements Serializable {
     PAssert.that(countStrs).containsInAnyOrder("baz: 1", "bar: 2", "foo: 3");
 
     DirectPipelineResult result = ((DirectPipelineResult) p.run());
-    result.awaitCompletion();
+    result.waitUntilFinish();
   }
 
   private static AtomicInteger changed;
@@ -164,10 +164,10 @@ public class DirectRunnerTest implements Serializable {
     PAssert.that(countStrs).containsInAnyOrder("baz: 1", "bar: 2", "foo: 3");
 
     DirectPipelineResult result = ((DirectPipelineResult) p.run());
-    result.awaitCompletion();
+    result.waitUntilFinish();
 
     DirectPipelineResult otherResult = ((DirectPipelineResult) p.run());
-    otherResult.awaitCompletion();
+    otherResult.waitUntilFinish();
 
     assertThat("Each element should have been processed twice", changed.get(), equalTo(6));
   }


Mime
View raw message