beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/4] incubator-beam git commit: Add handleEmpty to CompletionCallback
Date Thu, 04 Aug 2016 00:46:01 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 636b5f7f6 -> 23fa61ce0


Add handleEmpty to CompletionCallback

This is invoked when a Transform Executor has no work to do. Usually
this is due to reinvocation of a Source.


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

Branch: refs/heads/master
Commit: a8eb274aca3e4ab2b89a3e3d44bb2c755fd638eb
Parents: 8b1e64a
Author: Thomas Groh <tgroh@google.com>
Authored: Fri Jul 22 18:01:41 2016 -0700
Committer: Thomas Groh <tgroh@google.com>
Committed: Fri Jul 29 16:00:47 2016 -0700

----------------------------------------------------------------------
 .../beam/runners/direct/CompletionCallback.java |  5 ++
 .../direct/ExecutorServiceParallelExecutor.java | 57 +++++---------------
 .../beam/runners/direct/TransformExecutor.java  |  1 +
 .../runners/direct/TransformExecutorTest.java   | 27 ++++++++++
 4 files changed, 47 insertions(+), 43 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a8eb274a/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 0c5fe24..2f496e9 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
@@ -30,6 +30,11 @@ interface CompletionCallback {
       CommittedBundle<?> inputBundle, TransformResult result);
 
   /**
+   * Handle an input bundle that did not require processing.
+   */
+  void handleEmpty(CommittedBundle<?> inputBundle);
+
+  /**
    * Handle a result that terminated abnormally due to the provided {@link Throwable}.
    */
   void handleThrowable(CommittedBundle<?> inputBundle, Throwable t);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a8eb274a/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 43195e3..3901472 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
@@ -122,7 +122,8 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
     this.visibleUpdates = new ArrayBlockingQueue<>(20);
 
     parallelExecutorService = TransformExecutorServices.parallel(executorService);
-    defaultCompletionCallback = new DefaultCompletionCallback();
+    defaultCompletionCallback =
+        new TimerIterableCompletionCallback(Collections.<TimerData>emptyList());
   }
 
   private CacheLoader<StepAndKey, TransformExecutorService>
@@ -217,18 +218,19 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
   /**
    * The base implementation of {@link CompletionCallback} that provides implementations
for
    * {@link #handleResult(CommittedBundle, TransformResult)} and
-   * {@link #handleThrowable(CommittedBundle, Throwable)}, given an implementation of
-   * {@link #getCommittedResult(CommittedBundle, TransformResult)}.
+   * {@link #handleThrowable(CommittedBundle, Throwable)}.
    */
