beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/2] beam git commit: Removed OnceTriggerStateMachine
Date Thu, 29 Jun 2017 03:31:51 GMT
Repository: beam
Updated Branches:
  refs/heads/master 0d4c85dd1 -> b1ece0196


Removed OnceTriggerStateMachine


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

Branch: refs/heads/master
Commit: 4f9820b1f24103831f3b0a4f5783f9ca726f8cd7
Parents: 0d4c85d
Author: = <=>
Authored: Wed Jun 14 20:11:49 2017 -0400
Committer: Kenneth Knowles <klk@google.com>
Committed: Wed Jun 28 20:14:24 2017 -0700

----------------------------------------------------------------------
 .../core/triggers/AfterAllStateMachine.java     | 25 ++++++++----------
 .../AfterDelayFromFirstElementStateMachine.java |  6 ++---
 .../core/triggers/AfterFirstStateMachine.java   | 20 +++++++--------
 .../core/triggers/AfterPaneStateMachine.java    |  6 ++---
 .../triggers/AfterWatermarkStateMachine.java    |  7 ++---
 .../triggers/ExecutableTriggerStateMachine.java | 23 +++--------------
 .../core/triggers/NeverStateMachine.java        |  5 ++--
 .../core/triggers/TriggerStateMachine.java      | 27 --------------------
 .../triggers/AfterFirstStateMachineTest.java    |  5 ++--
 .../AfterWatermarkStateMachineTest.java         |  7 +++--
 .../core/triggers/StubTriggerStateMachine.java  |  7 +++--
 11 files changed, 44 insertions(+), 94 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java
index 0f0c17c..3530ed1 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java
@@ -23,7 +23,6 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import java.util.Arrays;
 import java.util.List;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.annotations.Experimental;
 
 /**
@@ -31,7 +30,7 @@ import org.apache.beam.sdk.annotations.Experimental;
  * have fired.
  */
 @Experimental(Experimental.Kind.TRIGGER)
