beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [4/5] beam git commit: Build trigger state machine from Runner API Trigger proto directly
Date Fri, 24 Feb 2017 15:28:52 GMT
Build trigger state machine from Runner API Trigger proto directly


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

Branch: refs/heads/master
Commit: 27a482baf498bab8c931670adec5134c0bdf08ac
Parents: bd8b72c
Author: Kenneth Knowles <klk@google.com>
Authored: Fri Feb 17 16:05:13 2017 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Fri Feb 24 07:16:36 2017 -0800

----------------------------------------------------------------------
 .../GroupAlsoByWindowViaOutputBufferDoFn.java   |   4 +-
 .../core/GroupAlsoByWindowViaWindowSetDoFn.java |   4 +-
 .../GroupAlsoByWindowViaWindowSetNewDoFn.java   |   4 +-
 .../core/triggers/AfterAllStateMachine.java     |   2 +-
 .../core/triggers/AfterFirstStateMachine.java   |   2 +-
 .../triggers/AfterWatermarkStateMachine.java    |  14 +-
 .../core/triggers/OrFinallyStateMachine.java    |   2 +-
 .../core/triggers/TriggerStateMachine.java      |   2 +-
 .../core/triggers/TriggerStateMachines.java     | 272 ++++++-------------
 .../beam/runners/core/ReduceFnTester.java       |   8 +-
 .../core/triggers/TriggerStateMachinesTest.java | 160 +++++++----
 .../GroupAlsoByWindowEvaluatorFactory.java      |   4 +-
 .../translation/SparkGroupAlsoByWindowFn.java   |  40 +--
 13 files changed, 243 insertions(+), 275 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaOutputBufferDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaOutputBufferDoFn.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaOutputBufferDoFn.java
index 5b2e051..e3ce1ef 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaOutputBufferDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaOutputBufferDoFn.java
@@ -22,6 +22,7 @@ import java.util.List;
 import org.apache.beam.runners.core.triggers.ExecutableTriggerStateMachine;
 import org.apache.beam.runners.core.triggers.TriggerStateMachines;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.joda.time.Instant;
@@ -63,7 +64,8 @@ public class GroupAlsoByWindowViaOutputBufferDoFn<K, InputT, OutputT,
W extends
             key,
             strategy,
             ExecutableTriggerStateMachine.create(
-                TriggerStateMachines.stateMachineForTrigger(strategy.getTrigger())),
+                TriggerStateMachines.stateMachineForTrigger(
+                    Triggers.toProto(strategy.getTrigger()))),
             stateInternals,
             timerInternals,
             WindowingInternalsAdapters.outputWindowedValue(c.windowingInternals()),

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
index ac0b1ab..8dc1502 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
@@ -22,6 +22,7 @@ import org.apache.beam.runners.core.triggers.TriggerStateMachines;
 import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.values.KV;
@@ -77,7 +78,8 @@ public class GroupAlsoByWindowViaWindowSetDoFn<
             key,
             windowingStrategy,
             ExecutableTriggerStateMachine.create(
-                TriggerStateMachines.stateMachineForTrigger(windowingStrategy.getTrigger())),
+                TriggerStateMachines.stateMachineForTrigger(
+                    Triggers.toProto(windowingStrategy.getTrigger()))),
             stateInternals,
             timerInternals,
             WindowingInternalsAdapters.outputWindowedValue(c.windowingInternals()),

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
index 7b65a0b..444f8fe 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetNewDoFn.java
@@ -25,6 +25,7 @@ import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.util.SideInputReader;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
 import org.apache.beam.sdk.util.WindowedValue;
