Merge pull request #21740 from Naireen/weird_timestamps

convert windmill min timestamp to beam min timestamp
apache · Jun 15, 2022 · 2104960 · 2104960
2 parents 9a74f17 + 955177b
commit 2104960
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 0 deletions.
diff --git a/...-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimeUtils.java b/...-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimeUtils.java
@@ -45,6 +45,9 @@ public static Instant windmillToHarnessTimestamp(long timestampUs) {
     // Windmill should never send us an unknown timestamp.
     Preconditions.checkArgument(timestampUs != Long.MIN_VALUE);
     Instant result = new Instant(divideAndRoundDown(timestampUs, 1000));
+    if (result.isBefore(BoundedWindow.TIMESTAMP_MIN_VALUE)) {
+      return BoundedWindow.TIMESTAMP_MIN_VALUE;
+    }
     if (result.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
       // End of time.
       return BoundedWindow.TIMESTAMP_MAX_VALUE;

diff --git a/...a/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillTimeUtilsTest.java b/...a/worker/src/test/java/org/apache/beam/runners/dataflow/worker/WindmillTimeUtilsTest.java
@@ -23,6 +23,7 @@
 import static org.junit.Assert.assertEquals;
 
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.joda.time.Duration;
 import org.joda.time.Instant;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -56,6 +57,15 @@ public void testWindmillToHarnessTimestamp() {
     assertEquals(new Instant(-17), windmillToHarnessTimestamp(-16987));
     assertEquals(new Instant(-17), windmillToHarnessTimestamp(-17000));
     assertEquals(new Instant(-18), windmillToHarnessTimestamp(-17001));
+    assertEquals(BoundedWindow.TIMESTAMP_MIN_VALUE, windmillToHarnessTimestamp(Long.MIN_VALUE + 1));
+    assertEquals(BoundedWindow.TIMESTAMP_MIN_VALUE, windmillToHarnessTimestamp(Long.MIN_VALUE + 2));
+    // Long.MIN_VALUE = -9223372036854775808, need to add 1808 microseconds to get to next
+    // millisecond returned by Beam.
+    assertEquals(
+        BoundedWindow.TIMESTAMP_MIN_VALUE.plus(Duration.millis(1)),
+        windmillToHarnessTimestamp(Long.MIN_VALUE + 1808));
+    assertEquals(
+        BoundedWindow.TIMESTAMP_MIN_VALUE, windmillToHarnessTimestamp(Long.MIN_VALUE + 1807));
   }
 
   @Test