beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/2] incubator-beam git commit: Remove remaining window parameterization from triggers and their tests
Date Wed, 20 Apr 2016 18:08:03 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master d5b1d5135 -> ad45d5f75


Remove remaining window parameterization from triggers and their tests


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

Branch: refs/heads/master
Commit: beb0b36e23940fbf2a201a983133bc07887fa5a0
Parents: b8951c2
Author: Kenneth Knowles <klk@google.com>
Authored: Tue Apr 19 13:35:31 2016 -0700
Committer: Kenneth Knowles <klk@google.com>
Committed: Wed Apr 20 09:29:34 2016 -0700

----------------------------------------------------------------------
 .../beam/sdk/transforms/windowing/AfterAll.java |  2 +-
 .../sdk/transforms/windowing/AfterEach.java     |  2 +-
 .../sdk/transforms/windowing/AfterFirst.java    |  2 +-
 .../windowing/AfterProcessingTime.java          |  2 +-
 .../transforms/windowing/DefaultTrigger.java    |  2 +-
 .../sdk/transforms/windowing/Repeatedly.java    |  2 +-
 .../sdk/transforms/windowing/AfterAllTest.java  | 14 +++++------
 .../sdk/transforms/windowing/AfterEachTest.java |  8 +++---
 .../transforms/windowing/AfterFirstTest.java    |  6 ++---
 .../sdk/transforms/windowing/AfterPaneTest.java |  6 ++---
 .../windowing/AfterProcessingTimeTest.java      |  6 ++---
 .../AfterSynchronizedProcessingTimeTest.java    |  4 +--
 .../windowing/AfterWatermarkTest.java           | 26 ++++++++++----------
 .../windowing/DefaultTriggerTest.java           |  6 ++---
 .../sdk/transforms/windowing/NeverTest.java     |  2 +-
 .../windowing/OrFinallyTriggerTest.java         | 18 +++++++-------
 .../transforms/windowing/RepeatedlyTest.java    | 18 +++++++-------
 .../beam/sdk/util/ReduceFnRunnerTest.java       | 22 ++++++++---------
 18 files changed, 74 insertions(+), 74 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterAll.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterAll.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterAll.java
