beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/2] beam git commit: Allow absolute timers
Date Mon, 06 Feb 2017 23:32:15 GMT
Repository: beam
Updated Branches:
  refs/heads/master f4e109767 -> 0c24286e1


Allow absolute timers


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

Branch: refs/heads/master
Commit: a93c5c0594dcd4519fcf4b842f2fe0b8244a81a3
Parents: f4e1097
Author: Kenneth Knowles <klk@google.com>
Authored: Mon Jan 23 20:50:50 2017 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Mon Feb 6 15:17:23 2017 -0800

----------------------------------------------------------------------
 .../beam/runners/core/SimpleDoFnRunner.java     | 83 +++++++++++++++--
 .../java/org/apache/beam/sdk/util/Timer.java    | 11 +++
 .../apache/beam/sdk/transforms/ParDoTest.java   | 93 ++++++++++++++++++++
 3 files changed, 178 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/a93c5c05/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
index 8c9b8b7..7a89389 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.core;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.common.collect.Iterables;
@@ -94,6 +95,8 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
 
   private final Coder<BoundedWindow> windowCoder;
 
+  private final Duration allowedLateness;
+
   // Because of setKey(Object), we really must refresh stateInternals() at each access
   private final StepContext stepContext;
 
@@ -121,6 +124,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     Coder<BoundedWindow> untypedCoder =
         (Coder<BoundedWindow>) windowingStrategy.getWindowFn().windowCoder();
     this.windowCoder = untypedCoder;
+    this.allowedLateness = windowingStrategy.getAllowedLateness();
 
     this.context =
         new DoFnContext<>(
@@ -182,7 +186,8 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     }
 
     OnTimerArgumentProvider<InputT, OutputT> argumentProvider =
-        new OnTimerArgumentProvider<>(fn, context, window, effectiveTimestamp, timeDomain);
+        new OnTimerArgumentProvider<>(
+            fn, context, window, allowedLateness, effectiveTimestamp, timeDomain);
     invoker.invokeOnTimer(timerId, argumentProvider);
   }
 
@@ -210,7 +215,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
 
   /** Returns a new {@link DoFn.ProcessContext} for the given element. */
   private DoFnProcessContext<InputT, OutputT> createProcessContext(WindowedValue<InputT>
elem) {
-    return new DoFnProcessContext<InputT, OutputT>(fn, context, elem);
+    return new DoFnProcessContext<InputT, OutputT>(fn, context, elem, allowedLateness);
   }
 
   private RuntimeException wrapUserCodeException(Throwable t) {
@@ -465,6 +470,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     final DoFn<InputT, OutputT> fn;
     final DoFnContext<InputT, OutputT> context;
     final WindowedValue<InputT> windowedValue;
+    private final Duration allowedLateness;
 
     /** Lazily initialized; should only be accessed via {@link #getNamespace()}. */
     @Nullable private StateNamespace namespace;
@@ -486,11 +492,13 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     private DoFnProcessContext(
         DoFn<InputT, OutputT> fn,
         DoFnContext<InputT, OutputT> context,
-        WindowedValue<InputT> windowedValue) {
+        WindowedValue<InputT> windowedValue,
+        Duration allowedLateness) {
       fn.super();
       this.fn = fn;
       this.context = context;
       this.windowedValue = windowedValue;
+      this.allowedLateness = allowedLateness;
     }
 
     @Override
@@ -633,7 +641,8 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
       try {
         TimerSpec spec =
             (TimerSpec) signature.timerDeclarations().get(timerId).field().get(fn);
-        return new TimerInternalsTimer(getNamespace(), timerId, spec, stepContext.timerInternals());
+        return new TimerInternalsTimer(
+            window(), getNamespace(), allowedLateness, timerId, spec, stepContext.timerInternals());
       } catch (IllegalAccessException e) {
         throw new RuntimeException(e);
       }
@@ -656,6 +665,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     private final BoundedWindow window;
     private final Instant timestamp;
     private final TimeDomain timeDomain;
+    private final Duration allowedLateness;
 
     /** Lazily initialized; should only be accessed via {@link #getNamespace()}. */
     private StateNamespace namespace;
@@ -678,12 +688,14 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
         DoFn<InputT, OutputT> fn,
         DoFnContext<InputT, OutputT> context,
         BoundedWindow window,
+        Duration allowedLateness,
         Instant timestamp,
         TimeDomain timeDomain) {
       fn.super();
       this.fn = fn;
       this.context = context;
       this.window = window;
+      this.allowedLateness = allowedLateness;
       this.timestamp = timestamp;
       this.timeDomain = timeDomain;
     }