@@ -128,7 +129,8 @@ public class GroupAlsoByWindowViaWindowSetNewDoFn<
             key,
             windowingStrategy,
             ExecutableTriggerStateMachine.create(
-                TriggerStateMachines.stateMachineForTrigger(windowingStrategy.getTrigger())),
+                TriggerStateMachines.stateMachineForTrigger(
+                    Triggers.toProto(windowingStrategy.getTrigger()))),
             stateInternals,
             timerInternals,
             outputWindowedValue(),

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/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 12cbc3d..0f0c17c 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
@@ -42,7 +42,7 @@ public class AfterAllStateMachine extends OnceTriggerStateMachine {
    * Returns an {@code AfterAll} {@code Trigger} with the given subtriggers.
    */
   @SafeVarargs
-  public static OnceTriggerStateMachine of(OnceTriggerStateMachine... triggers) {
+  public static OnceTriggerStateMachine of(TriggerStateMachine... triggers) {
     return new AfterAllStateMachine(Arrays.<TriggerStateMachine>asList(triggers));
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/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 f4b305e..840a65c 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
@@ -43,7 +43,7 @@ public class AfterFirstStateMachine extends OnceTriggerStateMachine {
    */
   @SafeVarargs
   public static OnceTriggerStateMachine of(
-      OnceTriggerStateMachine... triggers) {
+      TriggerStateMachine... triggers) {
     return new AfterFirstStateMachine(Arrays.<TriggerStateMachine>asList(triggers));
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/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 e83c2f8..0b12005 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
@@ -78,13 +78,13 @@ public class AfterWatermarkStateMachine {
     private static final int EARLY_INDEX = 0;
     private static final int LATE_INDEX = 1;
 
-    private final OnceTriggerStateMachine earlyTrigger;
+    private final TriggerStateMachine earlyTrigger;
     @Nullable
-    private final OnceTriggerStateMachine lateTrigger;
+    private final TriggerStateMachine lateTrigger;
 
     @SuppressWarnings("unchecked")
     private AfterWatermarkEarlyAndLate(
-        OnceTriggerStateMachine earlyTrigger, OnceTriggerStateMachine lateTrigger) {
+        TriggerStateMachine earlyTrigger, TriggerStateMachine lateTrigger) {
       super(
           lateTrigger == null
               ? ImmutableList.<TriggerStateMachine>of(earlyTrigger)
@@ -93,11 +93,11 @@ public class AfterWatermarkStateMachine {
       this.lateTrigger = lateTrigger;
     }
 
-    public AfterWatermarkEarlyAndLate withEarlyFirings(OnceTriggerStateMachine earlyTrigger)
{
+    public AfterWatermarkEarlyAndLate withEarlyFirings(TriggerStateMachine earlyTrigger)
{
       return new AfterWatermarkEarlyAndLate(earlyTrigger, lateTrigger);
     }
 
-    public AfterWatermarkEarlyAndLate withLateFirings(OnceTriggerStateMachine lateTrigger)
{
+    public AfterWatermarkEarlyAndLate withLateFirings(TriggerStateMachine lateTrigger) {
       return new AfterWatermarkEarlyAndLate(earlyTrigger, lateTrigger);
     }
 
@@ -254,7 +254,7 @@ public class AfterWatermarkStateMachine {
      * Creates a new {@code Trigger} like the this, except that it fires repeatedly whenever
      * the given {@code Trigger} fires before the watermark has passed the end of the window.
      */
-    public AfterWatermarkEarlyAndLate withEarlyFirings(OnceTriggerStateMachine earlyFirings)
{
+    public AfterWatermarkEarlyAndLate withEarlyFirings(TriggerStateMachine earlyFirings)
{
       checkNotNull(earlyFirings, "Must specify the trigger to use for early firings");
       return new AfterWatermarkEarlyAndLate(earlyFirings, null);
     }
@@ -263,7 +263,7 @@ public class AfterWatermarkStateMachine {
      * Creates a new {@code Trigger} like the this, except that it fires repeatedly whenever
      * the given {@code Trigger} fires after the watermark has passed the end of the window.
      */
-    public AfterWatermarkEarlyAndLate withLateFirings(OnceTriggerStateMachine lateFirings)
{
+    public AfterWatermarkEarlyAndLate withLateFirings(TriggerStateMachine lateFirings) {
       checkNotNull(lateFirings, "Must specify the trigger to use for late firings");
       return new AfterWatermarkEarlyAndLate(NeverStateMachine.ever(), lateFirings);
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java
index f9ec5e7..e3bfb4f 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java
@@ -29,7 +29,7 @@ class OrFinallyStateMachine extends TriggerStateMachine {
   private static final int UNTIL = 1;
 
   @VisibleForTesting
-  OrFinallyStateMachine(TriggerStateMachine actual, OnceTriggerStateMachine until) {
+  OrFinallyStateMachine(TriggerStateMachine actual, TriggerStateMachine until) {
     super(Arrays.asList(actual, until));
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/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 8b8d737..d622ac9 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
@@ -456,7 +456,7 @@ public abstract class TriggerStateMachine implements Serializable {
    * <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(OnceTriggerStateMachine until) {
+  public TriggerStateMachine orFinally(TriggerStateMachine until) {
     return new OrFinallyStateMachine(this, until);
   }
 

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
index b13ac40..1088435 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
@@ -17,32 +17,11 @@
  */
 package org.apache.beam.runners.core.triggers;
 
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.collect.FluentIterable;
-import com.google.common.collect.Lists;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
+import java.util.ArrayList;
 import java.util.List;
-import javax.annotation.Nonnull;
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.windowing.AfterAll;
-import org.apache.beam.sdk.transforms.windowing.AfterEach;
-import org.apache.beam.sdk.transforms.windowing.AfterFirst;
-import org.apache.beam.sdk.transforms.windowing.AfterPane;
-import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
-import org.apache.beam.sdk.transforms.windowing.AfterSynchronizedProcessingTime;
-import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
-import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
-import org.apache.beam.sdk.transforms.windowing.Never.NeverTrigger;
-import org.apache.beam.sdk.transforms.windowing.OrFinallyTrigger;
-import org.apache.beam.sdk.transforms.windowing.Repeatedly;
-import org.apache.beam.sdk.transforms.windowing.TimestampTransform;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi;
 import org.apache.beam.sdk.transforms.windowing.Trigger;
-import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
-import org.apache.beam.sdk.util.ReshuffleTrigger;
-import org.apache.beam.sdk.util.TimeDomain;
+import org.joda.time.Duration;
 import org.joda.time.Instant;
 
 /** Translates a {@link Trigger} to a {@link TriggerStateMachine}. */
@@ -50,187 +29,96 @@ public class TriggerStateMachines {
 
   private TriggerStateMachines() {}
 
-  @VisibleForTesting static final StateMachineConverter CONVERTER = new StateMachineConverter();
-
-  public static TriggerStateMachine stateMachineForTrigger(Trigger trigger) {
-    return CONVERTER.evaluateTrigger(trigger);
-  }
-
-  public static OnceTriggerStateMachine stateMachineForOnceTrigger(OnceTrigger trigger) {
-    return CONVERTER.evaluateOnceTrigger(trigger);
-  }
-
-  @VisibleForTesting
-  static class StateMachineConverter {
-
-    public TriggerStateMachine evaluateTrigger(Trigger trigger) {
-      Method evaluationMethod = getEvaluationMethod(trigger.getClass());
-      return tryEvaluate(evaluationMethod, trigger);
-    }
-
-    public OnceTriggerStateMachine evaluateOnceTrigger(OnceTrigger trigger) {
-      Method evaluationMethod = getEvaluationMethod(trigger.getClass());
-      return (OnceTriggerStateMachine) tryEvaluate(evaluationMethod, trigger);
-    }
-
-    private TriggerStateMachine tryEvaluate(Method evaluationMethod, Trigger trigger) {
-      try {
-        return (TriggerStateMachine) evaluationMethod.invoke(this, trigger);
-      } catch (InvocationTargetException exc) {
-        if (exc.getCause() instanceof RuntimeException) {
-          throw (RuntimeException) exc.getCause();
-        } else {
-          throw new RuntimeException(exc.getCause());
-        }
-      } catch (IllegalAccessException exc) {
-        throw new IllegalStateException(
-            String.format("Internal error: could not invoke %s", evaluationMethod));
-      }
-    }
-
-    private Method getEvaluationMethod(Class<?> clazz) {
-      try {
-        return getClass().getDeclaredMethod("evaluateSpecific", clazz);
-      } catch (NoSuchMethodException exc) {
-        throw new UnsupportedOperationException(
-            String.format(
-                "Cannot translate trigger class %s to a state machine.", clazz.getCanonicalName()),
-            exc);
-      }
-    }
-
-    private TriggerStateMachine evaluateSpecific(DefaultTrigger v) {
-      return DefaultTriggerStateMachine.of();
-    }
-
-    private TriggerStateMachine evaluateSpecific(ReshuffleTrigger v) {
-      return new ReshuffleTriggerStateMachine();
+  public static TriggerStateMachine stateMachineForTrigger(RunnerApi.Trigger trigger) {
+    switch (trigger.getTriggerCase()) {
+      case AFTER_ALL:
+        return AfterAllStateMachine.of(
+            stateMachinesForTriggers(trigger.getAfterAll().getSubtriggersList()));
+      case AFTER_ANY:
+        return AfterFirstStateMachine.of(
+            stateMachinesForTriggers(trigger.getAfterAny().getSubtriggersList()));
+      case AFTER_END_OF_WIDOW:
+        return stateMachineForAfterEndOfWindow(trigger.getAfterEndOfWidow());
+      case ELEMENT_COUNT:
+        return AfterPaneStateMachine.elementCountAtLeast(
+            trigger.getElementCount().getElementCount());
+      case AFTER_SYNCHRONIZED_PROCESSING_TIME:
+        return AfterSynchronizedProcessingTimeStateMachine.ofFirstElement();
+      case DEFAULT:
+        return DefaultTriggerStateMachine.of();
+      case NEVER:
+        return NeverStateMachine.ever();
+      case ALWAYS:
+        return ReshuffleTriggerStateMachine.create();
+      case OR_FINALLY:
+        return stateMachineForTrigger(trigger.getOrFinally().getMain())
+            .orFinally(stateMachineForTrigger(trigger.getOrFinally().getFinally()));
+      case REPEAT:
+        return RepeatedlyStateMachine.forever(
+            stateMachineForTrigger(trigger.getRepeat().getSubtrigger()));
+      case AFTER_EACH:
+        return AfterEachStateMachine.inOrder(
+            stateMachinesForTriggers(trigger.getAfterEach().getSubtriggersList()));
+      case AFTER_PROCESSING_TIME:
+        return stateMachineForAfterProcessingTime(trigger.getAfterProcessingTime());
+      case TRIGGER_NOT_SET:
+        throw new IllegalArgumentException(
+            String.format("Required field 'trigger' not set on %s", trigger));
+      default:
+        throw new IllegalArgumentException(String.format("Unknown trigger type %s", trigger));
     }
+  }
 
-    private OnceTriggerStateMachine evaluateSpecific(AfterWatermark.FromEndOfWindow v) {
+  private static TriggerStateMachine stateMachineForAfterEndOfWindow(
+      RunnerApi.Trigger.AfterEndOfWindow trigger) {
+    if (!trigger.hasEarlyFirings() && !trigger.hasLateFirings()) {
       return AfterWatermarkStateMachine.pastEndOfWindow();
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(NeverTrigger v) {
-      return NeverStateMachine.ever();
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(AfterSynchronizedProcessingTime v) {
-      return AfterSynchronizedProcessingTimeStateMachine.ofFirstElement();
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(AfterFirst v) {
-      List<OnceTriggerStateMachine> subStateMachines =
-          Lists.newArrayListWithCapacity(v.subTriggers().size());
-      for (Trigger subtrigger : v.subTriggers()) {
-        subStateMachines.add(stateMachineForOnceTrigger((OnceTrigger) subtrigger));
-      }
-      return AfterFirstStateMachine.of(subStateMachines);
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(AfterAll v) {
-      List<OnceTriggerStateMachine> subStateMachines =
-          Lists.newArrayListWithCapacity(v.subTriggers().size());
-      for (Trigger subtrigger : v.subTriggers()) {
-        subStateMachines.add(stateMachineForOnceTrigger((OnceTrigger) subtrigger));
-      }
-      return AfterAllStateMachine.of(subStateMachines);
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(AfterPane v) {
-      return AfterPaneStateMachine.elementCountAtLeast(v.getElementCount());
-    }
-
-    private TriggerStateMachine evaluateSpecific(AfterWatermark.AfterWatermarkEarlyAndLate
v) {
+    } else {
       AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate machine =
           AfterWatermarkStateMachine.pastEndOfWindow()
-              .withEarlyFirings(stateMachineForOnceTrigger(v.getEarlyTrigger()));
+              .withEarlyFirings(stateMachineForTrigger(trigger.getEarlyFirings()));
 
-      if (v.getLateTrigger() != null) {
-        machine = machine.withLateFirings(stateMachineForOnceTrigger(v.getLateTrigger()));
+      if (trigger.hasLateFirings()) {
+        machine = machine.withLateFirings(stateMachineForTrigger(trigger.getLateFirings()));
       }
       return machine;
     }
+  }
 
-    private TriggerStateMachine evaluateSpecific(AfterEach v) {
-      List<TriggerStateMachine> subStateMachines =
-          Lists.newArrayListWithCapacity(v.subTriggers().size());
-
-      for (Trigger subtrigger : v.subTriggers()) {
-        subStateMachines.add(stateMachineForTrigger(subtrigger));
+  private static TriggerStateMachine stateMachineForAfterProcessingTime(
+      RunnerApi.Trigger.AfterProcessingTime trigger) {
+    AfterDelayFromFirstElementStateMachine stateMachine =
+        AfterProcessingTimeStateMachine.pastFirstElementInPane();
+    for (RunnerApi.TimestampTransform transform : trigger.getTimestampTransformsList()) {
+      switch (transform.getTimestampTransformCase()) {
+        case ALIGN_TO:
+          stateMachine =
+              stateMachine.alignedTo(
+                  Duration.millis(transform.getAlignTo().getPeriod()),
+                  new Instant(transform.getAlignTo().getOffset()));
+          break;
+        case DELAY:
+          stateMachine =
+              stateMachine.plusDelayOf(Duration.millis(transform.getDelay().getDelayMillis()));
+          break;
+        case TIMESTAMPTRANSFORM_NOT_SET:
+          throw new IllegalArgumentException(
+              String.format("Required field 'timestamp_transform' not set in %s", transform));
+        default:
+          throw new IllegalArgumentException(
+              String.format(
+                  "Unknown timestamp transform case: %s", transform.getTimestampTransformCase()));
       }
-
-      return AfterEachStateMachine.inOrder(subStateMachines);
-    }
-
-    private TriggerStateMachine evaluateSpecific(Repeatedly v) {
-      return RepeatedlyStateMachine.forever(stateMachineForTrigger(v.getRepeatedTrigger()));
-    }
-
-    private TriggerStateMachine evaluateSpecific(OrFinallyTrigger v) {
-      return new OrFinallyStateMachine(
-          stateMachineForTrigger(v.getMainTrigger()),
-          stateMachineForOnceTrigger(v.getUntilTrigger()));
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(AfterProcessingTime v) {
-      return new AfterDelayFromFirstElementStateMachineAdapter(v);
     }
+    return stateMachine;
+  }
 
-    private static class AfterDelayFromFirstElementStateMachineAdapter
-        extends AfterDelayFromFirstElementStateMachine {
-
-      private static final Function<TimestampTransform, SerializableFunction<Instant,
Instant>>
-          CONVERT_TIMESTAMP_TRANSFORM =
-              new Function<TimestampTransform, SerializableFunction<Instant, Instant>>()
{
-                @Override
-                public SerializableFunction<Instant, Instant> apply(
-                    @Nonnull TimestampTransform transform) {
-                  if (transform instanceof TimestampTransform.Delay) {
-                    return new DelayFn(((TimestampTransform.Delay) transform).getDelay());
-                  } else if (transform instanceof TimestampTransform.AlignTo) {
-                    TimestampTransform.AlignTo alignTo = (TimestampTransform.AlignTo) transform;
-                    return new AlignFn(alignTo.getPeriod(), alignTo.getOffset());
-                  } else {
-                    throw new IllegalArgumentException(
-                        String.format(
-                            "Unknown %s: %s", TimestampTransform.class.getSimpleName(), transform));
-                  }
-                }
-              };
-
-      public AfterDelayFromFirstElementStateMachineAdapter(AfterProcessingTime v) {
-        this(
-            TimeDomain.PROCESSING_TIME,
-            FluentIterable.from(v.getTimestampTransforms())
-                .transform(CONVERT_TIMESTAMP_TRANSFORM)
-                .toList());
-      }
-
-      private AfterDelayFromFirstElementStateMachineAdapter(
-          TimeDomain timeDomain, List<SerializableFunction<Instant, Instant>>
timestampMappers) {
-        super(timeDomain, timestampMappers);
-      }
-
-      @Override
-      public Instant getCurrentTime(TriggerContext context) {
-        switch (timeDomain) {
-          case PROCESSING_TIME:
-            return context.currentProcessingTime();
-          case SYNCHRONIZED_PROCESSING_TIME:
-            return context.currentSynchronizedProcessingTime();
-          case EVENT_TIME:
-            return context.currentEventTime();
-          default:
-            throw new IllegalArgumentException("A time domain that doesn't exist was received!");
-        }
-      }
-
-      @Override
-      protected AfterDelayFromFirstElementStateMachine newWith(
-          List<SerializableFunction<Instant, Instant>> transform) {
-        return new AfterDelayFromFirstElementStateMachineAdapter(timeDomain, transform);
-      }
+  private static List<TriggerStateMachine> stateMachinesForTriggers(
+      List<RunnerApi.Trigger> triggers) {
+    List<TriggerStateMachine> stateMachines = new ArrayList<>(triggers.size());
+    for (RunnerApi.Trigger trigger : triggers) {
+      stateMachines.add(stateMachineForTrigger(trigger));
     }
+    return stateMachines;
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index dab2bf9..d18a1c3 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -60,6 +60,7 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
 import org.apache.beam.sdk.transforms.windowing.OutputTimeFns;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.transforms.windowing.Trigger;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.transforms.windowing.Window.ClosingBehavior;
 import org.apache.beam.sdk.transforms.windowing.WindowFn;
 import org.apache.beam.sdk.util.AppliedCombineFn;
@@ -122,7 +123,8 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow>
{
       nonCombining(WindowingStrategy<?, W> windowingStrategy) throws Exception {
     return new ReduceFnTester<Integer, Iterable<Integer>, W>(
         windowingStrategy,
-        TriggerStateMachines.stateMachineForTrigger(windowingStrategy.getTrigger()),
+        TriggerStateMachines.stateMachineForTrigger(
+            Triggers.toProto(windowingStrategy.getTrigger())),
         SystemReduceFn.<String, Integer, W>buffering(VarIntCoder.of()),
         IterableCoder.of(VarIntCoder.of()),
         PipelineOptionsFactory.create(),
@@ -186,7 +188,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow>
{
 
     return combining(
         strategy,
-        TriggerStateMachines.stateMachineForTrigger(strategy.getTrigger()),
+        TriggerStateMachines.stateMachineForTrigger(Triggers.toProto(strategy.getTrigger())),
         combineFn,
         outputCoder);
   }
@@ -236,7 +238,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow>
{
 
     return combining(
         strategy,
-        TriggerStateMachines.stateMachineForTrigger(strategy.getTrigger()),
+        TriggerStateMachines.stateMachineForTrigger(Triggers.toProto(strategy.getTrigger())),
         combineFn,
         outputCoder,
         options,

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
index 26c0597..497a3c2 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
@@ -19,21 +19,10 @@ package org.apache.beam.runners.core.triggers;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertThat;
 
-import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
-import org.apache.beam.sdk.transforms.windowing.AfterAll;
-import org.apache.beam.sdk.transforms.windowing.AfterEach;
-import org.apache.beam.sdk.transforms.windowing.AfterFirst;
-import org.apache.beam.sdk.transforms.windowing.AfterPane;
-import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
-import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
-import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
-import org.apache.beam.sdk.transforms.windowing.Never;
-import org.apache.beam.sdk.transforms.windowing.Never.NeverTrigger;
-import org.apache.beam.sdk.transforms.windowing.OrFinallyTrigger;
-import org.apache.beam.sdk.transforms.windowing.Repeatedly;
-import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi;
 import org.apache.beam.sdk.util.TimeDomain;
 import org.joda.time.Duration;
 import org.junit.Test;
@@ -51,49 +40,79 @@ public class TriggerStateMachinesTest {
   @Test
   public void testStateMachineForAfterPane() {
     int count = 37;
-    AfterPane trigger = AfterPane.elementCountAtLeast(count);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setElementCount(RunnerApi.Trigger.ElementCount.newBuilder().setElementCount(count))
+            .build();
+
     AfterPaneStateMachine machine =
-        (AfterPaneStateMachine) TriggerStateMachines.stateMachineForOnceTrigger(trigger);
+        (AfterPaneStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
 
-    assertThat(machine.getElementCount(), equalTo(trigger.getElementCount()));
+    assertThat(machine.getElementCount(), equalTo(trigger.getElementCount().getElementCount()));
   }
 
+  // TODO: make these all build the proto
   @Test
   public void testStateMachineForAfterProcessingTime() {
     Duration minutes = Duration.standardMinutes(94);
     Duration hours = Duration.standardHours(13);
 
-    AfterProcessingTime trigger =
-        AfterProcessingTime.pastFirstElementInPane().plusDelayOf(minutes).alignedTo(hours);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterProcessingTime(
+                RunnerApi.Trigger.AfterProcessingTime.newBuilder()
+                    .addTimestampTransforms(
+                        RunnerApi.TimestampTransform.newBuilder()
+                            .setDelay(
+                                RunnerApi.TimestampTransform.Delay.newBuilder()
+                                    .setDelayMillis(minutes.getMillis())))
+                    .addTimestampTransforms(
+                        RunnerApi.TimestampTransform.newBuilder()
+                            .setAlignTo(
+                                RunnerApi.TimestampTransform.AlignTo.newBuilder()
+                                    .setPeriod(hours.getMillis()))))
+            .build();
 
     AfterDelayFromFirstElementStateMachine machine =
         (AfterDelayFromFirstElementStateMachine)
-            TriggerStateMachines.stateMachineForOnceTrigger(trigger);
+            TriggerStateMachines.stateMachineForTrigger(trigger);
 
     assertThat(machine.getTimeDomain(), equalTo(TimeDomain.PROCESSING_TIME));
   }
 
   @Test
   public void testStateMachineForAfterWatermark() {
-    AfterWatermark.FromEndOfWindow trigger = AfterWatermark.pastEndOfWindow();
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterEndOfWidow(RunnerApi.Trigger.AfterEndOfWindow.getDefaultInstance())
+            .build();
     AfterWatermarkStateMachine.FromEndOfWindow machine =
         (AfterWatermarkStateMachine.FromEndOfWindow)
-            TriggerStateMachines.stateMachineForOnceTrigger(trigger);
-    // No parameters, so if it doesn't crash, we win!
+            TriggerStateMachines.stateMachineForTrigger(trigger);
+
+    assertThat(
+        TriggerStateMachines.stateMachineForTrigger(trigger),
+        instanceOf(AfterWatermarkStateMachine.FromEndOfWindow.class));
   }
 
   @Test
   public void testDefaultTriggerTranslation() {
-    DefaultTrigger trigger = DefaultTrigger.of();
-    DefaultTriggerStateMachine machine =
-        (DefaultTriggerStateMachine)
-            checkNotNull(TriggerStateMachines.stateMachineForTrigger(trigger));
-    // No parameters, so if it doesn't crash, we win!
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setDefault(RunnerApi.Trigger.Default.getDefaultInstance())
+            .build();
+
+    assertThat(
+        TriggerStateMachines.stateMachineForTrigger(trigger),
+        instanceOf(DefaultTriggerStateMachine.class));
   }
 
   @Test
   public void testNeverTranslation() {
-    NeverTrigger trigger = Never.ever();
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setNever(RunnerApi.Trigger.Never.getDefaultInstance())
+            .build();
     NeverStateMachine machine =
         (NeverStateMachine) checkNotNull(TriggerStateMachines.stateMachineForTrigger(trigger));
     // No parameters, so if it doesn't crash, we win!
@@ -109,18 +128,35 @@ public class TriggerStateMachinesTest {
   private static final int ELEM_COUNT = 472;
   private static final Duration DELAY = Duration.standardSeconds(95673);
 
-  private final OnceTrigger subtrigger1 = AfterPane.elementCountAtLeast(ELEM_COUNT);
-  private final OnceTrigger subtrigger2 =
-      AfterProcessingTime.pastFirstElementInPane().plusDelayOf(DELAY);
+  private final RunnerApi.Trigger subtrigger1 =
+      RunnerApi.Trigger.newBuilder()
+          .setElementCount(RunnerApi.Trigger.ElementCount.newBuilder().setElementCount(ELEM_COUNT))
+          .build();
+  private final RunnerApi.Trigger subtrigger2 =
+      RunnerApi.Trigger.newBuilder()
+          .setAfterProcessingTime(
+              RunnerApi.Trigger.AfterProcessingTime.newBuilder()
+                  .addTimestampTransforms(
+                      RunnerApi.TimestampTransform.newBuilder()
+                          .setDelay(
+                              RunnerApi.TimestampTransform.Delay.newBuilder()
+                                  .setDelayMillis(DELAY.getMillis()))))
+          .build();
 
-  private final OnceTriggerStateMachine submachine1 =
-      TriggerStateMachines.stateMachineForOnceTrigger(subtrigger1);
-  private final OnceTriggerStateMachine submachine2 =
-      TriggerStateMachines.stateMachineForOnceTrigger(subtrigger2);
+  private final TriggerStateMachine submachine1 =
+      TriggerStateMachines.stateMachineForTrigger(subtrigger1);
+  private final TriggerStateMachine submachine2 =
+      TriggerStateMachines.stateMachineForTrigger(subtrigger2);
 
   @Test
   public void testAfterEachTranslation() {
-    AfterEach trigger = AfterEach.inOrder(subtrigger1, subtrigger2);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterEach(
+                RunnerApi.Trigger.AfterEach.newBuilder()
+                    .addSubtriggers(subtrigger1)
+                    .addSubtriggers(subtrigger2))
+            .build();
     AfterEachStateMachine machine =
         (AfterEachStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
 
@@ -129,7 +165,13 @@ public class TriggerStateMachinesTest {
 
   @Test
   public void testAfterFirstTranslation() {
-    AfterFirst trigger = AfterFirst.of(subtrigger1, subtrigger2);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterAny(
+                RunnerApi.Trigger.AfterAny.newBuilder()
+                    .addSubtriggers(subtrigger1)
+                    .addSubtriggers(subtrigger2))
+            .build();
     AfterFirstStateMachine machine =
         (AfterFirstStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
 
@@ -138,7 +180,13 @@ public class TriggerStateMachinesTest {
 
   @Test
   public void testAfterAllTranslation() {
-    AfterAll trigger = AfterAll.of(subtrigger1, subtrigger2);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterAll(
+                RunnerApi.Trigger.AfterAll.newBuilder()
+                    .addSubtriggers(subtrigger1)
+                    .addSubtriggers(subtrigger2))
+            .build();
     AfterAllStateMachine machine =
         (AfterAllStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
 
@@ -147,8 +195,11 @@ public class TriggerStateMachinesTest {
 
   @Test
   public void testAfterWatermarkEarlyTranslation() {
-    AfterWatermark.AfterWatermarkEarlyAndLate trigger =
-        AfterWatermark.pastEndOfWindow().withEarlyFirings(subtrigger1);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterEndOfWidow(
+                RunnerApi.Trigger.AfterEndOfWindow.newBuilder().setEarlyFirings(subtrigger1))
+            .build();
     AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate machine =
         (AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate)
             TriggerStateMachines.stateMachineForTrigger(trigger);
@@ -160,8 +211,13 @@ public class TriggerStateMachinesTest {
 
   @Test
   public void testAfterWatermarkEarlyLateTranslation() {
-    AfterWatermark.AfterWatermarkEarlyAndLate trigger =
-        AfterWatermark.pastEndOfWindow().withEarlyFirings(subtrigger1).withLateFirings(subtrigger2);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setAfterEndOfWidow(
+                RunnerApi.Trigger.AfterEndOfWindow.newBuilder()
+                    .setEarlyFirings(subtrigger1)
+                    .setLateFirings(subtrigger2))
+            .build();
     AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate machine =
         (AfterWatermarkStateMachine.AfterWatermarkEarlyAndLate)
             TriggerStateMachines.stateMachineForTrigger(trigger);
@@ -176,18 +232,30 @@ public class TriggerStateMachinesTest {
 
   @Test
   public void testOrFinallyTranslation() {
-    OrFinallyTrigger trigger = subtrigger1.orFinally(subtrigger2);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setOrFinally(
+                RunnerApi.Trigger.OrFinally.newBuilder()
+                    .setMain(subtrigger1)
+                    .setFinally(subtrigger2))
+            .build();
     OrFinallyStateMachine machine =
-        (OrFinallyStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
+        (OrFinallyStateMachine)
+            TriggerStateMachines.stateMachineForTrigger(trigger);
 
     assertThat(machine, equalTo(submachine1.orFinally(submachine2)));
   }
 
   @Test
   public void testRepeatedlyTranslation() {
-    Repeatedly trigger = Repeatedly.forever(subtrigger1);
+    RunnerApi.Trigger trigger =
+        RunnerApi.Trigger.newBuilder()
+            .setRepeat(
+                RunnerApi.Trigger.Repeat.newBuilder()
+                    .setSubtrigger(subtrigger1)).build();
     RepeatedlyStateMachine machine =
-        (RepeatedlyStateMachine) TriggerStateMachines.stateMachineForTrigger(trigger);
+        (RepeatedlyStateMachine)
+            TriggerStateMachines.stateMachineForTrigger(trigger);
 
     assertThat(machine, equalTo(RepeatedlyStateMachine.forever(submachine1)));
   }

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
index ecf4ecd..dc0ac60 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
@@ -46,6 +46,7 @@ import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.util.SideInputReader;
 import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
@@ -172,7 +173,8 @@ class GroupAlsoByWindowEvaluatorFactory implements TransformEvaluatorFactory
{
               key,
               windowingStrategy,
               ExecutableTriggerStateMachine.create(
-                  TriggerStateMachines.stateMachineForTrigger(windowingStrategy.getTrigger())),
+                  TriggerStateMachines.stateMachineForTrigger(
+                      Triggers.toProto(windowingStrategy.getTrigger()))),
               stateInternals,
               timerInternals,
               new OutputWindowedValueToBundle<>(bundle),

http://git-wip-us.apache.org/repos/asf/beam/blob/27a482ba/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowFn.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowFn.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowFn.java
index 44d5b7c..bd37fdb 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowFn.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkGroupAlsoByWindowFn.java
@@ -37,6 +37,7 @@ import org.apache.beam.sdk.transforms.Aggregator;
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.Triggers;
 import org.apache.beam.sdk.util.SideInputReader;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
@@ -101,29 +102,30 @@ public class SparkGroupAlsoByWindowFn<K, InputT, W extends BoundedWindow>
             key,
             windowingStrategy,
             ExecutableTriggerStateMachine.create(
-                TriggerStateMachines.stateMachineForTrigger(windowingStrategy.getTrigger())),
+                TriggerStateMachines.stateMachineForTrigger(
+                    Triggers.toProto(windowingStrategy.getTrigger()))),
             stateInternals,
             timerInternals,
             outputter,
             new SideInputReader() {
-                @Override
-                public <T> T get(PCollectionView<T> view, BoundedWindow sideInputWindow)
{
-                  throw new UnsupportedOperationException(
-                      "GroupAlsoByWindow must not have side inputs");
-                }
-
-                @Override
-                public <T> boolean contains(PCollectionView<T> view) {
-                  throw new UnsupportedOperationException(
-                      "GroupAlsoByWindow must not have side inputs");
-                }
-
-                @Override
-                public boolean isEmpty() {
-                  throw new UnsupportedOperationException(
-                      "GroupAlsoByWindow must not have side inputs");
-                }
-              },
+              @Override
+              public <T> T get(PCollectionView<T> view, BoundedWindow sideInputWindow)
{
+                throw new UnsupportedOperationException(
+                    "GroupAlsoByWindow must not have side inputs");
+              }
+
+              @Override
+              public <T> boolean contains(PCollectionView<T> view) {
+                throw new UnsupportedOperationException(
+                    "GroupAlsoByWindow must not have side inputs");
+              }
+
+              @Override
+              public boolean isEmpty() {
+                throw new UnsupportedOperationException(
+                    "GroupAlsoByWindow must not have side inputs");
+              }
+            },
             droppedDueToClosedWindow,
             reduceFn,
             runtimeContext.getPipelineOptions());


Mime
View raw message