You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by tg...@apache.org on 2017/02/07 16:59:25 UTC
[2/2] beam git commit: Check that Elements,
Timers have permitted Timestamps
Check that Elements, Timers have permitted Timestamps
Add a checkArgument to ImmutableListBundleFactory that elements must
be timestamped earlier than the maximum representable timestamp.
Add a checkArgument to TimerData that Timers must fire before the
maximum representable timestamp.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/8c3f7773
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/8c3f7773
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/8c3f7773
Branch: refs/heads/master
Commit: 8c3f7773dffb12558c96fa72317bdcb18d576970
Parents: 144bffd
Author: Thomas Groh <tg...@google.com>
Authored: Mon Feb 6 15:49:05 2017 -0800
Committer: Thomas Groh <tg...@google.com>
Committed: Tue Feb 7 08:59:06 2017 -0800
----------------------------------------------------------------------
.../direct/ImmutableListBundleFactory.java | 6 +++++
.../beam/runners/direct/WatermarkManager.java | 6 +++++
.../direct/ImmutableListBundleFactoryTest.java | 23 +++++++++++++++++
.../runners/direct/WatermarkManagerTest.java | 27 ++++++++++++++++++++
4 files changed, 62 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/8c3f7773/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
index 6b342d6..36264ee 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ImmutableListBundleFactory.java
@@ -17,6 +17,7 @@
*/
package org.apache.beam.runners.direct;
+import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import com.google.auto.value.AutoValue;
@@ -94,6 +95,11 @@ class ImmutableListBundleFactory implements BundleFactory {
"Can't add element %s to committed bundle in PCollection %s",
element,
pcollection);
+ checkArgument(
+ element.getTimestamp().isBefore(BoundedWindow.TIMESTAMP_MAX_VALUE),
+ "Can't add an element past the end of time (%s), got timestamp %s",
+ BoundedWindow.TIMESTAMP_MAX_VALUE,
+ element.getTimestamp());
elements.add(element);
if (element.getTimestamp().isBefore(minSoFar)) {
minSoFar = element.getTimestamp();
http://git-wip-us.apache.org/repos/asf/beam/blob/8c3f7773/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
----------------------------------------------------------------------
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 2526d2d..942e67c 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
@@ -17,6 +17,7 @@
*/
package org.apache.beam.runners.direct;
+import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.auto.value.AutoValue;
@@ -1361,6 +1362,11 @@ public class WatermarkManager {
* it has previously been deleted. Returns this {@link TimerUpdateBuilder}.
*/
public TimerUpdateBuilder setTimer(TimerData setTimer) {
+ checkArgument(
+ setTimer.getTimestamp().isBefore(BoundedWindow.TIMESTAMP_MAX_VALUE),
+ "Got a timer for after the end of time (%s), got %s",
+ BoundedWindow.TIMESTAMP_MAX_VALUE,
+ setTimer.getTimestamp());
deletedTimers.remove(setTimer);
setTimers.add(setTimer);
return this;
http://git-wip-us.apache.org/repos/asf/beam/blob/8c3f7773/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
index 3327ccd..c5ad0cd 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/ImmutableListBundleFactoryTest.java
@@ -44,6 +44,7 @@ import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.hamcrest.Matcher;
import org.hamcrest.Matchers;
+import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Rule;
@@ -141,6 +142,28 @@ public class ImmutableListBundleFactoryTest {
afterCommitGetElementsShouldHaveAddedElements(ImmutableList.of(firstValue, secondValue));
}
+ @Test
+ public void addElementsAtEndOfTimeThrows() {
+ Instant timestamp = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ WindowedValue<Integer> value = WindowedValue.timestampedValueInGlobalWindow(1, timestamp);
+
+ UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle();
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage(timestamp.toString());
+ bundle.add(value);
+ }
+
+ @Test
+ public void addElementsPastEndOfTimeThrows() {
+ Instant timestamp = BoundedWindow.TIMESTAMP_MAX_VALUE.plus(Duration.standardMinutes(2));
+ WindowedValue<Integer> value = WindowedValue.timestampedValueInGlobalWindow(1, timestamp);
+
+ UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle();
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage(timestamp.toString());
+ bundle.add(value);
+ }
+
@SuppressWarnings("unchecked")
@Test
public void withElementsShouldReturnIndependentBundle() {
http://git-wip-us.apache.org/repos/asf/beam/blob/8c3f7773/runners/direct-java/src/test/java/org/apache/beam/runners/direct/WatermarkManagerTest.java
----------------------------------------------------------------------
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 493edbb..e36452e 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
@@ -67,11 +67,13 @@ import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.Matchers;
+import org.joda.time.Duration;
import org.joda.time.Instant;
import org.joda.time.ReadableInstant;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -80,6 +82,9 @@ import org.junit.runners.JUnit4;
*/
@RunWith(JUnit4.class)
public class WatermarkManagerTest implements Serializable {
+ @Rule
+ public transient ExpectedException thrown = ExpectedException.none();
+
private transient MockClock clock;
private transient PCollection<Integer> createdInts;
@@ -1348,6 +1353,28 @@ public class WatermarkManagerTest implements Serializable {
}
@Test
+ public void timerUpdateBuilderWithSetAtEndOfTime() {
+ Instant timerStamp = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ TimerData tooFar = TimerData.of(StateNamespaces.global(), timerStamp, TimeDomain.EVENT_TIME);
+
+ TimerUpdateBuilder builder = TimerUpdate.builder(StructuralKey.empty());
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage(timerStamp.toString());
+ builder.setTimer(tooFar);
+ }
+
+ @Test
+ public void timerUpdateBuilderWithSetPastEndOfTime() {
+ Instant timerStamp = BoundedWindow.TIMESTAMP_MAX_VALUE.plus(Duration.standardMinutes(2));
+ TimerData tooFar = TimerData.of(StateNamespaces.global(), timerStamp, TimeDomain.EVENT_TIME);
+
+ TimerUpdateBuilder builder = TimerUpdate.builder(StructuralKey.empty());
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage(timerStamp.toString());
+ builder.setTimer(tooFar);
+ }
+
+ @Test
public void timerUpdateBuilderWithSetThenDeleteHasOnlyDeleted() {
TimerUpdateBuilder builder = TimerUpdate.builder(null);
TimerData timer = TimerData.of(StateNamespaces.global(), Instant.now(), TimeDomain.EVENT_TIME);