index 0f609df..e6d8e84 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterAll.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterAll.java
@@ -44,7 +44,7 @@ public class AfterAll extends OnceTrigger {
    * Returns an {@code AfterAll} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static <W extends BoundedWindow> OnceTrigger of(OnceTrigger... triggers) {
+  public static OnceTrigger of(OnceTrigger... triggers) {
     return new AfterAll(Arrays.<Trigger>asList(triggers));
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterEach.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterEach.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterEach.java
index 59cb73c..bd57339 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterEach.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterEach.java
@@ -56,7 +56,7 @@ public class AfterEach extends Trigger {
    * Returns an {@code AfterEach} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static <W extends BoundedWindow> Trigger inOrder(Trigger... triggers) {
+  public static Trigger inOrder(Trigger... triggers) {
     return new AfterEach(Arrays.<Trigger>asList(triggers));
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterFirst.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterFirst.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterFirst.java
index a8508a3..a43d9ac 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterFirst.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterFirst.java
@@ -45,7 +45,7 @@ public class AfterFirst extends OnceTrigger {
    * Returns an {@code AfterFirst} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static <W extends BoundedWindow> OnceTrigger of(
+  public static OnceTrigger of(
       OnceTrigger... triggers) {
     return new AfterFirst(Arrays.<Trigger>asList(triggers));
   }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
index 05c6815..563455b 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
@@ -51,7 +51,7 @@ public class AfterProcessingTime extends AfterDelayFromFirstElement {
    * Creates a trigger that fires when the current processing time passes the processing
time
    * at which this trigger saw the first element in a pane.
    */
-  public static <W extends BoundedWindow> AfterProcessingTime pastFirstElementInPane()
{
+  public static AfterProcessingTime pastFirstElementInPane() {
     return new AfterProcessingTime(IDENTITY);
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/DefaultTrigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/DefaultTrigger.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/DefaultTrigger.java
index 3c6dbf3..fcea333 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/DefaultTrigger.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/DefaultTrigger.java
@@ -38,7 +38,7 @@ public class DefaultTrigger extends Trigger{
   /**
    * Returns the default trigger.
    */
-  public static <W extends BoundedWindow> DefaultTrigger of() {
+  public static DefaultTrigger of() {
     return new DefaultTrigger();
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Repeatedly.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Repeatedly.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Repeatedly.java
index ec79cf9..591bbf0 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Repeatedly.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Repeatedly.java
@@ -47,7 +47,7 @@ public class Repeatedly extends Trigger {
    *
    * @param repeated the trigger to execute repeatedly.
    */
-  public static <W extends BoundedWindow> Repeatedly forever(Trigger repeated) {
+  public static Repeatedly forever(Trigger repeated) {
     return new Repeatedly(repeated);
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterAllTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterAllTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterAllTest.java
index 969c1fe..b798013 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterAllTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterAllTest.java
@@ -43,8 +43,8 @@ public class AfterAllTest {
   public void testT1FiresFirst() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterAll.of(
-            AfterPane.<IntervalWindow>elementCountAtLeast(1),
-            AfterPane.<IntervalWindow>elementCountAtLeast(2)),
+            AfterPane.elementCountAtLeast(1),
+            AfterPane.elementCountAtLeast(2)),
         FixedWindows.of(Duration.millis(100)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100));
@@ -62,8 +62,8 @@ public class AfterAllTest {
   public void testT2FiresFirst() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterAll.of(
-            AfterPane.<IntervalWindow>elementCountAtLeast(2),
-            AfterPane.<IntervalWindow>elementCountAtLeast(1)),
+            AfterPane.elementCountAtLeast(2),
+            AfterPane.elementCountAtLeast(1)),
         FixedWindows.of(Duration.millis(100)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100));
@@ -86,9 +86,9 @@ public class AfterAllTest {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
             AfterAll.of(
-                AfterWatermark.<IntervalWindow>pastEndOfWindow(),
-                AfterPane.<IntervalWindow>elementCountAtLeast(1)),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(1))),
+                AfterWatermark.pastEndOfWindow(),
+                AfterPane.elementCountAtLeast(1)),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterEachTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterEachTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterEachTest.java
index f5d83a7..df557eb 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterEachTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterEachTest.java
@@ -53,10 +53,10 @@ public class AfterEachTest {
   public void testAfterEachInSequence() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2))
-                .orFinally(AfterPane.<IntervalWindow>elementCountAtLeast(3)),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(5))
-                .orFinally(AfterWatermark.<IntervalWindow>pastEndOfWindow())),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(2))
+                .orFinally(AfterPane.elementCountAtLeast(3)),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(5))
+                .orFinally(AfterWatermark.pastEndOfWindow())),
             FixedWindows.of(Duration.millis(10)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(10));

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterFirstTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterFirstTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterFirstTest.java
index c0a9f2b..a166694 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterFirstTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterFirstTest.java
@@ -123,9 +123,9 @@ public class AfterFirstTest {
   public void testShouldFireAfterMerge() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
-            AfterFirst.of(AfterPane.<IntervalWindow>elementCountAtLeast(5),
-                AfterWatermark.<IntervalWindow>pastEndOfWindow()),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(1))),
+            AfterFirst.of(AfterPane.elementCountAtLeast(5),
+                AfterWatermark.pastEndOfWindow()),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))),
         Sessions.withGapDuration(Duration.millis(10)));
 
     // Finished the AfterFirst in the first window

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterPaneTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterPaneTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterPaneTest.java
index 827d4c6..76ee49c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterPaneTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterPaneTest.java
@@ -44,7 +44,7 @@ public class AfterPaneTest {
   @Test
   public void testAfterPaneElementCountFixedWindows() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterPane.<IntervalWindow>elementCountAtLeast(2),
+        AfterPane.elementCountAtLeast(2),
         FixedWindows.of(Duration.millis(10)));
 
     tester.injectElements(1); // [0, 10)
@@ -65,7 +65,7 @@ public class AfterPaneTest {
   @Test
   public void testClear() throws Exception {
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
-        AfterPane.<IntervalWindow>elementCountAtLeast(2),
+        AfterPane.elementCountAtLeast(2),
         FixedWindows.of(Duration.millis(10)));
 
     tester.injectElements(1, 2, 3);
@@ -77,7 +77,7 @@ public class AfterPaneTest {
   @Test
   public void testAfterPaneElementCountSessions() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterPane.<IntervalWindow>elementCountAtLeast(2),
+        AfterPane.elementCountAtLeast(2),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
index 81aad33..8178d54 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
@@ -46,7 +46,7 @@ public class AfterProcessingTimeTest {
     Duration windowDuration = Duration.millis(10);
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
         AfterProcessingTime
-            .<IntervalWindow>pastFirstElementInPane()
+            .pastFirstElementInPane()
             .plusDelayOf(Duration.millis(5)),
         FixedWindows.of(windowDuration));
 
@@ -93,7 +93,7 @@ public class AfterProcessingTimeTest {
   public void testClear() throws Exception {
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
         AfterProcessingTime
-            .<IntervalWindow>pastFirstElementInPane()
+            .pastFirstElementInPane()
             .plusDelayOf(Duration.millis(5)),
         FixedWindows.of(Duration.millis(10)));
 
@@ -107,7 +107,7 @@ public class AfterProcessingTimeTest {
   public void testAfterProcessingTimeWithMergingWindow() throws Exception {
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
         AfterProcessingTime
-            .<IntervalWindow>pastFirstElementInPane()
+            .pastFirstElementInPane()
             .plusDelayOf(Duration.millis(5)),
         Sessions.withGapDuration(Duration.millis(10)));
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
index a44be90..4c089db 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
@@ -43,7 +43,7 @@ public class AfterSynchronizedProcessingTimeTest {
     Duration windowDuration = Duration.millis(10);
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
         AfterProcessingTime
-            .<IntervalWindow>pastFirstElementInPane()
+            .pastFirstElementInPane()
             .plusDelayOf(Duration.millis(5)),
         FixedWindows.of(windowDuration));
 
@@ -87,7 +87,7 @@ public class AfterSynchronizedProcessingTimeTest {
     Duration windowDuration = Duration.millis(10);
     SimpleTriggerTester<IntervalWindow> tester = TriggerTester.forTrigger(
         AfterProcessingTime
-            .<IntervalWindow>pastFirstElementInPane()
+            .pastFirstElementInPane()
             .plusDelayOf(Duration.millis(5)),
         Sessions.withGapDuration(windowDuration));
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterWatermarkTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterWatermarkTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterWatermarkTest.java
index ef84714..be0ec1c 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterWatermarkTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterWatermarkTest.java
@@ -84,7 +84,7 @@ public class AfterWatermarkTest {
   @Test
   public void testEarlyAndAtWatermark() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterWatermark.<IntervalWindow>pastEndOfWindow()
+        AfterWatermark.pastEndOfWindow()
             .withEarlyFirings(mockEarly),
         FixedWindows.of(Duration.millis(100)));
 
@@ -104,7 +104,7 @@ public class AfterWatermarkTest {
   @Test
   public void testAtWatermarkAndLate() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterWatermark.<IntervalWindow>pastEndOfWindow()
+        AfterWatermark.pastEndOfWindow()
             .withLateFirings(mockLate),
         FixedWindows.of(Duration.millis(100)));
 
@@ -130,7 +130,7 @@ public class AfterWatermarkTest {
   @Test
   public void testEarlyAndAtWatermarkAndLate() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterWatermark.<IntervalWindow>pastEndOfWindow()
+        AfterWatermark.pastEndOfWindow()
             .withEarlyFirings(mockEarly)
             .withLateFirings(mockLate),
         FixedWindows.of(Duration.millis(100)));
@@ -162,8 +162,8 @@ public class AfterWatermarkTest {
   public void testOnMergeAlreadyFinished() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
-            AfterWatermark.<IntervalWindow>pastEndOfWindow(),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(1))),
+            AfterWatermark.pastEndOfWindow(),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);
@@ -209,8 +209,8 @@ public class AfterWatermarkTest {
   public void testOnMergeRewinds() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
-            AfterWatermark.<IntervalWindow>pastEndOfWindow(),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(1))),
+            AfterWatermark.pastEndOfWindow(),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);
@@ -255,9 +255,9 @@ public class AfterWatermarkTest {
   @Test
   public void testEarlyAndLateOnMergeAlreadyFinished() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterWatermark.<IntervalWindow>pastEndOfWindow()
-            .withEarlyFirings(AfterPane.<IntervalWindow>elementCountAtLeast(100))
-            .withLateFirings(AfterPane.<IntervalWindow>elementCountAtLeast(1)),
+        AfterWatermark.pastEndOfWindow()
+            .withEarlyFirings(AfterPane.elementCountAtLeast(100))
+            .withLateFirings(AfterPane.elementCountAtLeast(1)),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);
@@ -302,9 +302,9 @@ public class AfterWatermarkTest {
   @Test
   public void testEarlyAndLateOnMergeRewinds() throws Exception {
     tester = TriggerTester.forTrigger(
-        AfterWatermark.<IntervalWindow>pastEndOfWindow()
-            .withEarlyFirings(AfterPane.<IntervalWindow>elementCountAtLeast(100))
-            .withLateFirings(AfterPane.<IntervalWindow>elementCountAtLeast(1)),
+        AfterWatermark.pastEndOfWindow()
+            .withEarlyFirings(AfterPane.elementCountAtLeast(100))
+            .withLateFirings(AfterPane.elementCountAtLeast(1)),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/DefaultTriggerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/DefaultTriggerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/DefaultTriggerTest.java
index b31ad56..6ed1c81 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/DefaultTriggerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/DefaultTriggerTest.java
@@ -42,7 +42,7 @@ public class DefaultTriggerTest {
   @Test
   public void testDefaultTriggerFixedWindows() throws Exception {
     tester = TriggerTester.forTrigger(
-        DefaultTrigger.<IntervalWindow>of(),
+        DefaultTrigger.of(),
         FixedWindows.of(Duration.millis(100)));
 
     tester.injectElements(
@@ -79,7 +79,7 @@ public class DefaultTriggerTest {
   @Test
   public void testDefaultTriggerSlidingWindows() throws Exception {
     tester = TriggerTester.forTrigger(
-        DefaultTrigger.<IntervalWindow>of(),
+        DefaultTrigger.of(),
         SlidingWindows.of(Duration.millis(100)).every(Duration.millis(50)));
 
     tester.injectElements(
@@ -125,7 +125,7 @@ public class DefaultTriggerTest {
   @Test
   public void testDefaultTriggerSessions() throws Exception {
     tester = TriggerTester.forTrigger(
-        DefaultTrigger.<IntervalWindow>of(),
+        DefaultTrigger.of(),
         Sessions.withGapDuration(Duration.millis(100)));
 
     tester.injectElements(

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/NeverTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/NeverTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/NeverTest.java
index 222fe4e..fb2b4d5 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/NeverTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/NeverTest.java
@@ -41,7 +41,7 @@ public class NeverTest {
   public void setup() throws Exception {
     triggerTester =
         TriggerTester.forTrigger(
-            Never.<IntervalWindow>ever(), FixedWindows.of(Duration.standardMinutes(5)));
+            Never.ever(), FixedWindows.of(Duration.standardMinutes(5)));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/OrFinallyTriggerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/OrFinallyTriggerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/OrFinallyTriggerTest.java
index ea178a8..9397179 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/OrFinallyTriggerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/OrFinallyTriggerTest.java
@@ -47,8 +47,8 @@ public class OrFinallyTriggerTest {
   public void testActualFiresAndFinishes() throws Exception {
     tester = TriggerTester.forTrigger(
         new OrFinallyTrigger(
-            AfterPane.<IntervalWindow>elementCountAtLeast(2),
-            AfterPane.<IntervalWindow>elementCountAtLeast(100)),
+            AfterPane.elementCountAtLeast(2),
+            AfterPane.elementCountAtLeast(100)),
         FixedWindows.of(Duration.millis(100)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100));
@@ -74,8 +74,8 @@ public class OrFinallyTriggerTest {
   public void testActualFiresOnly() throws Exception {
     tester = TriggerTester.forTrigger(
         new OrFinallyTrigger(
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2)),
-            AfterPane.<IntervalWindow>elementCountAtLeast(100)),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(2)),
+            AfterPane.elementCountAtLeast(100)),
         FixedWindows.of(Duration.millis(100)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100));
@@ -106,9 +106,9 @@ public class OrFinallyTriggerTest {
   public void testShouldFireAfterMerge() throws Exception {
     tester = TriggerTester.forTrigger(
         AfterEach.inOrder(
-            AfterPane.<IntervalWindow>elementCountAtLeast(5)
-                .orFinally(AfterWatermark.<IntervalWindow>pastEndOfWindow()),
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(1))),
+            AfterPane.elementCountAtLeast(5)
+                .orFinally(AfterWatermark.pastEndOfWindow()),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(1))),
         Sessions.withGapDuration(Duration.millis(10)));
 
     // Finished the orFinally in the first window
@@ -144,8 +144,8 @@ public class OrFinallyTriggerTest {
   public void testActualFiresButUntilFinishes() throws Exception {
     tester = TriggerTester.forTrigger(
         new OrFinallyTrigger(
-            Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2)),
-                AfterPane.<IntervalWindow>elementCountAtLeast(3)),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(2)),
+                AfterPane.elementCountAtLeast(3)),
         FixedWindows.of(Duration.millis(10)));
 
     IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(10));

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/RepeatedlyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/RepeatedlyTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/RepeatedlyTest.java
index ddb9f9a..3a33182 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/RepeatedlyTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/RepeatedlyTest.java
@@ -110,7 +110,7 @@ public class RepeatedlyTest {
   @Test
   public void testShouldFireAfterMerge() throws Exception {
     tester = TriggerTester.forTrigger(
-        Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2)),
+        Repeatedly.forever(AfterPane.elementCountAtLeast(2)),
         Sessions.withGapDuration(Duration.millis(10)));
 
     tester.injectElements(1);
@@ -132,10 +132,10 @@ public class RepeatedlyTest {
     SimpleTriggerTester<GlobalWindow> tester =
         TriggerTester.forTrigger(
             Repeatedly.forever(
-                AfterFirst.<GlobalWindow>of(
-                    AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+                AfterFirst.of(
+                    AfterProcessingTime.pastFirstElementInPane()
                         .plusDelayOf(Duration.standardMinutes(15)),
-                    AfterPane.<GlobalWindow>elementCountAtLeast(5))),
+                    AfterPane.elementCountAtLeast(5))),
             new GlobalWindows());
 
     GlobalWindow window = GlobalWindow.INSTANCE;
@@ -154,10 +154,10 @@ public class RepeatedlyTest {
     SimpleTriggerTester<GlobalWindow> tester =
         TriggerTester.forTrigger(
             Repeatedly.forever(
-                AfterFirst.<GlobalWindow>of(
-                    AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+                AfterFirst.of(
+                    AfterProcessingTime.pastFirstElementInPane()
                         .plusDelayOf(Duration.standardMinutes(15)),
-                    AfterPane.<GlobalWindow>elementCountAtLeast(5))),
+                    AfterPane.elementCountAtLeast(5))),
             new GlobalWindows());
 
     GlobalWindow window = GlobalWindow.INSTANCE;
@@ -175,7 +175,7 @@ public class RepeatedlyTest {
   public void testRepeatedlyElementCount() throws Exception {
     SimpleTriggerTester<GlobalWindow> tester =
         TriggerTester.forTrigger(
-            Repeatedly.forever(AfterPane.<GlobalWindow>elementCountAtLeast(5)),
+            Repeatedly.forever(AfterPane.elementCountAtLeast(5)),
             new GlobalWindows());
 
     GlobalWindow window = GlobalWindow.INSTANCE;
@@ -194,7 +194,7 @@ public class RepeatedlyTest {
     SimpleTriggerTester<GlobalWindow> tester =
         TriggerTester.forTrigger(
             Repeatedly.forever(
-                    AfterProcessingTime.<GlobalWindow>pastFirstElementInPane()
+                    AfterProcessingTime.pastFirstElementInPane()
                         .plusDelayOf(Duration.standardMinutes(15))),
             new GlobalWindows());
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/beb0b36e/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnRunnerTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnRunnerTest.java
index 65b5ee6..f2036eb 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnRunnerTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/ReduceFnRunnerTest.java
@@ -499,7 +499,7 @@ public class ReduceFnRunnerTest {
   public void testPaneInfoAllStatesAfterWatermark() throws Exception {
     ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = ReduceFnTester.nonCombining(
         WindowingStrategy.of(FixedWindows.of(Duration.millis(10)))
-            .withTrigger(Repeatedly.<IntervalWindow>forever(AfterFirst.<IntervalWindow>of(
+            .withTrigger(Repeatedly.forever(AfterFirst.of(
                 AfterPane.elementCountAtLeast(2),
                 AfterWatermark.pastEndOfWindow())))
             .withMode(AccumulationMode.DISCARDING_FIRED_PANES)
@@ -605,7 +605,7 @@ public class ReduceFnRunnerTest {
   public void testPaneInfoAllStatesAfterWatermarkAccumulating() throws Exception {
     ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = ReduceFnTester.nonCombining(
         WindowingStrategy.of(FixedWindows.of(Duration.millis(10)))
-            .withTrigger(Repeatedly.<IntervalWindow>forever(AfterFirst.<IntervalWindow>of(
+            .withTrigger(Repeatedly.forever(AfterFirst.of(
                 AfterPane.elementCountAtLeast(2),
                 AfterWatermark.pastEndOfWindow())))
             .withMode(AccumulationMode.ACCUMULATING_FIRED_PANES)
@@ -658,7 +658,7 @@ public class ReduceFnRunnerTest {
     ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = ReduceFnTester.nonCombining(
         WindowingStrategy.of(FixedWindows.of(Duration.millis(10)))
             .withTrigger(
-                Repeatedly.<IntervalWindow>forever(AfterPane.elementCountAtLeast(2))
+                Repeatedly.forever(AfterPane.elementCountAtLeast(2))
                     .orFinally(AfterWatermark.pastEndOfWindow()))
             .withMode(AccumulationMode.DISCARDING_FIRED_PANES)
             .withAllowedLateness(Duration.millis(100))
@@ -1075,12 +1075,12 @@ public class ReduceFnRunnerTest {
         ReduceFnTester.combining(FixedWindows.of(Duration.millis(10)),
             AfterEach.<IntervalWindow>inOrder(
                 Repeatedly
-                    .<IntervalWindow>forever(
-                        AfterProcessingTime.<IntervalWindow>pastFirstElementInPane().plusDelayOf(
+                    .forever(
+                        AfterProcessingTime.pastFirstElementInPane().plusDelayOf(
                             new Duration(5)))
                     .orFinally(AfterWatermark.pastEndOfWindow()),
-                Repeatedly.<IntervalWindow>forever(
-                    AfterProcessingTime.<IntervalWindow>pastFirstElementInPane().plusDelayOf(
+                Repeatedly.forever(
+                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(
                         new Duration(25)))),
             AccumulationMode.ACCUMULATING_FIRED_PANES, new Sum.SumIntegerFn().<String>asKeyedFn(),
             VarIntCoder.of(), Duration.millis(100));
@@ -1125,12 +1125,12 @@ public class ReduceFnRunnerTest {
         ReduceFnTester.combining(FixedWindows.of(Duration.millis(10)),
             AfterEach.<IntervalWindow>inOrder(
                 Repeatedly
-                    .<IntervalWindow>forever(
-                        AfterProcessingTime.<IntervalWindow>pastFirstElementInPane().plusDelayOf(
+                    .forever(
+                        AfterProcessingTime.pastFirstElementInPane().plusDelayOf(
                             new Duration(5)))
                     .orFinally(AfterWatermark.pastEndOfWindow()),
-                Repeatedly.<IntervalWindow>forever(
-                    AfterProcessingTime.<IntervalWindow>pastFirstElementInPane().plusDelayOf(
+                Repeatedly.forever(
+                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(
                         new Duration(25)))),
             AccumulationMode.ACCUMULATING_FIRED_PANES, new Sum.SumIntegerFn().<String>asKeyedFn(),
             VarIntCoder.of(), Duration.millis(100));


Mime
View raw message