You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2016/04/01 04:11:50 UTC
[1/2] incubator-beam git commit: Repeatedly#onFire should clear all
finished bits
Repository: incubator-beam
Updated Branches:
refs/heads/master e5bca60de -> 164676bc7
Repeatedly#onFire should clear all finished bits
Previously, Repeatedly#onFire only cleared the finished bits associated
with the root of the sub-tree, as demonstrated by the new unit tests.
This led to problems with AfterFirst#shouldFire, which checked to see if
any of the sub-triggers have their finished bits set.
Now, Repeatedly#onFire calls #resetTree, which clears all the finished
bits in the entire sub-tree.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/2ad027f4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/2ad027f4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/2ad027f4
Branch: refs/heads/master
Commit: 2ad027f49adbcb80ffac87c3fffa81fd68c89e86
Parents: cab0c57
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Mar 31 15:03:59 2016 -0700
Committer: bchambers <bc...@google.com>
Committed: Thu Mar 31 15:45:13 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/windowing/Repeatedly.java | 4 +-
.../transforms/windowing/RepeatedlyTest.java | 83 ++++++++++++++++++++
2 files changed, 85 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2ad027f4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
index 3416551..9be0259 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
@@ -90,8 +90,8 @@ public class Repeatedly<W extends BoundedWindow> extends Trigger<W> {
getRepeated(context).invokeOnFire(context);
if (context.trigger().isFinished(REPEATED)) {
- context.trigger().setFinished(false, REPEATED);
- getRepeated(context).invokeClear(context);
+ // Reset tree will recursively clear the finished bits, and invoke clear.
+ context.forTrigger(getRepeated(context)).trigger().resetTree();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/2ad027f4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
index ddfec1c..99907b2 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
@@ -126,4 +126,87 @@ public class RepeatedlyTest {
IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15));
assertTrue(tester.shouldFire(mergedWindow));
}
+
+ @Test
+ public void testRepeatedlyAfterFirstElementCount() throws Exception {
+ SimpleTriggerTester<GlobalWindow> tester =
+ TriggerTester.forTrigger(
+ Repeatedly.forever(
+ AfterFirst.<GlobalWindow>of(
+ AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+ .plusDelayOf(Duration.standardMinutes(15)),
+ AfterPane.<GlobalWindow>elementCountAtLeast(5))),
+ new GlobalWindows());
+
+ GlobalWindow window = GlobalWindow.INSTANCE;
+
+ tester.injectElements(1);
+ assertFalse(tester.shouldFire(window));
+
+ tester.injectElements(2, 3, 4, 5);
+ assertTrue(tester.shouldFire(window));
+ tester.fireIfShouldFire(window);
+ assertFalse(tester.shouldFire(window));
+ }
+
+ @Test
+ public void testRepeatedlyAfterFirstProcessingTime() throws Exception {
+ SimpleTriggerTester<GlobalWindow> tester =
+ TriggerTester.forTrigger(
+ Repeatedly.forever(
+ AfterFirst.<GlobalWindow>of(
+ AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+ .plusDelayOf(Duration.standardMinutes(15)),
+ AfterPane.<GlobalWindow>elementCountAtLeast(5))),
+ new GlobalWindows());
+
+ GlobalWindow window = GlobalWindow.INSTANCE;
+
+ tester.injectElements(1);
+ assertFalse(tester.shouldFire(window));
+
+ tester.advanceProcessingTime(new Instant(0).plus(Duration.standardMinutes(15)));
+ assertTrue(tester.shouldFire(window));
+ tester.fireIfShouldFire(window);
+ assertFalse(tester.shouldFire(window));
+ }
+
+ @Test
+ public void testRepeatedlyElementCount() throws Exception {
+ SimpleTriggerTester<GlobalWindow> tester =
+ TriggerTester.forTrigger(
+ Repeatedly.forever(AfterPane.<GlobalWindow>elementCountAtLeast(5)),
+ new GlobalWindows());
+
+ GlobalWindow window = GlobalWindow.INSTANCE;
+
+ tester.injectElements(1);
+ assertFalse(tester.shouldFire(window));
+
+ tester.injectElements(2, 3, 4, 5);
+ assertTrue(tester.shouldFire(window));
+ tester.fireIfShouldFire(window);
+ assertFalse(tester.shouldFire(window));
+ }
+
+ @Test
+ public void testRepeatedlyProcessingTime() throws Exception {
+ SimpleTriggerTester<GlobalWindow> tester =
+ TriggerTester.forTrigger(
+ Repeatedly.forever(
+ AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+ .plusDelayOf(Duration.standardMinutes(15))),
+ new GlobalWindows());
+
+ GlobalWindow window = GlobalWindow.INSTANCE;
+
+ tester.injectElements(1);
+ assertFalse(tester.shouldFire(window));
+
+ tester.advanceProcessingTime(new Instant(0).plus(Duration.standardMinutes(15)));
+ assertTrue(tester.shouldFire(window));
+ tester.fireIfShouldFire(window);
+ assertFalse(tester.shouldFire(window));
+ }
+
}
[2/2] incubator-beam git commit: This closes #110
Posted by ke...@apache.org.
This closes #110
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/164676bc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/164676bc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/164676bc
Branch: refs/heads/master
Commit: 164676bc7877bb0bb4e85381639b921b7f902234
Parents: e5bca60 2ad027f
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Mar 31 19:11:35 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu Mar 31 19:11:35 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/windowing/Repeatedly.java | 4 +-
.../transforms/windowing/RepeatedlyTest.java | 83 ++++++++++++++++++++
2 files changed, 85 insertions(+), 2 deletions(-)
----------------------------------------------------------------------