beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tg...@apache.org
Subject [2/2] beam git commit: Use a consistent calculation for GC Time
Date Thu, 11 May 2017 21:56:36 GMT
Use a consistent calculation for GC Time

Truncate all garbage collection timestamps to be at the end of the
global window at the latest.

Add a reshuffle test, which was failing when late data arrived.

Update ReifyTimestamps to permit infinite skew. Elements that have
timestamps extracted from them may be late, but that is not the concern
of ReifyTimestamps.


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

Branch: refs/heads/master
Commit: dbb850c3c3dedc7f62d90cdf7f15fe7cff79ce1b
Parents: e92ac6f
Author: Thomas Groh <tgroh@google.com>
Authored: Thu May 11 09:26:30 2017 -0700
Committer: Thomas Groh <tgroh@google.com>
Committed: Thu May 11 14:56:17 2017 -0700

----------------------------------------------------------------------
 .../core/LateDataDroppingDoFnRunner.java        |  2 +-
 .../apache/beam/runners/core/LateDataUtils.java | 33 ++++++-
 .../beam/runners/core/ReduceFnRunner.java       | 43 +++-------
 .../beam/runners/core/SimpleDoFnRunner.java     |  2 +-
 .../beam/runners/core/StatefulDoFnRunner.java   |  6 +-
 .../apache/beam/runners/core/WatermarkHold.java |  9 +-
 .../beam/runners/core/LateDataUtilsTest.java    | 90 ++++++++++++++++++++
 .../beam/sdk/transforms/ReifyTimestamps.java    |  6 ++
 .../apache/beam/sdk/transforms/Reshuffle.java   | 29 ++++---
 .../sdk/transforms/ReifyTimestampsTest.java     | 36 ++++++++
 .../beam/sdk/transforms/ReshuffleTest.java      | 27 ++++++
 11 files changed, 234 insertions(+), 49 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