@@ -741,7 +753,8 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
       try {
         TimerSpec spec =
             (TimerSpec) signature.timerDeclarations().get(timerId).field().get(fn);
-        return new TimerInternalsTimer(getNamespace(), timerId, spec, stepContext.timerInternals());
+        return new TimerInternalsTimer(
+            window, getNamespace(), allowedLateness, timerId, spec, stepContext.timerInternals());
       } catch (IllegalAccessException e) {
         throw new RuntimeException(e);
       }
@@ -782,12 +795,25 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
 
   private static class TimerInternalsTimer implements Timer {
     private final TimerInternals timerInternals;
+
+    // The window and the namespace represent the same thing, but the namespace is a cached
+    // and specially encoded form. Since the namespace can be cached across timers, it is
+    // passed in whole rather than being computed here.
+    private final BoundedWindow window;
+    private final Duration allowedLateness;
+    private final StateNamespace namespace;
     private final String timerId;
     private final TimerSpec spec;
-    private final StateNamespace namespace;
 
     public TimerInternalsTimer(
-        StateNamespace namespace, String timerId, TimerSpec spec, TimerInternals timerInternals)
{
+        BoundedWindow window,
+        StateNamespace namespace,
+        Duration allowedLateness,
+        String timerId,
+        TimerSpec spec,
+        TimerInternals timerInternals) {
+      this.window = window;
+      this.allowedLateness = allowedLateness;
       this.namespace = namespace;
       this.timerId = timerId;
       this.spec = spec;
@@ -795,9 +821,48 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
     }
 
     @Override
+    public void set(Instant target) {
+      verifyAbsoluteTimeDomain();
+      verifyTargetTime(target);
+      setUnderlyingTimer(target);
+    }
+
+    @Override
     public void setForNowPlus(Duration durationFromNow) {
-      timerInternals.setTimer(
-          namespace, timerId, getCurrentTime().plus(durationFromNow), spec.getTimeDomain());
+      Instant target = getCurrentTime().plus(durationFromNow);
+      verifyTargetTime(target);
+      setUnderlyingTimer(target);
+    }
+
+    /**
+     * Ensures that the target time is reasonable. For event time timers this means that
the
+     * time should be prior to window GC time.
+     */
+    private void verifyTargetTime(Instant target) {
+      if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
+        Instant windowExpiry = window.maxTimestamp().plus(allowedLateness);
+        checkArgument(!target.isAfter(windowExpiry),
+            "Attempted to set event time timer for %s but that is after"
+            + " the expiration of window %s", target, windowExpiry);
+      }
+    }
+
+    /** Verifies that the time domain of this timer is acceptable for absolute timers. */
+    private void verifyAbsoluteTimeDomain() {
+      if (!TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
+        throw new IllegalStateException(
+            "Cannot only set relative timers in processing time domain."
+                + " Use #setForNowPlus(Duration)");
+      }
+    }
+
+    /**
+     * Sets the timer for the target time without checking anything about whether it is
+     * a reasonable thing to do. For example, absolute processing time timers are not
+     * really sensible since the user has no way to compute a good choice of time.
+     */
+    private void setUnderlyingTimer(Instant target) {
+      timerInternals.setTimer(namespace, timerId, target, spec.getTimeDomain());
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/a93c5c05/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
index 556287d..45a2a66 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
@@ -20,6 +20,7 @@ package org.apache.beam.sdk.util;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.joda.time.Duration;
+import org.joda.time.Instant;
 
 /**
  * A timer for a specified time domain that can be set to register the desire for further
processing
@@ -43,6 +44,16 @@ import org.joda.time.Duration;
 @Experimental(Experimental.Kind.TIMERS)
 public interface Timer {
   /**
+   * Sets or resets the time in the timer's {@link TimeDomain} at which it should fire. If
the timer
+   * was already set, resets it to the new requested time.
+   *
+   * <p>For {@link TimeDomain#PROCESSING_TIME}, the behavior is be unpredictable, since
processing
+   * time timers are ignored after a window has expired. Instead, it is recommended to use
+   * {@link #setForNowPlus(Duration)}.
+   */
+  void set(Instant instant);
+
+  /**
    * Sets or resets the time relative to the current time in the timer's {@link TimeDomain}
at which
    * this it should fire. If the timer was already set, resets it to the new requested time.
    */

http://git-wip-us.apache.org/repos/asf/beam/blob/a93c5c05/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
index 2e3fb85..54aad0c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
@@ -48,6 +48,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import org.apache.beam.sdk.Pipeline.PipelineExecutionException;
 import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.sdk.coders.KvCoder;
@@ -1675,6 +1676,98 @@ public class ParDoTest implements Serializable {
     pipeline.run();
   }
 
+  /**
+   * Tests that an event time timer set absolutely for the last possible moment fires and
results in
+   * supplementary output. The test is otherwise identical to {@link #testEventTimeTimerBounded()}.
+   */
+  @Test
+  @Category({RunnableOnService.class, UsesTimersInParDo.class})
+  public void testEventTimeTimerAbsolute() throws Exception {
+    final String timerId = "foo";
+
+    DoFn<KV<String, Integer>, Integer> fn =
+        new DoFn<KV<String, Integer>, Integer>() {
+
+          @TimerId(timerId)
+          private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+          @ProcessElement
+          public void processElement(
+              ProcessContext context, @TimerId(timerId) Timer timer, BoundedWindow window)
{
+            timer.set(window.maxTimestamp());
+            context.output(3);
+          }
+
+          @OnTimer(timerId)
+          public void onTimer(OnTimerContext context) {
+            context.output(42);
+          }
+        };
+
+    PCollection<Integer> output = pipeline.apply(Create.of(KV.of("hello", 37))).apply(ParDo.of(fn));
+    PAssert.that(output).containsInAnyOrder(3, 42);
+    pipeline.run();
+  }
+
+  @Test
+  @Category({RunnableOnService.class, UsesTimersInParDo.class})
+  public void testAbsoluteProcessingTimeTimerRejected() throws Exception {
+    final String timerId = "foo";
+
+    DoFn<KV<String, Integer>, Integer> fn =
+        new DoFn<KV<String, Integer>, Integer>() {
+
+          @TimerId(timerId)
+          private final TimerSpec spec = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
+
+          @ProcessElement
+          public void processElement(ProcessContext context, @TimerId(timerId) Timer timer)
{
+            timer.set(new Instant(0));
+          }
+
+          @OnTimer(timerId)
+          public void onTimer(OnTimerContext context) {}
+        };
+
+    PCollection<Integer> output = pipeline.apply(Create.of(KV.of("hello", 37))).apply(ParDo.of(fn));
+    thrown.expect(PipelineExecutionException.class);
+    // Note that runners can reasonably vary their message - this matcher should be flexible
+    // and can be evolved.
+    thrown.expectMessage("relative timers");
+    thrown.expectMessage("processing time");
+    pipeline.run();
+  }
+
+  @Test
+  @Category({RunnableOnService.class, UsesTimersInParDo.class})
+  public void testOutOfBoundsEventTimeTimer() throws Exception {
+    final String timerId = "foo";
+
+    DoFn<KV<String, Integer>, Integer> fn =
+        new DoFn<KV<String, Integer>, Integer>() {
+
+          @TimerId(timerId)
+          private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+          @ProcessElement
+          public void processElement(
+              ProcessContext context, BoundedWindow window, @TimerId(timerId) Timer timer)
{
+            timer.set(window.maxTimestamp().plus(1L));
+          }
+
+          @OnTimer(timerId)
+          public void onTimer(OnTimerContext context) {}
+        };
+
+    PCollection<Integer> output = pipeline.apply(Create.of(KV.of("hello", 37))).apply(ParDo.of(fn));
+    thrown.expect(PipelineExecutionException.class);
+    // Note that runners can reasonably vary their message - this matcher should be flexible
+    // and can be evolved.
+    thrown.expectMessage("event time timer");
+    thrown.expectMessage("expiration");
+    pipeline.run();
+  }
+
   @Test
   @Category({NeedsRunner.class, UsesTimersInParDo.class, UsesTestStream.class})
   public void testSimpleProcessingTimerTimer() throws Exception {


Mime
View raw message