From da57ae814bdc9c8d0661c27906f4a6e3c9e7474f Mon Sep 17 00:00:00 2001 From: Thomas Groh Date: Fri, 7 Oct 2016 15:50:05 -0700 Subject: [PATCH] Only remove Elements that were pending from Pending Elements Update the comparator so elements will only be removed from the collection of pending elements if they were formerly added. This ensures that elements cannot be removed if a timer delivery has an identical timestamp to them. --- .../beam/runners/direct/WatermarkManager.java | 31 +++++----------- .../runners/direct/WatermarkManagerTest.java | 36 +++++++++++++++++++ 2 files changed, 45 insertions(+), 22 deletions(-) diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java index b3d1fc5daec3..4792c39afd5e 100644 --- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java +++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java @@ -21,10 +21,8 @@ import com.google.auto.value.AutoValue; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Function; import com.google.common.base.MoreObjects; import com.google.common.collect.ComparisonChain; -import com.google.common.collect.FluentIterable; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; @@ -210,7 +208,13 @@ private static class AppliedPTransformInputWatermark implements Watermark { public AppliedPTransformInputWatermark(Collection inputWatermarks) { this.inputWatermarks = inputWatermarks; - this.pendingElements = TreeMultiset.create(new WindowedValueByTimestampComparator()); + // The ordering must order elements by timestamp, and must not compare two distinct elements + // as equal. This is built on the assumption that any element added as a pending element will + // be consumed without modifications. + Ordering> pendingElementComparator = + new WindowedValueByTimestampComparator().compound(Ordering.arbitrary()); + this.pendingElements = + TreeMultiset.create(pendingElementComparator); this.objectTimers = new HashMap<>(); currentWatermark = new AtomicReference<>(BoundedWindow.TIMESTAMP_MIN_VALUE); } @@ -625,19 +629,6 @@ public Instant get() { private static final Ordering INSTANT_ORDERING = Ordering.natural(); - /** - * A function that takes a WindowedValue and returns the exploded representation of that - * {@link WindowedValue}. - */ - private static final Function, ? extends Iterable>> - EXPLODE_WINDOWS_FN = - new Function, Iterable>>() { - @Override - public Iterable> apply(WindowedValue input) { - return input.explodeWindows(); - } - }; - /** * For each (Object, PriorityQueue) pair in the provided map, remove each Timer that is before the * latestTime argument and put in in the result with the same key, then remove all of the keys @@ -1128,19 +1119,15 @@ private void setEventTimeHold(Object key, Instant newHold) { } private void removePending(CommittedBundle bundle) { - inputWatermark.removePendingElements(elementsFromBundle(bundle)); + inputWatermark.removePendingElements(bundle.getElements()); synchronizedProcessingInputWatermark.removePending(bundle); } private void addPending(CommittedBundle bundle) { - inputWatermark.addPendingElements(elementsFromBundle(bundle)); + inputWatermark.addPendingElements(bundle.getElements()); synchronizedProcessingInputWatermark.addPending(bundle); } - private Iterable> elementsFromBundle(CommittedBundle bundle) { - return FluentIterable.from(bundle.getElements()).transformAndConcat(EXPLODE_WINDOWS_FN); - } - private Map, FiredTimers> extractFiredTimers() { Map, List> eventTimeTimers = inputWatermark.extractFiredEventTimeTimers(); diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java index a722b4934aaf..8a2724301923 100644 --- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java +++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java @@ -563,6 +563,7 @@ public void updateWatermarkWithUnprocessedElements() { .add(second) .add(third) .commit(clock.now()); + manager.updateWatermarks(null, TimerUpdate.empty(), result(createdInts.getProducingTransformInternal(), @@ -585,6 +586,41 @@ public void updateWatermarkWithUnprocessedElements() { keyedWatermarks.getInputWatermark(), not(laterThan(new Instant(-1000L)))); } + @Test + public void updateWatermarkWithCompletedElementsNotPending() { + WindowedValue first = WindowedValue.timestampedValueInGlobalWindow(1, new Instant(22)); + CommittedBundle createdBundle = bundleFactory.createBundle(createdInts) + .add(first) + .commit(clock.now()); + + WindowedValue second = + WindowedValue.timestampedValueInGlobalWindow(2, new Instant(22)); + CommittedBundle neverCreatedBundle = bundleFactory.createBundle(createdInts) + .add(second) + .commit(clock.now()); + + manager.updateWatermarks(null, + TimerUpdate.empty(), + result(createdInts.getProducingTransformInternal(), + null, + Collections.>singleton(createdBundle)), + BoundedWindow.TIMESTAMP_MAX_VALUE); + + manager.updateWatermarks( + neverCreatedBundle, + TimerUpdate.empty(), + result( + filtered.getProducingTransformInternal(), + neverCreatedBundle.withElements(Collections.>emptyList()), + Collections.>emptyList()), + BoundedWindow.TIMESTAMP_MAX_VALUE); + + manager.refreshAll(); + TransformWatermarks filteredWms = + manager.getWatermarks(filtered.getProducingTransformInternal()); + assertThat(filteredWms.getInputWatermark(), equalTo(new Instant(22L))); + } + /** * Demonstrates that updateWatermarks in the presence of late data is monotonic. */