This is an automated email from the ASF dual-hosted git repository. kenn pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push: new 88acc52 [BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle (#11924) 88acc52 is described below commit 88acc5267f759d81e9836a9db17b9e0ee521c785 Author: Rehman Murad Ali <rehmanmurada...@gmail.com> AuthorDate: Thu Jul 30 01:06:52 2020 +0500 [BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle (#11924) --- CHANGES.md | 1 + .../worker/StreamingModeExecutionContext.java | 75 +++++++++++++++++----- .../dataflow/worker/WindmillTimerInternals.java | 25 ++++++++ .../org/apache/beam/sdk/transforms/ParDoTest.java | 69 ++++++++++++++++---- 4 files changed, 141 insertions(+), 29 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 1cfc52c..c2665e7 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -146,6 +146,7 @@ * Upgrade Sphinx to 3.0.3 for building PyDoc. * Added a PTransform for image annotation using Google Cloud AI image processing service ([BEAM-9646](https://issues.apache.org/jira/browse/BEAM-9646)) +* Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle ([BEAM-8543](https://issues.apache.org/jira/browse/BEAM-8543)). ## Breaking Changes diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java index 7ed390b..334f145 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java @@ -24,10 +24,12 @@ import com.google.api.services.dataflow.model.CounterUpdate; import com.google.api.services.dataflow.model.SideInputInfo; import java.io.Closeable; import java.io.IOException; +import java.util.Comparator; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.PriorityQueue; import java.util.Set; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicLong; @@ -519,7 +521,7 @@ public class StreamingModeExecutionContext extends DataflowExecutionContext<Step synchronizedProcessingTime); this.cachedFiredTimers = null; - this.cachedFiredUserTimers = null; + this.toBeFiredTimersOrdered = null; } public void flushState() { @@ -559,28 +561,67 @@ public class StreamingModeExecutionContext extends DataflowExecutionContext<Step return nextTimer; } - // Lazily initialized - private Iterator<TimerData> cachedFiredUserTimers = null; + private PriorityQueue<TimerData> toBeFiredTimersOrdered = null; + + // to track if timer is reset earlier mid-bundle. + // Map of timer's id to timer's firing time to check + // the actual firing time of a timer. + private Map<String, Instant> firedTimer = new HashMap<>(); public <W extends BoundedWindow> TimerData getNextFiredUserTimer(Coder<W> windowCoder) { - if (cachedFiredUserTimers == null) { - cachedFiredUserTimers = - FluentIterable.<Timer>from(StreamingModeExecutionContext.this.getFiredTimers()) - .filter( - timer -> - WindmillTimerInternals.isUserTimer(timer) - && timer.getStateFamily().equals(stateFamily)) - .transform( - timer -> - WindmillTimerInternals.windmillTimerToTimerData( - WindmillNamespacePrefix.USER_NAMESPACE_PREFIX, timer, windowCoder)) - .iterator(); + if (toBeFiredTimersOrdered == null) { + + toBeFiredTimersOrdered = new PriorityQueue<>(Comparator.comparing(TimerData::getTimestamp)); + FluentIterable.from(StreamingModeExecutionContext.this.getFiredTimers()) + .filter( + timer -> + WindmillTimerInternals.isUserTimer(timer) + && timer.getStateFamily().equals(stateFamily)) + .transform( + timer -> + WindmillTimerInternals.windmillTimerToTimerData( + WindmillNamespacePrefix.USER_NAMESPACE_PREFIX, timer, windowCoder)) + .iterator() + .forEachRemaining( + timerData -> { + firedTimer.put( + timerData.getTimerId() + '+' + timerData.getTimerFamilyId(), + timerData.getTimestamp()); + toBeFiredTimersOrdered.add(timerData); + }); } - if (!cachedFiredUserTimers.hasNext()) { + Instant currentInputWatermark = userTimerInternals.currentInputWatermarkTime(); + + if (userTimerInternals.hasTimerBefore(currentInputWatermark)) { + List<TimerData> currentTimers = userTimerInternals.getCurrentTimers(); + + for (TimerData timerData : currentTimers) { + firedTimer.put( + timerData.getTimerId() + '+' + timerData.getTimerFamilyId(), + timerData.getTimestamp()); + toBeFiredTimersOrdered.add(timerData); + } + } + + TimerData nextTimer = null; + + // fire timer only if its timestamp matched. Else it is either reset or obsolete. + while (!toBeFiredTimersOrdered.isEmpty()) { + nextTimer = toBeFiredTimersOrdered.poll(); + String timerUniqueId = nextTimer.getTimerId() + '+' + nextTimer.getTimerFamilyId(); + if (firedTimer.containsKey(timerUniqueId) + && firedTimer.get(timerUniqueId).isEqual(nextTimer.getTimestamp())) { + break; + } else { + nextTimer = null; + } + } + + if (nextTimer == null) { return null; } - TimerData nextTimer = cachedFiredUserTimers.next(); + // User timers must be explicitly deleted when delivered, to release the implied hold userTimerInternals.deleteTimer(nextTimer); return nextTimer; diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java index f46fd49..5269cf2 100644 --- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java +++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java @@ -22,6 +22,8 @@ import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Prec import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; import org.apache.beam.runners.core.StateNamespace; import org.apache.beam.runners.core.StateNamespaces; import org.apache.beam.runners.core.TimerInternals; @@ -225,6 +227,29 @@ class WindmillTimerInternals implements TimerInternals { timers.clear(); } + public boolean hasTimerBefore(Instant time) { + for (Cell<String, StateNamespace, Boolean> cell : timerStillPresent.cellSet()) { + TimerData timerData = timers.get(cell.getRowKey(), cell.getColumnKey()); + if (cell.getValue()) { + if (timerData.getTimestamp().isBefore(time)) { + return true; + } + } + } + return false; + } + + public List<TimerData> getCurrentTimers() { + List<TimerData> timerDataList = new ArrayList<>(); + for (Cell<String, StateNamespace, Boolean> cell : timerStillPresent.cellSet()) { + TimerData timerData = timers.get(cell.getRowKey(), cell.getColumnKey()); + if (cell.getValue()) { + timerDataList.add(timerData); + } + } + return timerDataList; + } + private boolean needsWatermarkHold(TimerData timerData) { // If it is a user timer or a system timer with outputTimestamp different than timestamp return WindmillNamespacePrefix.USER_NAMESPACE_PREFIX.equals(prefix) 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 3f0d13c..776fd3e 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 @@ -3959,7 +3959,7 @@ public class ParDoTest implements Serializable { } testEventTimeTimerOrderingWithInputPTransform( - now, numTestElements, builder.advanceWatermarkToInfinity()); + now, numTestElements, builder.advanceWatermarkToInfinity(), IsBounded.BOUNDED); } /** A test makes sure that an event time timers are correctly ordered using Create transform. */ @@ -3970,7 +3970,7 @@ public class ParDoTest implements Serializable { UsesStatefulParDo.class, UsesStrictTimerOrdering.class }) - public void testEventTimeTimerOrderingWithCreate() throws Exception { + public void testEventTimeTimerOrderingWithCreateBounded() throws Exception { final int numTestElements = 100; final Instant now = new Instant(1500000000000L); @@ -3980,13 +3980,39 @@ public class ParDoTest implements Serializable { } testEventTimeTimerOrderingWithInputPTransform( - now, numTestElements, Create.timestamped(elements)); + now, numTestElements, Create.timestamped(elements), IsBounded.BOUNDED); + } + + /** + * A test makes sure that an event time timers are correctly ordered using Create transform + * unbounded. + */ + @Test + @Category({ + ValidatesRunner.class, + UsesTimersInParDo.class, + UsesStatefulParDo.class, + UsesUnboundedPCollections.class, + UsesStrictTimerOrdering.class + }) + public void testEventTimeTimerOrderingWithCreateUnbounded() throws Exception { + final int numTestElements = 100; + final Instant now = new Instant(1500000000000L); + + List<TimestampedValue<KV<String, String>>> elements = new ArrayList<>(); + for (int i = 0; i < numTestElements; i++) { + elements.add(TimestampedValue.of(KV.of("dummy", "" + i), now.plus(i * 1000))); + } + + testEventTimeTimerOrderingWithInputPTransform( + now, numTestElements, Create.timestamped(elements), IsBounded.UNBOUNDED); } private void testEventTimeTimerOrderingWithInputPTransform( Instant now, int numTestElements, - PTransform<PBegin, PCollection<KV<String, String>>> transform) + PTransform<PBegin, PCollection<KV<String, String>>> transform, + IsBounded isBounded) throws Exception { final String timerIdBagAppend = "append"; @@ -4070,7 +4096,8 @@ public class ParDoTest implements Serializable { } }; - PCollection<String> output = pipeline.apply(transform).apply(ParDo.of(fn)); + PCollection<String> output = + pipeline.apply(transform).setIsBoundedInternal(isBounded).apply(ParDo.of(fn)); List<String> expected = IntStream.rangeClosed(0, numTestElements) .mapToObj(expandFn(numTestElements)) @@ -4154,16 +4181,25 @@ public class ParDoTest implements Serializable { TestStream.create(KvCoder.of(VoidCoder.of(), VoidCoder.of())) .addElements(KV.of(null, null)) .advanceWatermarkToInfinity(); - pipeline.apply(TwoTimerTest.of(now, end, input)); + pipeline.apply(TwoTimerTest.of(now, end, input, IsBounded.BOUNDED)); + pipeline.run(); + } + + @Test + @Category({ValidatesRunner.class, UsesTimersInParDo.class, UsesStrictTimerOrdering.class}) + public void testTwoTimersSettingEachOtherWithCreateAsInputBounded() { + Instant now = new Instant(1500000000000L); + Instant end = now.plus(100); + pipeline.apply(TwoTimerTest.of(now, end, Create.of(KV.of(null, null)), IsBounded.BOUNDED)); pipeline.run(); } @Test @Category({ValidatesRunner.class, UsesTimersInParDo.class, UsesStrictTimerOrdering.class}) - public void testTwoTimersSettingEachOtherWithCreateAsInput() { + public void testTwoTimersSettingEachOtherWithCreateAsInputUnbounded() { Instant now = new Instant(1500000000000L); Instant end = now.plus(100); - pipeline.apply(TwoTimerTest.of(now, end, Create.of(KV.of(null, null)))); + pipeline.apply(TwoTimerTest.of(now, end, Create.of(KV.of(null, null)), IsBounded.UNBOUNDED)); pipeline.run(); } @@ -4337,18 +4373,26 @@ public class ParDoTest implements Serializable { private static class TwoTimerTest extends PTransform<PBegin, PDone> { private static PTransform<PBegin, PDone> of( - Instant start, Instant end, PTransform<PBegin, PCollection<KV<Void, Void>>> input) { - return new TwoTimerTest(start, end, input); + Instant start, + Instant end, + PTransform<PBegin, PCollection<KV<Void, Void>>> input, + IsBounded isBounded) { + return new TwoTimerTest(start, end, input, isBounded); } private final Instant start; private final Instant end; + private final IsBounded isBounded; private final transient PTransform<PBegin, PCollection<KV<Void, Void>>> inputPTransform; public TwoTimerTest( - Instant start, Instant end, PTransform<PBegin, PCollection<KV<Void, Void>>> input) { + Instant start, + Instant end, + PTransform<PBegin, PCollection<KV<Void, Void>>> input, + IsBounded isBounded) { this.start = start; this.end = end; + this.isBounded = isBounded; this.inputPTransform = input; } @@ -4361,6 +4405,7 @@ public class ParDoTest implements Serializable { PCollection<String> result = input .apply(inputPTransform) + .setIsBoundedInternal(isBounded) .apply( ParDo.of( new DoFn<KV<Void, Void>, String>() { @@ -4425,7 +4470,7 @@ public class ParDoTest implements Serializable { })); List<String> expected = - LongStream.rangeClosed(0, 100) + LongStream.rangeClosed(0, end.minus(start.getMillis()).getMillis()) .mapToObj(e -> (Long) e) .flatMap(e -> Arrays.asList("t1:" + e + ":" + e, "t2:" + e + ":" + e).stream()) .collect(Collectors.toList());