FIX: windmill uses kint64max usec as the timer max timestamp, keep th…

…e Java side in sync. ----Release Notes---- [] ------------- Created by MOE: http://code.google.com/p/moe-java MOE_MIGRATED_REVID=88582344
bhargavvar · Mar 13, 2015 · 4e0ecff · 4e0ecff
1 parent d6d4a34
commit 4e0ecff
Show file tree

Hide file tree

Showing 6 changed files with 17 additions and 6 deletions.
diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/ShuffleSink.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/ShuffleSink.java
@@ -22,6 +22,7 @@
 import com.google.cloud.dataflow.sdk.coders.InstantCoder;
 import com.google.cloud.dataflow.sdk.coders.KvCoder;
 import com.google.cloud.dataflow.sdk.options.PipelineOptions;
+import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
 import com.google.cloud.dataflow.sdk.util.CoderUtils;
 import com.google.cloud.dataflow.sdk.util.WindowedValue;
 import com.google.cloud.dataflow.sdk.util.WindowedValue.WindowedValueCoder;
@@ -194,7 +195,7 @@ public long add(WindowedValue<T> windowedElem) throws IOException {
 
         } else if (groupValues) {
           // Sort values by timestamp so that GroupAlsoByWindows can run efficiently.
-          if (windowedElem.getTimestamp().getMillis() == Long.MIN_VALUE) {
+          if (windowedElem.getTimestamp().equals(BoundedWindow.TIMESTAMP_MIN_VALUE)) {
             // Empty secondary keys sort before all other secondary keys, so we
             // can omit this common value here for efficiency.
             secondaryKeyBytes = null;

diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/BoundedWindow.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/BoundedWindow.java
@@ -18,6 +18,8 @@
 
 import org.joda.time.Instant;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * A {@code BoundedWindow} represents a finite grouping of elements, with an
  * upper bound (larger timestamps represent more recent data) on the timestamps
@@ -30,6 +32,13 @@
  * be treated as equal by {@code equals()} and {@code hashCode()}.
  */
 public abstract class BoundedWindow {
+  // The min and max timestmaps that won't overflow when they are converted to
+  // usec.
+  public static final Instant TIMESTAMP_MIN_VALUE =
+      new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MIN_VALUE));
+  public static final Instant TIMESTAMP_MAX_VALUE =
+      new Instant(TimeUnit.MICROSECONDS.toMillis(Long.MAX_VALUE));
+
   /**
    * Returns the upper bound of timestamps for values in this window.
    */

diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/GlobalWindow.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/GlobalWindow.java
@@ -32,7 +32,7 @@ public class GlobalWindow extends BoundedWindow {
 
   @Override
   public Instant maxTimestamp() {
-    return new Instant(Long.MAX_VALUE);
+    return TIMESTAMP_MAX_VALUE;
   }
 
   private GlobalWindow() {}

diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/DoFnContext.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/DoFnContext.java
@@ -123,7 +123,7 @@ <T> WindowedValue<T> makeWindowedValue(
     final Instant inputTimestamp = timestamp;
 
     if (timestamp == null) {
-      timestamp = new Instant(Long.MIN_VALUE);
+      timestamp = BoundedWindow.TIMESTAMP_MIN_VALUE;
     }
 
     if (windows == null) {

diff --git a/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java b/sdk/src/main/java/com/google/cloud/dataflow/sdk/util/WindowedValue.java
@@ -71,7 +71,7 @@ public static <V> WindowedValue<V> of(
    */
   public static <V> WindowedValue<V> valueInGlobalWindow(V value) {
     return new WindowedValue<>(value,
-                               new Instant(Long.MIN_VALUE),
+                               BoundedWindow.TIMESTAMP_MIN_VALUE,
                                Arrays.asList(GlobalWindow.INSTANCE));
   }
 
@@ -80,7 +80,7 @@ public static <V> WindowedValue<V> valueInGlobalWindow(V value) {
    */
   public static <V> WindowedValue<V> valueInEmptyWindows(V value) {
     return new WindowedValue<V>(value,
-                                new Instant(Long.MIN_VALUE),
+                                BoundedWindow.TIMESTAMP_MIN_VALUE,
                                 Collections.<BoundedWindow>emptyList());
   }
 

diff --git a/...src/test/java/com/google/cloud/dataflow/sdk/runners/worker/GroupingShuffleReaderTest.java b/...src/test/java/com/google/cloud/dataflow/sdk/runners/worker/GroupingShuffleReaderTest.java
@@ -37,6 +37,7 @@
 import com.google.cloud.dataflow.sdk.coders.KvCoder;
 import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
 import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
 import com.google.cloud.dataflow.sdk.transforms.windowing.IntervalWindow;
 import com.google.cloud.dataflow.sdk.util.BatchModeExecutionContext;
 import com.google.cloud.dataflow.sdk.util.CoderUtils;
@@ -150,7 +151,7 @@ private void runTestReadFromShuffle(
 
         WindowedValue<KV<Integer, Reiterable<String>>> windowedValue = iter.next();
         // Verify value is in an empty windows.
-        assertEquals(Long.MIN_VALUE, windowedValue.getTimestamp().getMillis());
+        assertEquals(BoundedWindow.TIMESTAMP_MIN_VALUE, windowedValue.getTimestamp());
         assertEquals(0, windowedValue.getWindows().size());
 
         KV<Integer, Reiterable<String>> elem = windowedValue.getValue();