You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/02/24 15:28:52 UTC
[4/5] beam git commit: Build trigger state machine from Runner API
Trigger proto directly
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 <kl...@google.com>
Authored: Fri Feb 17 16:05:13 2017 -0800
Committer: Kenneth Knowles <kl...@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());