-  private abstract class CompletionCallbackBase implements CompletionCallback {
-    protected abstract CommittedResult getCommittedResult(
-        CommittedBundle<?> inputBundle,
-        TransformResult result);
+  private class TimerIterableCompletionCallback implements CompletionCallback {
+    private final Iterable<TimerData> timers;
+
+    protected TimerIterableCompletionCallback(Iterable<TimerData> timers) {
+      this.timers = timers;
+    }
 
     @Override
     public final CommittedResult handleResult(
         CommittedBundle<?> inputBundle, TransformResult result) {
-      CommittedResult committedResult = getCommittedResult(inputBundle, result);
+      CommittedResult committedResult = evaluationContext.handleResult(inputBundle, timers,
result);
       for (CommittedBundle<?> outputBundle : committedResult.getOutputs()) {
         allUpdates.offer(ExecutorUpdate.fromBundle(outputBundle,
             valueToConsumers.get(outputBundle.getPCollection())));
@@ -242,43 +244,12 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
     }
 
     @Override
-    public final void handleThrowable(CommittedBundle<?> inputBundle, Throwable t)
{
-      allUpdates.offer(ExecutorUpdate.fromThrowable(t));
-    }
-  }
-
-  /**
-   * The default {@link CompletionCallback}. The default completion callback is used to complete
-   * transform evaluations that are triggered due to the arrival of elements from an upstream
-   * transform, or for a source transform.
-   */
-  private class DefaultCompletionCallback extends CompletionCallbackBase {
-    @Override
-    public CommittedResult getCommittedResult(
-        CommittedBundle<?> inputBundle, TransformResult result) {
-      return evaluationContext.handleResult(inputBundle,
-          Collections.<TimerData>emptyList(),
-          result);
-    }
-  }
-
-  /**
-   * A {@link CompletionCallback} where the completed bundle was produced to deliver some
collection
-   * of {@link TimerData timers}. When the evaluator completes successfully, reports all
of the
-   * timers used to create the input to the {@link EvaluationContext evaluation context}
-   * as part of the result.
-   */
-  private class TimerCompletionCallback extends CompletionCallbackBase {
-    private final Iterable<TimerData> timers;
-
-    private TimerCompletionCallback(Iterable<TimerData> timers) {
-      this.timers = timers;
+    public void handleEmpty(CommittedBundle<?> inputBundle) {
     }
 
     @Override
-    public CommittedResult getCommittedResult(
-        CommittedBundle<?> inputBundle, TransformResult result) {
-          return evaluationContext.handleResult(inputBundle, timers, result);
+    public final void handleThrowable(CommittedBundle<?> inputBundle, Throwable t)
{
+      allUpdates.offer(ExecutorUpdate.fromThrowable(t));
     }
   }
 
@@ -424,7 +395,7 @@ final class ExecutorServiceParallelExecutor implements PipelineExecutor
{
                           null, keyTimers.getKey(), (PCollection) transform.getInput())
                       .add(WindowedValue.valueInEmptyWindows(work))
                       .commit(Instant.now());
-              scheduleConsumption(transform, bundle, new TimerCompletionCallback(delivery));
+              scheduleConsumption(transform, bundle, new TimerIterableCompletionCallback(delivery));
               firedTimers = true;
             }
           }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a8eb274a/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 ef31ba7..793b508 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
@@ -111,6 +111,7 @@ class TransformExecutor<T> implements Runnable {
       TransformEvaluator<T> evaluator =
           evaluatorFactory.forApplication(transform, inputBundle, evaluationContext);
       if (evaluator == null) {
+        onComplete.handleEmpty(inputBundle);
         // Nothing to do
         return;
       }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a8eb274a/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 cb5cd46..95206f3 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
@@ -134,6 +134,26 @@ public class TransformExecutorTest {
   }
 
   @Test
+  public void nullTransformEvaluatorTerminates() throws Exception {
+    when(registry.forApplication(created.getProducingTransformInternal(),
+        null,
+        evaluationContext)).thenReturn(null);
+
+    TransformExecutor<Object> executor = TransformExecutor.create(registry,
+        Collections.<ModelEnforcementFactory>emptyList(),
+        evaluationContext,
+        null,
+        created.getProducingTransformInternal(),
+        completionCallback,
+        transformEvaluationState);
+    executor.run();
+
+    assertThat(completionCallback.handledResult, is(nullValue()));
+    assertThat(completionCallback.handledEmpty, equalTo(true));
+    assertThat(completionCallback.handledThrowable, is(nullValue()));
+  }
+
+  @Test
   public void inputBundleProcessesEachElementFinishesAndCompletes() throws Exception {
     final TransformResult result =
         StepTransformResult.withoutHold(downstream.getProducingTransformInternal()).build();
@@ -471,6 +491,7 @@ public class TransformExecutorTest {
 
   private static class RegisteringCompletionCallback implements CompletionCallback {
     private TransformResult handledResult = null;
+    private boolean handledEmpty = false;
     private Throwable handledThrowable = null;
     private final CountDownLatch onMethod;
 
@@ -496,6 +517,12 @@ public class TransformExecutorTest {
     }
 
     @Override
+    public void handleEmpty(CommittedBundle<?> inputBundle) {
+      handledEmpty = true;
+      onMethod.countDown();
+    }
+
+    @Override
     public void handleThrowable(CommittedBundle<?> inputBundle, Throwable t) {
       handledThrowable = t;
       onMethod.countDown();


Mime
View raw message