-public class AfterAllStateMachine extends OnceTriggerStateMachine {
+public class AfterAllStateMachine extends TriggerStateMachine {
 
   private AfterAllStateMachine(List<TriggerStateMachine> subTriggers) {
     super(subTriggers);
@@ -42,11 +41,11 @@ public class AfterAllStateMachine extends OnceTriggerStateMachine {
    * Returns an {@code AfterAll} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static OnceTriggerStateMachine of(TriggerStateMachine... triggers) {
+  public static TriggerStateMachine of(TriggerStateMachine... triggers) {
     return new AfterAllStateMachine(Arrays.<TriggerStateMachine>asList(triggers));
   }
 
-  public static OnceTriggerStateMachine of(Iterable<? extends TriggerStateMachine>
triggers) {
+  public static TriggerStateMachine of(Iterable<? extends TriggerStateMachine> triggers)
{
     return new AfterAllStateMachine(ImmutableList.copyOf(triggers));
   }
 
@@ -78,24 +77,21 @@ public class AfterAllStateMachine extends OnceTriggerStateMachine {
    */
   @Override
   public boolean shouldFire(TriggerContext context) throws Exception {
-    for (ExecutableTriggerStateMachine subtrigger : context.trigger().subTriggers()) {
-      if (!context.forTrigger(subtrigger).trigger().isFinished()
-          && !subtrigger.invokeShouldFire(context)) {
+    for (ExecutableTriggerStateMachine subTrigger : context.trigger().subTriggers()) {
+      if (!context.forTrigger(subTrigger).trigger().isFinished()
+          && !subTrigger.invokeShouldFire(context)) {
         return false;
       }
     }
     return true;
   }
 
-  /**
-   * Invokes {@link #onFire} for all subtriggers, eliding redundant calls to {@link #shouldFire}
-   * because they all must be ready to fire.
-   */
   @Override
-  public void onOnlyFiring(TriggerContext context) throws Exception {
-    for (ExecutableTriggerStateMachine subtrigger : context.trigger().subTriggers()) {
-      subtrigger.invokeOnFire(context);
+  public void onFire(TriggerContext context) throws Exception {
+    for (ExecutableTriggerStateMachine subTrigger : context.trigger().subTriggers()) {
+      subTrigger.invokeOnFire(context);
     }
+    context.trigger().setFinished(true);
   }
 
   @Override
@@ -103,7 +99,6 @@ public class AfterAllStateMachine extends OnceTriggerStateMachine {
     StringBuilder builder = new StringBuilder("AfterAll.of(");
     Joiner.on(", ").appendTo(builder, subTriggers);
     builder.append(")");
-
     return builder.toString();
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
index 8d8d0de..06c2066 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
@@ -27,7 +27,6 @@ import org.apache.beam.runners.core.StateAccessor;
 import org.apache.beam.runners.core.StateMerging;
 import org.apache.beam.runners.core.StateTag;
 import org.apache.beam.runners.core.StateTags;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.coders.InstantCoder;
 import org.apache.beam.sdk.state.CombiningState;
@@ -50,7 +49,7 @@ import org.joda.time.format.PeriodFormatter;
 // This class should be inlined to subclasses and deleted, simplifying them too
 // https://issues.apache.org/jira/browse/BEAM-1486
 @Experimental(Experimental.Kind.TRIGGER)
-public abstract class AfterDelayFromFirstElementStateMachine extends OnceTriggerStateMachine
{
+public abstract class AfterDelayFromFirstElementStateMachine extends TriggerStateMachine
{
 
   protected static final List<SerializableFunction<Instant, Instant>> IDENTITY
=
       ImmutableList.<SerializableFunction<Instant, Instant>>of();
@@ -237,8 +236,9 @@ public abstract class AfterDelayFromFirstElementStateMachine extends OnceTrigger
   }
 
   @Override
-  protected void onOnlyFiring(TriggerStateMachine.TriggerContext context) throws Exception
{
+  public final void onFire(TriggerContext context) throws Exception {
     clear(context);
+    context.trigger().setFinished(true);
   }
 
   protected Instant computeTargetTimestamp(Instant time) {

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java
index 840a65c..58c24c5 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java
@@ -23,7 +23,6 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import java.util.Arrays;
 import java.util.List;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.annotations.Experimental;
 
 /**
@@ -31,7 +30,7 @@ import org.apache.beam.sdk.annotations.Experimental;
  * sub-triggers have fired.
  */
 @Experimental(Experimental.Kind.TRIGGER)
-public class AfterFirstStateMachine extends OnceTriggerStateMachine {
+public class AfterFirstStateMachine extends TriggerStateMachine {
 
   AfterFirstStateMachine(List<TriggerStateMachine> subTriggers) {
     super(subTriggers);
@@ -42,12 +41,12 @@ public class AfterFirstStateMachine extends OnceTriggerStateMachine {
    * Returns an {@code AfterFirst} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static OnceTriggerStateMachine of(
+  public static TriggerStateMachine of(
       TriggerStateMachine... triggers) {
     return new AfterFirstStateMachine(Arrays.<TriggerStateMachine>asList(triggers));
   }
 
-  public static OnceTriggerStateMachine of(
+  public static TriggerStateMachine of(
       Iterable<? extends TriggerStateMachine> triggers) {
     return new AfterFirstStateMachine(ImmutableList.copyOf(triggers));
   }
@@ -79,18 +78,19 @@ public class AfterFirstStateMachine extends OnceTriggerStateMachine {
   }
 
   @Override
-  protected void onOnlyFiring(TriggerContext context) throws Exception {
-    for (ExecutableTriggerStateMachine subtrigger : context.trigger().subTriggers()) {
-      TriggerContext subContext = context.forTrigger(subtrigger);
-      if (subtrigger.invokeShouldFire(subContext)) {
+  public void onFire(TriggerContext context) throws Exception {
+    for (ExecutableTriggerStateMachine subTrigger : context.trigger().subTriggers()) {
+      TriggerContext subContext = context.forTrigger(subTrigger);
+      if (subTrigger.invokeShouldFire(subContext)) {
         // If the trigger is ready to fire, then do whatever it needs to do.
-        subtrigger.invokeOnFire(subContext);
+        subTrigger.invokeOnFire(subContext);
       } else {
         // If the trigger is not ready to fire, it is nonetheless true that whatever
         // pending pane it was tracking is now gone.
-        subtrigger.invokeClear(subContext);
+        subTrigger.invokeClear(subContext);
       }
     }
+    context.trigger().setFinished(true);
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java
index b9fbac3..1ce035a 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java
@@ -23,7 +23,6 @@ import org.apache.beam.runners.core.StateAccessor;
 import org.apache.beam.runners.core.StateMerging;
 import org.apache.beam.runners.core.StateTag;
 import org.apache.beam.runners.core.StateTags;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.coders.VarLongCoder;
 import org.apache.beam.sdk.state.CombiningState;
@@ -33,7 +32,7 @@ import org.apache.beam.sdk.transforms.Sum;
  * {@link TriggerStateMachine}s that fire based on properties of the elements in the current
pane.
  */
 @Experimental(Experimental.Kind.TRIGGER)
-public class AfterPaneStateMachine extends OnceTriggerStateMachine {
+public class AfterPaneStateMachine extends TriggerStateMachine {
 
 private static final StateTag<CombiningState<Long, long[], Long>>
       ELEMENTS_IN_PANE_TAG =
@@ -130,7 +129,8 @@ private static final StateTag<CombiningState<Long, long[], Long>>
   }
 
   @Override
-  protected void onOnlyFiring(TriggerStateMachine.TriggerContext context) throws Exception
{
+  public void onFire(TriggerStateMachine.TriggerContext context) throws Exception {
     clear(context);
+    context.trigger().setFinished(true);
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java
index c9eee15..509c96b 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java
@@ -22,7 +22,6 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import com.google.common.collect.ImmutableList;
 import java.util.Objects;
 import javax.annotation.Nullable;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.state.TimeDomain;
 
@@ -242,7 +241,7 @@ public class AfterWatermarkStateMachine {
   /**
    * A watermark trigger targeted relative to the end of the window.
    */
-  public static class FromEndOfWindow extends OnceTriggerStateMachine {
+  public static class FromEndOfWindow extends TriggerStateMachine {
 
     private FromEndOfWindow() {
       super(null);
@@ -319,6 +318,8 @@ public class AfterWatermarkStateMachine {
     }
 
     @Override
-    protected void onOnlyFiring(TriggerStateMachine.TriggerContext context) throws Exception
{ }
+    public void onFire(TriggerStateMachine.TriggerContext context) throws Exception {
+      context.trigger().setFinished(true);
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/ExecutableTriggerStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/ExecutableTriggerStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/ExecutableTriggerStateMachine.java
index c4d89c2..cdcff64 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/ExecutableTriggerStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/ExecutableTriggerStateMachine.java
@@ -23,7 +23,6 @@ import static com.google.common.base.Preconditions.checkState;
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 
 /**
@@ -46,17 +45,14 @@ public class ExecutableTriggerStateMachine implements Serializable {
 
   private static <W extends BoundedWindow> ExecutableTriggerStateMachine create(
       TriggerStateMachine trigger, int nextUnusedIndex) {
-    if (trigger instanceof OnceTriggerStateMachine) {
-      return new ExecutableOnceTriggerStateMachine(
-          (OnceTriggerStateMachine) trigger, nextUnusedIndex);
-    } else {
+
       return new ExecutableTriggerStateMachine(trigger, nextUnusedIndex);
-    }
+
   }
 
   public static <W extends BoundedWindow> ExecutableTriggerStateMachine createForOnceTrigger(
-      OnceTriggerStateMachine trigger, int nextUnusedIndex) {
-    return new ExecutableOnceTriggerStateMachine(trigger, nextUnusedIndex);
+      TriggerStateMachine trigger, int nextUnusedIndex) {
+    return new ExecutableTriggerStateMachine(trigger, nextUnusedIndex);
   }
 
   private ExecutableTriggerStateMachine(TriggerStateMachine trigger, int nextUnusedIndex)
{
@@ -146,15 +142,4 @@ public class ExecutableTriggerStateMachine implements Serializable {
   public void invokeClear(TriggerStateMachine.TriggerContext c) throws Exception {
     trigger.clear(c.forTrigger(this));
   }
-
-  /**
-   * {@link ExecutableTriggerStateMachine} that enforces the fact that the trigger should
always
-   * FIRE_AND_FINISH and never just FIRE.
-   */
-  private static class ExecutableOnceTriggerStateMachine extends ExecutableTriggerStateMachine
{
-
-    public ExecutableOnceTriggerStateMachine(OnceTriggerStateMachine trigger, int nextUnusedIndex)
{
-      super(trigger, nextUnusedIndex);
-    }
-  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/NeverStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/NeverStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/NeverStateMachine.java
index f32c7a8..f8c5e8b 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/NeverStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/NeverStateMachine.java
@@ -17,7 +17,6 @@
  */
 package org.apache.beam.runners.core.triggers;
 
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.transforms.GroupByKey;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 
@@ -27,7 +26,7 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
  * <p>Using this trigger will only produce output when the watermark passes the end
of the
  * {@link BoundedWindow window} plus the allowed lateness.
  */
-public final class NeverStateMachine extends OnceTriggerStateMachine {
+public final class NeverStateMachine extends TriggerStateMachine {
   /**
    * Returns a trigger which never fires. Output will be produced from the using {@link GroupByKey}
    * when the {@link BoundedWindow} closes.
@@ -53,7 +52,7 @@ public final class NeverStateMachine extends OnceTriggerStateMachine {
   }
 
   @Override
-  protected void onOnlyFiring(TriggerStateMachine.TriggerContext context) {
+  public void onFire(TriggerStateMachine.TriggerContext context) {
     throw new UnsupportedOperationException(
         String.format("%s should never fire", getClass().getSimpleName()));
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachine.java
index 6a2cf0c..880aa48 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachine.java
@@ -453,35 +453,8 @@ public abstract class TriggerStateMachine implements Serializable {
    * }
    * </pre>
    *
-   * <p>Note that if {@code t1} is {@link OnceTriggerStateMachine}, then {@code t1.orFinally(t2)}
is
-   * the same as {@code AfterFirst.of(t1, t2)}.
    */
   public TriggerStateMachine orFinally(TriggerStateMachine until) {
     return new OrFinallyStateMachine(this, until);
   }
-
-  /**
-   * {@link TriggerStateMachine}s that are guaranteed to fire at most once should extend
from this,
-   * rather than the general {@link TriggerStateMachine} class to indicate that behavior.
-   */
-  public abstract static class OnceTriggerStateMachine extends TriggerStateMachine {
-    protected OnceTriggerStateMachine(List<TriggerStateMachine> subTriggers) {
-      super(subTriggers);
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    @Override
-    public final void onFire(TriggerContext context) throws Exception {
-      onOnlyFiring(context);
-      context.trigger().setFinished(true);
-    }
-
-    /**
-     * Called exactly once by {@link #onFire} when the trigger is fired. By default,
-     * invokes {@link #onFire} on all subtriggers for which {@link #shouldFire} is {@code
true}.
-     */
-    protected abstract void onOnlyFiring(TriggerContext context) throws Exception;
-  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachineTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachineTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachineTest.java
index 453c8ff..2be90de 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachineTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachineTest.java
@@ -21,7 +21,6 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.when;
 
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.runners.core.triggers.TriggerStateMachineTester.SimpleTriggerStateMachineTester;
 import org.apache.beam.sdk.transforms.windowing.FixedWindows;
 import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
@@ -42,8 +41,8 @@ import org.mockito.MockitoAnnotations;
 @RunWith(JUnit4.class)
 public class AfterFirstStateMachineTest {
 
-  @Mock private OnceTriggerStateMachine mockTrigger1;
-  @Mock private OnceTriggerStateMachine mockTrigger2;
+  @Mock private TriggerStateMachine mockTrigger1;
+  @Mock private TriggerStateMachine mockTrigger2;
   private SimpleTriggerStateMachineTester<IntervalWindow> tester;
   private static TriggerStateMachine.TriggerContext anyTriggerContext() {
     return Mockito.<TriggerStateMachine.TriggerContext>any();

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachineTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachineTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachineTest.java
index e4d10a0..45a5cfb 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachineTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachineTest.java
@@ -25,7 +25,6 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnMergeContext;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.runners.core.triggers.TriggerStateMachineTester.SimpleTriggerStateMachineTester;
 import org.apache.beam.sdk.transforms.windowing.FixedWindows;
 import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
@@ -46,8 +45,8 @@ import org.mockito.MockitoAnnotations;
 @RunWith(JUnit4.class)
 public class AfterWatermarkStateMachineTest {
 
-  @Mock private OnceTriggerStateMachine mockEarly;
-  @Mock private OnceTriggerStateMachine mockLate;
+  @Mock private TriggerStateMachine mockEarly;
+  @Mock private TriggerStateMachine mockLate;
 
   private SimpleTriggerStateMachineTester<IntervalWindow> tester;
   private static TriggerStateMachine.TriggerContext anyTriggerContext() {
@@ -70,7 +69,7 @@ public class AfterWatermarkStateMachineTest {
     MockitoAnnotations.initMocks(this);
   }
 
-  public void testRunningAsTrigger(OnceTriggerStateMachine mockTrigger, IntervalWindow window)
+  public void testRunningAsTrigger(TriggerStateMachine mockTrigger, IntervalWindow window)
       throws Exception {
 
     // Don't fire due to mock saying no

http://git-wip-us.apache.org/repos/asf/beam/blob/4f9820b1/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/StubTriggerStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/StubTriggerStateMachine.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/StubTriggerStateMachine.java
index 4512848..1bc757e 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/StubTriggerStateMachine.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/StubTriggerStateMachine.java
@@ -18,12 +18,11 @@
 package org.apache.beam.runners.core.triggers;
 
 import com.google.common.collect.Lists;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 
 /**
- * No-op {@link OnceTriggerStateMachine} implementation for testing.
+ * No-op {@link TriggerStateMachine} implementation for testing.
  */
-abstract class StubTriggerStateMachine extends OnceTriggerStateMachine {
+abstract class StubTriggerStateMachine extends TriggerStateMachine {
   /**
    * Create a stub {@link TriggerStateMachine} instance which returns the specified name
on {@link
    * #toString()}.
@@ -42,7 +41,7 @@ abstract class StubTriggerStateMachine extends OnceTriggerStateMachine {
   }
 
   @Override
-  protected void onOnlyFiring(TriggerContext context) throws Exception {
+  public void onFire(TriggerContext context) throws Exception {
   }
 
   @Override


Mime
View raw message