index 66385c1..570f524 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java
@@ -159,7 +159,7 @@ public class LateDataDroppingDoFnRunner<K, InputT, OutputT, W extends
BoundedWin
     /** Is {@code window} expired w.r.t. the garbage collection watermark? */
     private boolean canDropDueToExpiredWindow(BoundedWindow window) {
       Instant inputWM = timerInternals.currentInputWatermarkTime();
-      return window.maxTimestamp().plus(windowingStrategy.getAllowedLateness()).isBefore(inputWM);
+      return LateDataUtils.garbageCollectionTime(window, windowingStrategy).isBefore(inputWM);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java
index 732e60c..8a2b7c6 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataUtils.java
@@ -24,15 +24,46 @@ import com.google.common.collect.Iterables;
 import javax.annotation.Nullable;
 import org.apache.beam.runners.core.metrics.CounterCell;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.WindowingStrategy;
-
+import org.joda.time.Duration;
+import org.joda.time.Instant;
 
 /**
  * Utils to handle late data.
  */
 public class LateDataUtils {
+  private LateDataUtils() {}
+
+  /**
+   * Return when {@code window} should be garbage collected. If the window's expiration time
is on
+   * or after the end of the global window, it will be truncated to the end of the global
window.
+   */
+  public static Instant garbageCollectionTime(
+      BoundedWindow window, WindowingStrategy windowingStrategy) {
+    return garbageCollectionTime(window, windowingStrategy.getAllowedLateness());
+  }
+
+  /**
+   * Return when {@code window} should be garbage collected. If the window's expiration time
is on
+   * or after the end of the global window, it will be truncated to the end of the global
window.
+   */
+  public static Instant garbageCollectionTime(BoundedWindow window, Duration allowedLateness)
{
+
+    // If the end of the window + allowed lateness is beyond the "end of time" aka the end
of the
+    // global window, then we truncate it. The conditional is phrased like it is because
the
+    // addition of EOW + allowed lateness might even overflow the maximum allowed Instant
+    if (GlobalWindow.INSTANCE
+        .maxTimestamp()
+        .minus(allowedLateness)
+        .isBefore(window.maxTimestamp())) {
+      return GlobalWindow.INSTANCE.maxTimestamp();
+    } else {
+      return window.maxTimestamp().plus(allowedLateness);
+    }
+  }
 
   /**
    * Returns an {@code Iterable<WindowedValue<InputT>>} that only contains non-late
input elements.

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
index d2ed835..62d519f 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
@@ -48,7 +48,6 @@ import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.state.TimeDomain;
 import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
-import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo.Timing;
 import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
@@ -663,7 +662,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
       W window = directContext.window();
       this.isEndOfWindow = TimeDomain.EVENT_TIME == timer.getDomain()
           && timer.getTimestamp().equals(window.maxTimestamp());
-      Instant cleanupTime = garbageCollectionTime(window);
+      Instant cleanupTime = LateDataUtils.garbageCollectionTime(window, windowingStrategy);
       this.isGarbageCollection = !timer.getTimestamp().isBefore(cleanupTime);
     }
 
@@ -769,9 +768,11 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
           // cleanup event and handled by the above).
           // Note we must do this even if the trigger is finished so that we are sure to
cleanup
           // any final trigger finished bits.
-          checkState(windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO),
+          checkState(
+              windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO),
               "Unexpected zero getAllowedLateness");
-          Instant cleanupTime = garbageCollectionTime(directContext.window());
+          Instant cleanupTime =
+              LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);
           WindowTracing.debug(
               "ReduceFnRunner.onTimer: Scheduling cleanup timer for key:{}; window:{} at
{} with "
                   + "inputWatermark:{}; outputWatermark:{}",
@@ -957,6 +958,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
     // Extract the window hold, and as a side effect clear it.
     final WatermarkHold.OldAndNewHolds pair =
         watermarkHold.extractAndRelease(renamedContext, isFinished).read();
+    // TODO: This isn't accurate if the elements are late. See BEAM-2262
     final Instant outputTimestamp = pair.oldHold;
     @Nullable Instant newHold = pair.newHold;
 
@@ -972,11 +974,12 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
       if (newHold.isAfter(directContext.window().maxTimestamp())) {
         // The hold must be for garbage collection, which can't have happened yet.
         checkState(
-          newHold.isEqual(garbageCollectionTime(directContext.window())),
-          "new hold %s should be at garbage collection for window %s plus %s",
-          newHold,
-          directContext.window(),
-          windowingStrategy.getAllowedLateness());
+            newHold.isEqual(
+                LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy)),
+            "new hold %s should be at garbage collection for window %s plus %s",
+            newHold,
+            directContext.window(),
+            windowingStrategy.getAllowedLateness());
       } else {
         // The hold must be for the end-of-window, which can't have happened yet.
         checkState(
@@ -1042,7 +1045,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
     String which;
     Instant timer;
     if (endOfWindow.isBefore(inputWM)) {
-      timer = garbageCollectionTime(directContext.window());
+      timer = LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);
       which = "garbage collection";
     } else {
       timer = endOfWindow;
@@ -1072,28 +1075,10 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow>
{
         timerInternals.currentOutputWatermarkTime());
     Instant eow = directContext.window().maxTimestamp();
     directContext.timers().deleteTimer(eow, TimeDomain.EVENT_TIME);
-    Instant gc = garbageCollectionTime(directContext.window());
+    Instant gc = LateDataUtils.garbageCollectionTime(directContext.window(), windowingStrategy);
     if (gc.isAfter(eow)) {
       directContext.timers().deleteTimer(gc, TimeDomain.EVENT_TIME);
     }
   }
 
-  /**
-   * Return when {@code window} should be garbage collected. If the window's expiration time
is on
-   * or after the end of the global window, it will be truncated to the end of the global
window.
-   */
-  private Instant garbageCollectionTime(W window) {
-
-    // If the end of the window + allowed lateness is beyond the "end of time" aka the end
of the
-    // global window, then we truncate it. The conditional is phrased like it is because
the
-    // addition of EOW + allowed lateness might even overflow the maximum allowed Instant
-    if (GlobalWindow.INSTANCE
-        .maxTimestamp()
-        .minus(windowingStrategy.getAllowedLateness())
-        .isBefore(window.maxTimestamp())) {
-      return GlobalWindow.INSTANCE.maxTimestamp();
-    } else {
-      return window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
-    }
-  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/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 85423c0..7ca305e 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
@@ -949,7 +949,7 @@ public class SimpleDoFnRunner<InputT, OutputT> implements DoFnRunner<InputT,
Out
      */
     private Instant minTargetAndGcTime(Instant target) {
       if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
-        Instant windowExpiry = window.maxTimestamp().plus(allowedLateness);
+        Instant windowExpiry = LateDataUtils.garbageCollectionTime(window, allowedLateness);
         if (target.isAfter(windowExpiry)) {
           return windowExpiry;
         }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
index 28a9dee..c68a943 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java
@@ -104,7 +104,7 @@ public class StatefulDoFnRunner<InputT, OutputT, W extends BoundedWindow>
   }
 
   private boolean isLate(BoundedWindow window) {
-    Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+    Instant gcTime = LateDataUtils.garbageCollectionTime(window, windowingStrategy);
     Instant inputWM = cleanupTimer.currentInputWatermarkTime();
     return gcTime.isBefore(inputWM);
   }
@@ -208,7 +208,7 @@ public class StatefulDoFnRunner<InputT, OutputT, W extends BoundedWindow>
 
     @Override
     public void setForWindow(BoundedWindow window) {
-      Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+      Instant gcTime = LateDataUtils.garbageCollectionTime(window, windowingStrategy);
       // make sure this fires after any window.maxTimestamp() timers
       gcTime = gcTime.plus(GC_DELAY_MS);
       timerInternals.setTimer(StateNamespaces.window(windowCoder, window),
@@ -222,7 +222,7 @@ public class StatefulDoFnRunner<InputT, OutputT, W extends BoundedWindow>
         Instant timestamp,
         TimeDomain timeDomain) {
       boolean isEventTimer = timeDomain.equals(TimeDomain.EVENT_TIME);
-      Instant gcTime = window.maxTimestamp().plus(windowingStrategy.getAllowedLateness());
+      Instant gcTime = LateDataUtils.garbageCollectionTime(window, windowingStrategy);
       gcTime = gcTime.plus(GC_DELAY_MS);
       return isEventTimer && GC_TIMER_ID.equals(timerId) && gcTime.equals(timestamp);
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
index 64f5d9b..13e4c43 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
@@ -365,8 +365,7 @@ class WatermarkHold<W extends BoundedWindow> implements Serializable
{
       ReduceFn<?, ?, ?, W>.Context context, boolean paneIsEmpty) {
     Instant outputWM = timerInternals.currentOutputWatermarkTime();
     Instant inputWM = timerInternals.currentInputWatermarkTime();
-    Instant eow = context.window().maxTimestamp();
-    Instant gcHold = eow.plus(windowingStrategy.getAllowedLateness());
+    Instant gcHold = LateDataUtils.garbageCollectionTime(context.window(), windowingStrategy);
 
     if (!windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO)) {
       WindowTracing.trace(
@@ -387,6 +386,12 @@ class WatermarkHold<W extends BoundedWindow> implements Serializable
{
       return null;
     }
 
+    if (!gcHold.isBefore(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
+      // If the garbage collection hold is past the timestamp we can represent, instead truncate
+      // to the maximum timestamp that is not positive infinity. This ensures all windows
will
+      // eventually be garbage collected.
+      gcHold = BoundedWindow.TIMESTAMP_MAX_VALUE.minus(Duration.millis(1L));
+    }
     checkState(!gcHold.isBefore(inputWM),
         "Garbage collection hold %s cannot be before input watermark %s",
         gcHold, inputWM);

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataUtilsTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataUtilsTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataUtilsTest.java
new file mode 100644
index 0000000..f0f315d
--- /dev/null
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/LateDataUtilsTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.beam.runners.core;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.hamcrest.Matchers;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.joda.time.ReadableInstant;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link LateDataUtils}.
+ */
+@RunWith(JUnit4.class)
+public class LateDataUtilsTest {
+  @Test
+  public void beforeEndOfGlobalWindowSame() {
+    FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5));
+    Duration allowedLateness = Duration.standardMinutes(2);
+    WindowingStrategy<?, ?> strategy =
+        WindowingStrategy.globalDefault()
+            .withWindowFn(windowFn)
+            .withAllowedLateness(allowedLateness);
+
+    IntervalWindow window = windowFn.assignWindow(new Instant(10));
+    assertThat(
+        LateDataUtils.garbageCollectionTime(window, strategy),
+        equalTo(window.maxTimestamp().plus(allowedLateness)));
+  }
+
+  @Test
+  public void garbageCollectionTimeAfterEndOfGlobalWindow() {
+    FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5));
+    WindowingStrategy<?, ?> strategy =
+        WindowingStrategy.globalDefault()
+            .withWindowFn(windowFn);
+
+    IntervalWindow window = windowFn.assignWindow(new Instant(BoundedWindow.TIMESTAMP_MAX_VALUE));
+    assertThat(
+        window.maxTimestamp(),
+        Matchers.<ReadableInstant>greaterThan(GlobalWindow.INSTANCE.maxTimestamp()));
+    assertThat(
+        LateDataUtils.garbageCollectionTime(window, strategy),
+        equalTo(GlobalWindow.INSTANCE.maxTimestamp()));
+  }
+
+  @Test
+  public void garbageCollectionTimeAfterEndOfGlobalWindowWithLateness() {
+    FixedWindows windowFn = FixedWindows.of(Duration.standardMinutes(5));
+    Duration allowedLateness = Duration.millis(Long.MAX_VALUE);
+    WindowingStrategy<?, ?> strategy =
+        WindowingStrategy.globalDefault()
+            .withWindowFn(windowFn)
+            .withAllowedLateness(allowedLateness);
+
+    IntervalWindow window = windowFn.assignWindow(new Instant(-100));
+    assertThat(
+        window.maxTimestamp().plus(allowedLateness),
+        Matchers.<ReadableInstant>greaterThan(GlobalWindow.INSTANCE.maxTimestamp()));
+    assertThat(
+        LateDataUtils.garbageCollectionTime(window, strategy),
+        equalTo(GlobalWindow.INSTANCE.maxTimestamp()));
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ReifyTimestamps.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ReifyTimestamps.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ReifyTimestamps.java
index 0b1ab25..990f235 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ReifyTimestamps.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ReifyTimestamps.java
@@ -21,6 +21,7 @@ package org.apache.beam.sdk.transforms;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.TimestampedValue;
+import org.joda.time.Duration;
 
 /**
  * {@link PTransform PTransforms} for reifying the timestamp of values and reemitting the
original
@@ -63,6 +64,11 @@ class ReifyTimestamps {
 
   private static class ExtractTimestampedValueDoFn<K, V>
       extends DoFn<KV<K, TimestampedValue<V>>, KV<K, V>> {
+    @Override
+    public Duration getAllowedTimestampSkew() {
+      return Duration.millis(Long.MAX_VALUE);
+    }
+
     @ProcessElement
     public void processElement(ProcessContext context) {
       KV<K, TimestampedValue<V>> kv = context.element();

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java
index 5394826..3b7122c 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java
@@ -71,22 +71,27 @@ public class Reshuffle<K, V> extends PTransform<PCollection<KV<K,
V>>, PCollecti
             .withTimestampCombiner(TimestampCombiner.EARLIEST)
             .withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
 
-    return input.apply(rewindow)
+    return input
+        .apply(rewindow)
         .apply("ReifyOriginalTimestamps", ReifyTimestamps.<K, V>inValues())
         .apply(GroupByKey.<K, TimestampedValue<V>>create())
         // Set the windowing strategy directly, so that it doesn't get counted as the user
having
         // set allowed lateness.
         .setWindowingStrategyInternal(originalStrategy)
-        .apply("ExpandIterable", ParDo.of(
-            new DoFn<KV<K, Iterable<TimestampedValue<V>>>, KV<K,
TimestampedValue<V>>>() {
-              @ProcessElement
-              public void processElement(ProcessContext c) {
-                K key = c.element().getKey();
-                for (TimestampedValue<V> value : c.element().getValue()) {
-                  c.output(KV.of(key, value));
-                }
-              }
-            }))
-        .apply("RestoreOriginalTimestamps", ReifyTimestamps.<K, V>extractFromValues());
+        .apply(
+            "ExpandIterable",
+            ParDo.of(
+                new DoFn<KV<K, Iterable<TimestampedValue<V>>>, KV<K,
TimestampedValue<V>>>() {
+                  @ProcessElement
+                  public void processElement(ProcessContext c) {
+                    K key = c.element().getKey();
+                    for (TimestampedValue<V> value : c.element().getValue()) {
+                      c.output(KV.of(key, value));
+                    }
+                  }
+                }))
+        .apply(
+            "RestoreOriginalTimestamps",
+            ReifyTimestamps.<K, V>extractFromValues());
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java
index 181433e..e872842 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReifyTimestampsTest.java
@@ -101,4 +101,40 @@ public class ReifyTimestampsTest implements Serializable {
 
     pipeline.run();
   }
+
+  @Test
+  @Category(ValidatesRunner.class)
+  public void extractFromValuesWhenValueTimestampedLaterSucceeds() {
+    PCollection<KV<String, TimestampedValue<Integer>>> preified =
+        pipeline.apply(
+            Create.timestamped(
+                TimestampedValue.of(
+                    KV.of("foo", TimestampedValue.of(0, new Instant((0)))), new Instant(100)),
+                TimestampedValue.of(
+                    KV.of("foo", TimestampedValue.of(1, new Instant(1))), new Instant(101L)),
+                TimestampedValue.of(
+                    KV.of("bar", TimestampedValue.of(2, new Instant(2))), new Instant(102L)),
+                TimestampedValue.of(
+                    KV.of("baz", TimestampedValue.of(3, new Instant(3))), new Instant(103L))));
+
+    PCollection<KV<String, Integer>> timestamped =
+        preified.apply(ReifyTimestamps.<String, Integer>extractFromValues());
+
+    PAssert.that(timestamped)
+        .containsInAnyOrder(KV.of("foo", 0), KV.of("foo", 1), KV.of("bar", 2), KV.of("baz",
3));
+
+    timestamped.apply(
+        "AssertElementTimestamps",
+        ParDo.of(
+            new DoFn<KV<String, Integer>, Void>() {
+              @ProcessElement
+              public void verifyTimestampsEqualValue(ProcessContext context) {
+                assertThat(
+                    new Instant(context.element().getValue().longValue()),
+                    equalTo(context.timestamp()));
+              }
+            }));
+
+    pipeline.run();
+  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/dbb850c3/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java
index 1038fd6..3cd7cf9 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ReshuffleTest.java
@@ -27,8 +27,11 @@ import java.util.List;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.testing.UsesTestStream;
 import org.apache.beam.sdk.testing.ValidatesRunner;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.FixedWindows;
@@ -263,4 +266,28 @@ public class ReshuffleTest implements Serializable {
 
     pipeline.run();
   }
+
+  @Test
+  @Category({ValidatesRunner.class, UsesTestStream.class})
+  public void testReshuffleWithTimestampsStreaming() {
+    TestStream<Long> stream =
+        TestStream.create(VarLongCoder.of())
+            .advanceWatermarkTo(new Instant(0L).plus(Duration.standardDays(48L)))
+            .addElements(
+                TimestampedValue.of(0L, new Instant(0L)),
+                TimestampedValue.of(1L, new Instant(0L).plus(Duration.standardDays(48L))),
+                TimestampedValue.of(
+                    2L, BoundedWindow.TIMESTAMP_MAX_VALUE.minus(Duration.standardDays(48L))))
+            .advanceWatermarkToInfinity();
+    PCollection<KV<String, Long>> input =
+        pipeline
+            .apply(stream).apply(WithKeys.<String, Long>of(""))
+            .apply(
+                Window.<KV<String, Long>>into(FixedWindows.of(Duration.standardMinutes(10L))));
+
+    PCollection<KV<String, Long>> reshuffled = input.apply(Reshuffle.<String,
Long>of());
+    PAssert.that(reshuffled.apply(Values.<Long>create())).containsInAnyOrder(0L, 1L,
2L);
+
+    pipeline.run();
+  }
 }


Mime
View raw message