This is an automated email from the ASF dual-hosted git repository. amaliujia pushed a commit to branch release-2.20.0 in repository https://gitbox.apache.org/repos/asf/beam.git
commit cd67dbf1c0209824a06f764f3a1d4c591441c5c8 Author: reuvenlax <re...@google.com> AuthorDate: Tue Apr 7 09:50:14 2020 -0700 Merge pull request #11226: [BEAM-9557] Fix timer window boundary checking --- .../apache/beam/runners/core/SimpleDoFnRunner.java | 4 +- .../org/apache/beam/sdk/transforms/ParDoTest.java | 91 +++++++++++++++++++++- 2 files changed, 92 insertions(+), 3 deletions(-) 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 fa5c695..9cc1b8d 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 @@ -1190,13 +1190,13 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT, Out if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) { checkArgument( !outputTimestamp.isAfter(target), - "Attempted to set an event time timer with an output timestamp of %s that is" + "Attempted to set an event-time timer with an output timestamp of %s that is" + " after the timer firing timestamp %s", outputTimestamp, target); checkArgument( !target.isAfter(windowExpiry), - "Attempted to set an event time timer with a firing timestamp of %s that is" + "Attempted to set an event-time timer with a firing timestamp of %s that is" + " after the expiration of window %s", target, windowExpiry); 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 f3a3b03..7a2978b 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 @@ -3465,10 +3465,99 @@ public class ParDoTest implements Serializable { ProcessContext context, BoundedWindow window, @TimerId(timerId) Timer timer) { try { timer.set(window.maxTimestamp().plus(1L)); + fail("Should have failed due to out-of-bounds timer."); + } catch (RuntimeException e) { + String message = e.getMessage(); + List<String> expectedSubstrings = Arrays.asList("event-time timer", "expiration"); + expectedSubstrings.forEach( + str -> + Preconditions.checkState( + message.contains(str), + "Pipeline didn't fail with the expected strings: %s", + expectedSubstrings)); + } + } + + @OnTimer(timerId) + public void onTimer() {} + }; + + pipeline.apply(Create.of(KV.of("hello", 37))).apply(ParDo.of(fn)); + pipeline.run(); + } + + @Test + @Category({ + ValidatesRunner.class, + UsesTimersInParDo.class, + DataflowPortabilityApiUnsupported.class + }) + public void testOutOfBoundsEventTimeTimerHold() 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) { + try { + timer + .withOutputTimestamp(window.maxTimestamp().plus(1L)) + .set(window.maxTimestamp()); + fail("Should have failed due to out-of-bounds timer."); + } catch (RuntimeException e) { + String message = e.getMessage(); + List<String> expectedSubstrings = + Arrays.asList("event-time timer", "output timestamp"); + expectedSubstrings.forEach( + str -> + Preconditions.checkState( + message.contains(str), + "Pipeline didn't fail with the expected strings: %s", + expectedSubstrings)); + } + } + + @OnTimer(timerId) + public void onTimer() {} + }; + + pipeline.apply(Create.of(KV.of("hello", 37))).apply(ParDo.of(fn)); + pipeline.run(); + } + + @Test + @Category({ + ValidatesRunner.class, + UsesTimersInParDo.class, + DataflowPortabilityApiUnsupported.class + }) + public void testOutOfBoundsProcessingTimeTimerHold() 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, BoundedWindow window, @TimerId(timerId) Timer timer) { + try { + timer + .withOutputTimestamp(window.maxTimestamp().plus(1L)) + .offset(Duration.standardSeconds(1)) + .setRelative(); fail("Should have failed due to processing time with absolute timer."); } catch (RuntimeException e) { String message = e.getMessage(); - List<String> expectedSubstrings = Arrays.asList("event time timer", "expiration"); + List<String> expectedSubstrings = + Arrays.asList("processing-time timer", "output timestamp"); expectedSubstrings.forEach( str -> Preconditions.checkState(