You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2016/04/12 18:32:13 UTC
[1/3] incubator-beam git commit: Remove extraneous type parameter
from Trigger, etc
Repository: incubator-beam
Updated Branches:
refs/heads/master af1e57555 -> c558a64c3
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerContextFactory.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerContextFactory.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerContextFactory.java
index 2408e9e..5e2dc5e 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerContextFactory.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerContextFactory.java
@@ -65,40 +65,40 @@ public class TriggerContextFactory<W extends BoundedWindow> {
this.windowCoder = windowingStrategy.getWindowFn().windowCoder();
}
- public Trigger<W>.TriggerContext base(W window, Timers timers,
- ExecutableTrigger<W> rootTrigger, FinishedTriggers finishedSet) {
+ public Trigger.TriggerContext base(W window, Timers timers,
+ ExecutableTrigger rootTrigger, FinishedTriggers finishedSet) {
return new TriggerContextImpl(window, timers, rootTrigger, finishedSet);
}
- public Trigger<W>.OnElementContext createOnElementContext(
+ public Trigger.OnElementContext createOnElementContext(
W window, Timers timers, Instant elementTimestamp,
- ExecutableTrigger<W> rootTrigger, FinishedTriggers finishedSet) {
+ ExecutableTrigger rootTrigger, FinishedTriggers finishedSet) {
return new OnElementContextImpl(window, timers, rootTrigger, finishedSet, elementTimestamp);
}
- public Trigger<W>.OnMergeContext createOnMergeContext(W window, Timers timers,
- ExecutableTrigger<W> rootTrigger, FinishedTriggers finishedSet,
+ public Trigger.OnMergeContext createOnMergeContext(W window, Timers timers,
+ ExecutableTrigger rootTrigger, FinishedTriggers finishedSet,
Map<W, FinishedTriggers> finishedSets) {
return new OnMergeContextImpl(window, timers, rootTrigger, finishedSet, finishedSets);
}
- public StateAccessor<?> createStateAccessor(W window, ExecutableTrigger<W> trigger) {
+ public StateAccessor<?> createStateAccessor(W window, ExecutableTrigger trigger) {
return new StateAccessorImpl(window, trigger);
}
public MergingStateAccessor<?, W> createMergingStateAccessor(
- W mergeResult, Collection<W> mergingWindows, ExecutableTrigger<W> trigger) {
+ W mergeResult, Collection<W> mergingWindows, ExecutableTrigger trigger) {
return new MergingStateAccessorImpl(trigger, mergingWindows, mergeResult);
}
- private class TriggerInfoImpl implements Trigger.TriggerInfo<W> {
+ private class TriggerInfoImpl implements Trigger.TriggerInfo {
- protected final ExecutableTrigger<W> trigger;
+ protected final ExecutableTrigger trigger;
protected final FinishedTriggers finishedSet;
- private final Trigger<W>.TriggerContext context;
+ private final Trigger.TriggerContext context;
- public TriggerInfoImpl(ExecutableTrigger<W> trigger, FinishedTriggers finishedSet,
- Trigger<W>.TriggerContext context) {
+ public TriggerInfoImpl(ExecutableTrigger trigger, FinishedTriggers finishedSet,
+ Trigger.TriggerContext context) {
this.trigger = trigger;
this.finishedSet = finishedSet;
this.context = context;
@@ -110,12 +110,12 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Iterable<ExecutableTrigger<W>> subTriggers() {
+ public Iterable<ExecutableTrigger> subTriggers() {
return trigger.subTriggers();
}
@Override
- public ExecutableTrigger<W> subTrigger(int subtriggerIndex) {
+ public ExecutableTrigger subTrigger(int subtriggerIndex) {
return trigger.subTriggers().get(subtriggerIndex);
}
@@ -135,20 +135,20 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Iterable<ExecutableTrigger<W>> unfinishedSubTriggers() {
+ public Iterable<ExecutableTrigger> unfinishedSubTriggers() {
return FluentIterable
.from(trigger.subTriggers())
- .filter(new Predicate<ExecutableTrigger<W>>() {
+ .filter(new Predicate<ExecutableTrigger>() {
@Override
- public boolean apply(ExecutableTrigger<W> trigger) {
+ public boolean apply(ExecutableTrigger trigger) {
return !finishedSet.isFinished(trigger);
}
});
}
@Override
- public ExecutableTrigger<W> firstUnfinishedSubTrigger() {
- for (ExecutableTrigger<W> subTrigger : trigger.subTriggers()) {
+ public ExecutableTrigger firstUnfinishedSubTrigger() {
+ for (ExecutableTrigger subTrigger : trigger.subTriggers()) {
if (!finishedSet.isFinished(subTrigger)) {
return subTrigger;
}
@@ -217,14 +217,14 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
private class MergingTriggerInfoImpl
- extends TriggerInfoImpl implements Trigger.MergingTriggerInfo<W> {
+ extends TriggerInfoImpl implements Trigger.MergingTriggerInfo {
private final Map<W, FinishedTriggers> finishedSets;
public MergingTriggerInfoImpl(
- ExecutableTrigger<W> trigger,
+ ExecutableTrigger trigger,
FinishedTriggers finishedSet,
- Trigger<W>.TriggerContext context,
+ Trigger.TriggerContext context,
Map<W, FinishedTriggers> finishedSets) {
super(trigger, finishedSet, context);
this.finishedSets = finishedSets;
@@ -251,7 +251,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Iterable<W> getFinishedMergingWindows() {
+ public Iterable<? extends BoundedWindow> getFinishedMergingWindows() {
return Maps.filterValues(finishedSets, new Predicate<FinishedTriggers>() {
@Override
public boolean apply(FinishedTriggers finishedSet) {
@@ -267,7 +267,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
public StateAccessorImpl(
W window,
- ExecutableTrigger<W> trigger) {
+ ExecutableTrigger trigger) {
this.triggerIndex = trigger.getTriggerIndex();
this.windowNamespace = namespaceFor(window);
}
@@ -286,7 +286,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
implements MergingStateAccessor<Object, W> {
private final Collection<W> activeToBeMerged;
- public MergingStateAccessorImpl(ExecutableTrigger<W> trigger, Collection<W> activeToBeMerged,
+ public MergingStateAccessorImpl(ExecutableTrigger trigger, Collection<W> activeToBeMerged,
W mergeResult) {
super(mergeResult, trigger);
this.activeToBeMerged = activeToBeMerged;
@@ -310,7 +310,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
}
- private class TriggerContextImpl extends Trigger<W>.TriggerContext {
+ private class TriggerContextImpl extends Trigger.TriggerContext {
private final W window;
private final StateAccessorImpl state;
@@ -320,7 +320,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
private TriggerContextImpl(
W window,
Timers timers,
- ExecutableTrigger<W> trigger,
+ ExecutableTrigger trigger,
FinishedTriggers finishedSet) {
trigger.getSpec().super();
this.window = window;
@@ -330,17 +330,17 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Trigger<W>.TriggerContext forTrigger(ExecutableTrigger<W> trigger) {
+ public Trigger.TriggerContext forTrigger(ExecutableTrigger trigger) {
return new TriggerContextImpl(window, timers, trigger, triggerInfo.finishedSet);
}
@Override
- public TriggerInfo<W> trigger() {
+ public TriggerInfo trigger() {
return triggerInfo;
}
@Override
- public StateAccessor state() {
+ public StateAccessor<?> state() {
return state;
}
@@ -372,7 +372,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
}
- private class OnElementContextImpl extends Trigger<W>.OnElementContext {
+ private class OnElementContextImpl extends Trigger.OnElementContext {
private final W window;
private final StateAccessorImpl state;
@@ -383,7 +383,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
private OnElementContextImpl(
W window,
Timers timers,
- ExecutableTrigger<W> trigger,
+ ExecutableTrigger trigger,
FinishedTriggers finishedSet,
Instant eventTimestamp) {
trigger.getSpec().super();
@@ -401,18 +401,18 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Trigger<W>.OnElementContext forTrigger(ExecutableTrigger<W> trigger) {
+ public Trigger.OnElementContext forTrigger(ExecutableTrigger trigger) {
return new OnElementContextImpl(
window, timers, trigger, triggerInfo.finishedSet, eventTimestamp);
}
@Override
- public TriggerInfo<W> trigger() {
+ public TriggerInfo trigger() {
return triggerInfo;
}
@Override
- public StateAccessor state() {
+ public StateAccessor<?> state() {
return state;
}
@@ -450,7 +450,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
}
- private class OnMergeContextImpl extends Trigger<W>.OnMergeContext {
+ private class OnMergeContextImpl extends Trigger.OnMergeContext {
private final MergingStateAccessor<?, W> state;
private final W window;
private final Collection<W> mergingWindows;
@@ -460,7 +460,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
private OnMergeContextImpl(
W window,
Timers timers,
- ExecutableTrigger<W> trigger,
+ ExecutableTrigger trigger,
FinishedTriggers finishedSet,
Map<W, FinishedTriggers> finishedSets) {
trigger.getSpec().super();
@@ -472,7 +472,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public Trigger<W>.OnMergeContext forTrigger(ExecutableTrigger<W> trigger) {
+ public Trigger.OnMergeContext forTrigger(ExecutableTrigger trigger) {
return new OnMergeContextImpl(
window, timers, trigger, triggerInfo.finishedSet, triggerInfo.finishedSets);
}
@@ -483,7 +483,7 @@ public class TriggerContextFactory<W extends BoundedWindow> {
}
@Override
- public MergingTriggerInfo<W> trigger() {
+ public MergingTriggerInfo trigger() {
return triggerInfo;
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerRunner.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerRunner.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerRunner.java
index aa7e101..24e92f8 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerRunner.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/TriggerRunner.java
@@ -61,10 +61,10 @@ public class TriggerRunner<W extends BoundedWindow> {
static final StateTag<Object, ValueState<BitSet>> FINISHED_BITS_TAG =
StateTags.makeSystemTagInternal(StateTags.value("closed", BitSetCoder.of()));
- private final ExecutableTrigger<W> rootTrigger;
+ private final ExecutableTrigger rootTrigger;
private final TriggerContextFactory<W> contextFactory;
- public TriggerRunner(ExecutableTrigger<W> rootTrigger, TriggerContextFactory<W> contextFactory) {
+ public TriggerRunner(ExecutableTrigger rootTrigger, TriggerContextFactory<W> contextFactory) {
Preconditions.checkState(rootTrigger.getTriggerIndex() == 0);
this.rootTrigger = rootTrigger;
this.contextFactory = contextFactory;
@@ -129,7 +129,7 @@ public class TriggerRunner<W extends BoundedWindow> {
// Clone so that we can detect changes and so that changes here don't pollute merging.
FinishedTriggersBitSet finishedSet =
readFinishedBits(state.access(FINISHED_BITS_TAG)).copy();
- Trigger<W>.OnElementContext triggerContext = contextFactory.createOnElementContext(
+ Trigger.OnElementContext triggerContext = contextFactory.createOnElementContext(
window, timers, timestamp, rootTrigger, finishedSet);
rootTrigger.invokeOnElement(triggerContext);
persistFinishedSet(state, finishedSet);
@@ -165,7 +165,7 @@ public class TriggerRunner<W extends BoundedWindow> {
}
ImmutableMap<W, FinishedTriggers> mergingFinishedSets = builder.build();
- Trigger<W>.OnMergeContext mergeContext = contextFactory.createOnMergeContext(
+ Trigger.OnMergeContext mergeContext = contextFactory.createOnMergeContext(
window, timers, rootTrigger, finishedSet, mergingFinishedSets);
// Run the merge from the trigger
@@ -176,7 +176,7 @@ public class TriggerRunner<W extends BoundedWindow> {
public boolean shouldFire(W window, Timers timers, StateAccessor<?> state) throws Exception {
FinishedTriggers finishedSet = readFinishedBits(state.access(FINISHED_BITS_TAG)).copy();
- Trigger<W>.TriggerContext context = contextFactory.base(window, timers,
+ Trigger.TriggerContext context = contextFactory.base(window, timers,
rootTrigger, finishedSet);
return rootTrigger.invokeShouldFire(context);
}
@@ -186,7 +186,7 @@ public class TriggerRunner<W extends BoundedWindow> {
// However it is too expensive to assert.
FinishedTriggersBitSet finishedSet =
readFinishedBits(state.access(FINISHED_BITS_TAG)).copy();
- Trigger<W>.TriggerContext context = contextFactory.base(window, timers,
+ Trigger.TriggerContext context = contextFactory.base(window, timers,
rootTrigger, finishedSet);
rootTrigger.invokeOnFire(context);
persistFinishedSet(state, finishedSet);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
index 8b78ce4..6068656 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/WindowingStrategy.java
@@ -57,7 +57,7 @@ public class WindowingStrategy<T, W extends BoundedWindow> implements Serializab
private final WindowFn<T, W> windowFn;
private final OutputTimeFn<? super W> outputTimeFn;
- private final ExecutableTrigger<W> trigger;
+ private final ExecutableTrigger trigger;
private final AccumulationMode mode;
private final Duration allowedLateness;
private final ClosingBehavior closingBehavior;
@@ -68,7 +68,7 @@ public class WindowingStrategy<T, W extends BoundedWindow> implements Serializab
private WindowingStrategy(
WindowFn<T, W> windowFn,
- ExecutableTrigger<W> trigger, boolean triggerSpecified,
+ ExecutableTrigger trigger, boolean triggerSpecified,
AccumulationMode mode, boolean modeSpecified,
Duration allowedLateness, boolean allowedLatenessSpecified,
OutputTimeFn<? super W> outputTimeFn, boolean outputTimeFnSpecified,
@@ -106,7 +106,7 @@ public class WindowingStrategy<T, W extends BoundedWindow> implements Serializab
return windowFn;
}
- public ExecutableTrigger<W> getTrigger() {
+ public ExecutableTrigger getTrigger() {
return trigger;
}
@@ -146,12 +146,10 @@ public class WindowingStrategy<T, W extends BoundedWindow> implements Serializab
* Returns a {@link WindowingStrategy} identical to {@code this} but with the trigger set to
* {@code wildcardTrigger}.
*/
- public WindowingStrategy<T, W> withTrigger(Trigger<?> wildcardTrigger) {
- @SuppressWarnings("unchecked")
- Trigger<W> typedTrigger = (Trigger<W>) wildcardTrigger;
+ public WindowingStrategy<T, W> withTrigger(Trigger trigger) {
return new WindowingStrategy<T, W>(
windowFn,
- ExecutableTrigger.create(typedTrigger), true,
+ ExecutableTrigger.create(trigger), true,
mode, modeSpecified,
allowedLateness, allowedLatenessSpecified,
outputTimeFn, outputTimeFnSpecified,
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAllTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAllTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAllTest.java
index 5876d94..0499448 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAllTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAllTest.java
@@ -141,9 +141,9 @@ public class AfterAllTest {
@Test
public void testContinuation() throws Exception {
- OnceTrigger<IntervalWindow> trigger1 = AfterProcessingTime.pastFirstElementInPane();
- OnceTrigger<IntervalWindow> trigger2 = AfterWatermark.pastEndOfWindow();
- Trigger<IntervalWindow> afterAll = AfterAll.of(trigger1, trigger2);
+ OnceTrigger trigger1 = AfterProcessingTime.pastFirstElementInPane();
+ OnceTrigger trigger2 = AfterWatermark.pastEndOfWindow();
+ Trigger afterAll = AfterAll.of(trigger1, trigger2);
assertEquals(
AfterAll.of(trigger1.getContinuationTrigger(), trigger2.getContinuationTrigger()),
afterAll.getContinuationTrigger());
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEachTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEachTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEachTest.java
index 185ae0d..5dde528 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEachTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEachTest.java
@@ -112,9 +112,9 @@ public class AfterEachTest {
@Test
public void testContinuation() throws Exception {
- OnceTrigger<IntervalWindow> trigger1 = AfterProcessingTime.pastFirstElementInPane();
- OnceTrigger<IntervalWindow> trigger2 = AfterWatermark.pastEndOfWindow();
- Trigger<IntervalWindow> afterEach = AfterEach.inOrder(trigger1, trigger2);
+ OnceTrigger trigger1 = AfterProcessingTime.pastFirstElementInPane();
+ OnceTrigger trigger2 = AfterWatermark.pastEndOfWindow();
+ Trigger afterEach = AfterEach.inOrder(trigger1, trigger2);
assertEquals(
Repeatedly.forever(AfterFirst.of(
trigger1.getContinuationTrigger(), trigger2.getContinuationTrigger())),
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirstTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirstTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirstTest.java
index bc6a2b8..2483107 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirstTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirstTest.java
@@ -42,11 +42,11 @@ import org.mockito.MockitoAnnotations;
@RunWith(JUnit4.class)
public class AfterFirstTest {
- @Mock private OnceTrigger<IntervalWindow> mockTrigger1;
- @Mock private OnceTrigger<IntervalWindow> mockTrigger2;
+ @Mock private OnceTrigger mockTrigger1;
+ @Mock private OnceTrigger mockTrigger2;
private SimpleTriggerTester<IntervalWindow> tester;
- private static Trigger<IntervalWindow>.TriggerContext anyTriggerContext() {
- return Mockito.<Trigger<IntervalWindow>.TriggerContext>any();
+ private static Trigger.TriggerContext anyTriggerContext() {
+ return Mockito.<Trigger.TriggerContext>any();
}
@Before
@@ -166,9 +166,9 @@ public class AfterFirstTest {
@Test
public void testContinuation() throws Exception {
- OnceTrigger<IntervalWindow> trigger1 = AfterProcessingTime.pastFirstElementInPane();
- OnceTrigger<IntervalWindow> trigger2 = AfterWatermark.pastEndOfWindow();
- Trigger<IntervalWindow> afterFirst = AfterFirst.of(trigger1, trigger2);
+ OnceTrigger trigger1 = AfterProcessingTime.pastFirstElementInPane();
+ OnceTrigger trigger2 = AfterWatermark.pastEndOfWindow();
+ Trigger afterFirst = AfterFirst.of(trigger1, trigger2);
assertEquals(
AfterFirst.of(trigger1.getContinuationTrigger(), trigger2.getContinuationTrigger()),
afterFirst.getContinuationTrigger());
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTimeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTimeTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTimeTest.java
index 75677b5..04f37a1 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTimeTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTimeTest.java
@@ -137,10 +137,10 @@ public class AfterProcessingTimeTest {
@Test
public void testContinuation() throws Exception {
- OnceTrigger<?> firstElementPlus1 =
+ OnceTrigger firstElementPlus1 =
AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardHours(1));
assertEquals(
- new AfterSynchronizedProcessingTime<>(),
+ new AfterSynchronizedProcessingTime(),
firstElementPlus1.getContinuationTrigger());
}
@@ -149,9 +149,9 @@ public class AfterProcessingTimeTest {
*/
@Test
public void testCompatibilityIdentical() throws Exception {
- Trigger<?> t1 = AfterProcessingTime.pastFirstElementInPane()
+ Trigger t1 = AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.standardMinutes(1L));
- Trigger<?> t2 = AfterProcessingTime.pastFirstElementInPane()
+ Trigger t2 = AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.standardMinutes(1L));
assertTrue(t1.isCompatible(t2));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
index c9d09f7..047f0ca 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTimeTest.java
@@ -36,8 +36,7 @@ import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class AfterSynchronizedProcessingTimeTest {
- private Trigger<IntervalWindow> underTest =
- new AfterSynchronizedProcessingTime<IntervalWindow>();
+ private Trigger underTest = new AfterSynchronizedProcessingTime();
@Test
public void testAfterProcessingTimeWithFixedWindows() throws Exception {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermarkTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermarkTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermarkTest.java
index ca909f9..7417953 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermarkTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermarkTest.java
@@ -42,15 +42,15 @@ import org.mockito.MockitoAnnotations;
@RunWith(JUnit4.class)
public class AfterWatermarkTest {
- @Mock private OnceTrigger<IntervalWindow> mockEarly;
- @Mock private OnceTrigger<IntervalWindow> mockLate;
+ @Mock private OnceTrigger mockEarly;
+ @Mock private OnceTrigger mockLate;
private SimpleTriggerTester<IntervalWindow> tester;
- private static Trigger<IntervalWindow>.TriggerContext anyTriggerContext() {
- return Mockito.<Trigger<IntervalWindow>.TriggerContext>any();
+ private static Trigger.TriggerContext anyTriggerContext() {
+ return Mockito.<Trigger.TriggerContext>any();
}
- private static Trigger<IntervalWindow>.OnElementContext anyElementContext() {
- return Mockito.<Trigger<IntervalWindow>.OnElementContext>any();
+ private static Trigger.OnElementContext anyElementContext() {
+ return Mockito.<Trigger.OnElementContext>any();
}
private void injectElements(int... elements) throws Exception {
@@ -66,7 +66,7 @@ public class AfterWatermarkTest {
MockitoAnnotations.initMocks(this);
}
- public void testRunningAsTrigger(OnceTrigger<IntervalWindow> mockTrigger, IntervalWindow window)
+ public void testRunningAsTrigger(OnceTrigger mockTrigger, IntervalWindow window)
throws Exception {
// Don't fire due to mock saying no
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java
index 8635873..22dde66 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTriggerTest.java
@@ -46,7 +46,7 @@ public class OrFinallyTriggerTest {
@Test
public void testActualFiresAndFinishes() throws Exception {
tester = TriggerTester.forTrigger(
- new OrFinallyTrigger<>(
+ new OrFinallyTrigger(
AfterPane.<IntervalWindow>elementCountAtLeast(2),
AfterPane.<IntervalWindow>elementCountAtLeast(100)),
FixedWindows.of(Duration.millis(100)));
@@ -73,7 +73,7 @@ public class OrFinallyTriggerTest {
@Test
public void testActualFiresOnly() throws Exception {
tester = TriggerTester.forTrigger(
- new OrFinallyTrigger<>(
+ new OrFinallyTrigger(
Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2)),
AfterPane.<IntervalWindow>elementCountAtLeast(100)),
FixedWindows.of(Duration.millis(100)));
@@ -143,7 +143,7 @@ public class OrFinallyTriggerTest {
@Test
public void testActualFiresButUntilFinishes() throws Exception {
tester = TriggerTester.forTrigger(
- new OrFinallyTrigger<IntervalWindow>(
+ new OrFinallyTrigger(
Repeatedly.forever(AfterPane.<IntervalWindow>elementCountAtLeast(2)),
AfterPane.<IntervalWindow>elementCountAtLeast(3)),
FixedWindows.of(Duration.millis(10)));
@@ -194,10 +194,10 @@ public class OrFinallyTriggerTest {
@Test
public void testContinuation() throws Exception {
- OnceTrigger<IntervalWindow> triggerA = AfterProcessingTime.pastFirstElementInPane();
- OnceTrigger<IntervalWindow> triggerB = AfterWatermark.pastEndOfWindow();
- Trigger<IntervalWindow> aOrFinallyB = triggerA.orFinally(triggerB);
- Trigger<IntervalWindow> bOrFinallyA = triggerB.orFinally(triggerA);
+ OnceTrigger triggerA = AfterProcessingTime.pastFirstElementInPane();
+ OnceTrigger triggerB = AfterWatermark.pastEndOfWindow();
+ Trigger aOrFinallyB = triggerA.orFinally(triggerB);
+ Trigger bOrFinallyA = triggerB.orFinally(triggerA);
assertEquals(
Repeatedly.forever(
triggerA.getContinuationTrigger().orFinally(triggerB.getContinuationTrigger())),
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
index 99907b2..c98090e 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/RepeatedlyTest.java
@@ -43,10 +43,10 @@ import org.mockito.MockitoAnnotations;
@RunWith(JUnit4.class)
public class RepeatedlyTest {
- @Mock private Trigger<IntervalWindow> mockTrigger;
+ @Mock private Trigger mockTrigger;
private SimpleTriggerTester<IntervalWindow> tester;
- private static Trigger<IntervalWindow>.TriggerContext anyTriggerContext() {
- return Mockito.<Trigger<IntervalWindow>.TriggerContext>any();
+ private static Trigger.TriggerContext anyTriggerContext() {
+ return Mockito.<Trigger.TriggerContext>any();
}
public void setUp(WindowFn<Object, IntervalWindow> windowFn) throws Exception {
@@ -61,7 +61,7 @@ public class RepeatedlyTest {
public void testOnElement() throws Exception {
setUp(FixedWindows.of(Duration.millis(10)));
tester.injectElements(37);
- verify(mockTrigger).onElement(Mockito.<Trigger<IntervalWindow>.OnElementContext>any());
+ verify(mockTrigger).onElement(Mockito.<Trigger.OnElementContext>any());
}
/**
@@ -74,7 +74,7 @@ public class RepeatedlyTest {
when(mockTrigger.shouldFire(anyTriggerContext())).thenReturn(true);
assertTrue(tester.shouldFire(new IntervalWindow(new Instant(0), new Instant(10))));
- when(mockTrigger.shouldFire(Mockito.<Trigger<IntervalWindow>.TriggerContext>any()))
+ when(mockTrigger.shouldFire(Mockito.<Trigger.TriggerContext>any()))
.thenReturn(false);
assertFalse(tester.shouldFire(new IntervalWindow(new Instant(0), new Instant(10))));
}
@@ -98,8 +98,8 @@ public class RepeatedlyTest {
@Test
public void testContinuation() throws Exception {
- Trigger<IntervalWindow> trigger = AfterProcessingTime.pastFirstElementInPane();
- Trigger<IntervalWindow> repeatedly = Repeatedly.forever(trigger);
+ Trigger trigger = AfterProcessingTime.pastFirstElementInPane();
+ Trigger repeatedly = Repeatedly.forever(trigger);
assertEquals(
Repeatedly.forever(trigger.getContinuationTrigger()), repeatedly.getContinuationTrigger());
assertEquals(
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerTest.java
index 50b30ee..8bc0826 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerTest.java
@@ -45,75 +45,75 @@ public class TriggerTest {
@Test
public void testIsCompatible() throws Exception {
assertTrue(new Trigger1(null).isCompatible(new Trigger1(null)));
- assertTrue(new Trigger1(Arrays.<Trigger<IntervalWindow>>asList(new Trigger2(null)))
- .isCompatible(new Trigger1(Arrays.<Trigger<IntervalWindow>>asList(new Trigger2(null)))));
+ assertTrue(new Trigger1(Arrays.<Trigger>asList(new Trigger2(null)))
+ .isCompatible(new Trigger1(Arrays.<Trigger>asList(new Trigger2(null)))));
assertFalse(new Trigger1(null).isCompatible(new Trigger2(null)));
- assertFalse(new Trigger1(Arrays.<Trigger<IntervalWindow>>asList(new Trigger1(null)))
- .isCompatible(new Trigger1(Arrays.<Trigger<IntervalWindow>>asList(new Trigger2(null)))));
+ assertFalse(new Trigger1(Arrays.<Trigger>asList(new Trigger1(null)))
+ .isCompatible(new Trigger1(Arrays.<Trigger>asList(new Trigger2(null)))));
}
- private static class Trigger1 extends Trigger<IntervalWindow> {
+ private static class Trigger1 extends Trigger {
- private Trigger1(List<Trigger<IntervalWindow>> subTriggers) {
+ private Trigger1(List<Trigger> subTriggers) {
super(subTriggers);
}
@Override
- public void onElement(Trigger<IntervalWindow>.OnElementContext c) { }
+ public void onElement(Trigger.OnElementContext c) { }
@Override
- public void onMerge(Trigger<IntervalWindow>.OnMergeContext c) { }
+ public void onMerge(Trigger.OnMergeContext c) { }
@Override
- protected Trigger<IntervalWindow> getContinuationTrigger(
- List<Trigger<IntervalWindow>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(
+ List<Trigger> continuationTriggers) {
return null;
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(IntervalWindow window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return null;
}
@Override
- public boolean shouldFire(Trigger<IntervalWindow>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return false;
}
@Override
- public void onFire(Trigger<IntervalWindow>.TriggerContext context) throws Exception { }
+ public void onFire(Trigger.TriggerContext context) throws Exception { }
}
- private static class Trigger2 extends Trigger<IntervalWindow> {
+ private static class Trigger2 extends Trigger {
- private Trigger2(List<Trigger<IntervalWindow>> subTriggers) {
+ private Trigger2(List<Trigger> subTriggers) {
super(subTriggers);
}
@Override
- public void onElement(Trigger<IntervalWindow>.OnElementContext c) { }
+ public void onElement(Trigger.OnElementContext c) { }
@Override
- public void onMerge(Trigger<IntervalWindow>.OnMergeContext c) { }
+ public void onMerge(Trigger.OnMergeContext c) { }
@Override
- protected Trigger<IntervalWindow> getContinuationTrigger(
- List<Trigger<IntervalWindow>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(
+ List<Trigger> continuationTriggers) {
return null;
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(IntervalWindow window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return null;
}
@Override
- public boolean shouldFire(Trigger<IntervalWindow>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return false;
}
@Override
- public void onFire(Trigger<IntervalWindow>.TriggerContext context) throws Exception { }
+ public void onFire(Trigger.TriggerContext context) throws Exception { }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/WindowTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/WindowTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/WindowTest.java
index c4eda99..39a33cc 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/WindowTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/windowing/WindowTest.java
@@ -74,7 +74,7 @@ public class WindowTest implements Serializable {
@Test
public void testWindowIntoTriggersAndAccumulating() {
FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
- Repeatedly<BoundedWindow> trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
+ Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
WindowingStrategy<?, ?> strategy = TestPipeline.create()
.apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
.apply(Window.<String>into(fixed10)
@@ -91,7 +91,7 @@ public class WindowTest implements Serializable {
@Test
public void testWindowPropagatesEachPart() {
FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
- Repeatedly<BoundedWindow> trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
+ Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
WindowingStrategy<?, ?> strategy = TestPipeline.create()
.apply(Create.of("hello", "world").withCoder(StringUtf8Coder.of()))
.apply("Mode", Window.<String>accumulatingFiredPanes())
@@ -149,7 +149,7 @@ public class WindowTest implements Serializable {
@Test
public void testMissingMode() {
FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
- Repeatedly<BoundedWindow> trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
+ Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage("requires that the accumulation mode");
@@ -163,7 +163,7 @@ public class WindowTest implements Serializable {
@Test
public void testMissingLateness() {
FixedWindows fixed10 = FixedWindows.of(Duration.standardMinutes(10));
- Repeatedly<BoundedWindow> trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
+ Repeatedly trigger = Repeatedly.forever(AfterPane.elementCountAtLeast(5));
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage("requires that the allowed lateness");
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ExecutableTriggerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ExecutableTriggerTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ExecutableTriggerTest.java
index 16d2640..c975e08 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ExecutableTriggerTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ExecutableTriggerTest.java
@@ -21,7 +21,6 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import com.google.cloud.dataflow.sdk.transforms.windowing.BoundedWindow;
-import com.google.cloud.dataflow.sdk.transforms.windowing.IntervalWindow;
import com.google.cloud.dataflow.sdk.transforms.windowing.Trigger;
import org.joda.time.Instant;
@@ -41,7 +40,7 @@ public class ExecutableTriggerTest {
@Test
public void testIndexAssignmentLeaf() throws Exception {
StubTrigger t1 = new StubTrigger();
- ExecutableTrigger<?> executable = ExecutableTrigger.create(t1);
+ ExecutableTrigger executable = ExecutableTrigger.create(t1);
assertEquals(0, executable.getTriggerIndex());
}
@@ -51,7 +50,7 @@ public class ExecutableTriggerTest {
StubTrigger t2 = new StubTrigger();
StubTrigger t = new StubTrigger(t1, t2);
- ExecutableTrigger<?> executable = ExecutableTrigger.create(t);
+ ExecutableTrigger executable = ExecutableTrigger.create(t);
assertEquals(0, executable.getTriggerIndex());
assertEquals(1, executable.subTriggers().get(0).getTriggerIndex());
@@ -72,7 +71,7 @@ public class ExecutableTriggerTest {
StubTrigger t2 = new StubTrigger(t21, t22);
StubTrigger t = new StubTrigger(t1, t2);
- ExecutableTrigger<?> executable = ExecutableTrigger.create(t);
+ ExecutableTrigger executable = ExecutableTrigger.create(t);
assertEquals(0, executable.getTriggerIndex());
assertEquals(1, executable.subTriggers().get(0).getTriggerIndex());
@@ -87,10 +86,10 @@ public class ExecutableTriggerTest {
assertSame(t2, executable.getSubTriggerContaining(7).getSpec());
}
- private static class StubTrigger extends Trigger<IntervalWindow> {
+ private static class StubTrigger extends Trigger {
@SafeVarargs
- protected StubTrigger(Trigger<IntervalWindow>... subTriggers) {
+ protected StubTrigger(Trigger... subTriggers) {
super(Arrays.asList(subTriggers));
}
@@ -105,18 +104,17 @@ public class ExecutableTriggerTest {
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(IntervalWindow window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- public boolean isCompatible(Trigger<?> other) {
+ public boolean isCompatible(Trigger other) {
return false;
}
@Override
- public Trigger<IntervalWindow> getContinuationTrigger(
- List<Trigger<IntervalWindow>> continuationTriggers) {
+ public Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersProperties.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersProperties.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersProperties.java
index 3460f40..bc2a514 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersProperties.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersProperties.java
@@ -34,7 +34,7 @@ public class FinishedTriggersProperties {
* Tests that for the provided trigger and {@link FinishedTriggers}, when the trigger is set
* finished, it is correctly reported as finished.
*/
- public static void verifyGetAfterSet(FinishedTriggers finishedSet, ExecutableTrigger<?> trigger) {
+ public static void verifyGetAfterSet(FinishedTriggers finishedSet, ExecutableTrigger trigger) {
assertFalse(finishedSet.isFinished(trigger));
finishedSet.setFinished(trigger, true);
assertTrue(finishedSet.isFinished(trigger));
@@ -45,7 +45,7 @@ public class FinishedTriggersProperties {
* reported as finished.
*/
public static void verifyGetAfterSet(FinishedTriggers finishedSet) {
- ExecutableTrigger<?> trigger = ExecutableTrigger.create(AfterAll.of(
+ ExecutableTrigger trigger = ExecutableTrigger.create(AfterAll.of(
AfterFirst.of(AfterPane.elementCountAtLeast(3), AfterWatermark.pastEndOfWindow()),
AfterAll.of(
AfterPane.elementCountAtLeast(10), AfterProcessingTime.pastFirstElementInPane())));
@@ -63,7 +63,7 @@ public class FinishedTriggersProperties {
* others.
*/
public static void verifyClearRecursively(FinishedTriggers finishedSet) {
- ExecutableTrigger<?> trigger = ExecutableTrigger.create(AfterAll.of(
+ ExecutableTrigger trigger = ExecutableTrigger.create(AfterAll.of(
AfterFirst.of(AfterPane.elementCountAtLeast(3), AfterWatermark.pastEndOfWindow()),
AfterAll.of(
AfterPane.elementCountAtLeast(10), AfterProcessingTime.pastFirstElementInPane())));
@@ -85,25 +85,25 @@ public class FinishedTriggersProperties {
}
private static void setFinishedRecursively(
- FinishedTriggers finishedSet, ExecutableTrigger<?> trigger) {
+ FinishedTriggers finishedSet, ExecutableTrigger trigger) {
finishedSet.setFinished(trigger, true);
- for (ExecutableTrigger<?> subTrigger : trigger.subTriggers()) {
+ for (ExecutableTrigger subTrigger : trigger.subTriggers()) {
setFinishedRecursively(finishedSet, subTrigger);
}
}
private static void verifyFinishedRecursively(
- FinishedTriggers finishedSet, ExecutableTrigger<?> trigger) {
+ FinishedTriggers finishedSet, ExecutableTrigger trigger) {
assertTrue(finishedSet.isFinished(trigger));
- for (ExecutableTrigger<?> subTrigger : trigger.subTriggers()) {
+ for (ExecutableTrigger subTrigger : trigger.subTriggers()) {
verifyFinishedRecursively(finishedSet, subTrigger);
}
}
private static void verifyUnfinishedRecursively(
- FinishedTriggers finishedSet, ExecutableTrigger<?> trigger) {
+ FinishedTriggers finishedSet, ExecutableTrigger trigger) {
assertFalse(finishedSet.isFinished(trigger));
- for (ExecutableTrigger<?> subTrigger : trigger.subTriggers()) {
+ for (ExecutableTrigger subTrigger : trigger.subTriggers()) {
verifyUnfinishedRecursively(finishedSet, subTrigger);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSetTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSetTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSetTest.java
index ca648aa..a84b73e 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSetTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSetTest.java
@@ -38,7 +38,7 @@ public class FinishedTriggersSetTest {
@Test
public void testSetGet() {
FinishedTriggersProperties.verifyGetAfterSet(
- FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger<?>>()));
+ FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger>()));
}
/**
@@ -48,13 +48,13 @@ public class FinishedTriggersSetTest {
@Test
public void testClearRecursively() {
FinishedTriggersProperties.verifyClearRecursively(
- FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger<?>>()));
+ FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger>()));
}
@Test
public void testCopy() throws Exception {
FinishedTriggersSet finishedSet =
- FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger<?>>());
+ FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger>());
assertThat(finishedSet.copy().getFinishedTriggers(),
not(theInstance(finishedSet.getFinishedTriggers())));
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnRunnerTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnRunnerTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnRunnerTest.java
index b2d246e..ef499f2 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnRunnerTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnRunnerTest.java
@@ -85,26 +85,23 @@ import java.util.List;
@RunWith(JUnit4.class)
public class ReduceFnRunnerTest {
@Mock private SideInputReader mockSideInputReader;
- private Trigger<IntervalWindow> mockTrigger;
+ private Trigger mockTrigger;
private PCollectionView<Integer> mockView;
private IntervalWindow firstWindow;
- private static Trigger<IntervalWindow>.TriggerContext anyTriggerContext() {
- return Mockito.<Trigger<IntervalWindow>.TriggerContext>any();
+ private static Trigger.TriggerContext anyTriggerContext() {
+ return Mockito.<Trigger.TriggerContext>any();
}
- private static Trigger<IntervalWindow>.OnElementContext anyElementContext() {
- return Mockito.<Trigger<IntervalWindow>.OnElementContext>any();
+ private static Trigger.OnElementContext anyElementContext() {
+ return Mockito.<Trigger.OnElementContext>any();
}
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
- @SuppressWarnings("unchecked")
- Trigger<IntervalWindow> mockTriggerUnchecked =
- mock(Trigger.class, withSettings().serializable());
- mockTrigger = mockTriggerUnchecked;
+ mockTrigger = mock(Trigger.class, withSettings().serializable());
when(mockTrigger.buildTrigger()).thenReturn(mockTrigger);
@SuppressWarnings("unchecked")
@@ -120,13 +117,13 @@ public class ReduceFnRunnerTest {
tester.injectElements(TimestampedValue.of(element, new Instant(element)));
}
- private void triggerShouldFinish(Trigger<IntervalWindow> mockTrigger) throws Exception {
+ private void triggerShouldFinish(Trigger mockTrigger) throws Exception {
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Exception {
@SuppressWarnings("unchecked")
- Trigger<IntervalWindow>.TriggerContext context =
- (Trigger<IntervalWindow>.TriggerContext) invocation.getArguments()[0];
+ Trigger.TriggerContext context =
+ (Trigger.TriggerContext) invocation.getArguments()[0];
context.trigger().setFinished(true);
return null;
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java
index f10341b..86c2d11 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/ReduceFnTester.java
@@ -114,7 +114,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
*/
private boolean autoAdvanceOutputWatermark;
- private ExecutableTrigger<W> executableTrigger;
+ private ExecutableTrigger executableTrigger;
private final InMemoryLongSumAggregator droppedDueToClosedWindow =
new InMemoryLongSumAggregator(GroupAlsoByWindowsDoFn.DROPPED_DUE_TO_CLOSED_WINDOW_COUNTER);
@@ -130,7 +130,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
}
public static <W extends BoundedWindow> ReduceFnTester<Integer, Iterable<Integer>, W>
- nonCombining(WindowFn<?, W> windowFn, TriggerBuilder<W> trigger, AccumulationMode mode,
+ nonCombining(WindowFn<?, W> windowFn, TriggerBuilder trigger, AccumulationMode mode,
Duration allowedDataLateness, ClosingBehavior closingBehavior) throws Exception {
WindowingStrategy<?, W> strategy =
WindowingStrategy.of(windowFn)
@@ -180,7 +180,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
sideInputReader);
}
public static <W extends BoundedWindow, AccumT, OutputT> ReduceFnTester<Integer, OutputT, W>
- combining(WindowFn<?, W> windowFn, Trigger<W> trigger, AccumulationMode mode,
+ combining(WindowFn<?, W> windowFn, Trigger trigger, AccumulationMode mode,
KeyedCombineFn<String, Integer, AccumT, OutputT> combineFn, Coder<OutputT> outputCoder,
Duration allowedDataLateness) throws Exception {
@@ -228,7 +228,7 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
options);
}
- public ExecutableTrigger<W> getTrigger() {
+ public ExecutableTrigger getTrigger() {
return executableTrigger;
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java
index 75708c4..61eeb90 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/util/TriggerTester.java
@@ -109,7 +109,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
* An {@link ExecutableTrigger} built from the {@link Trigger} or {@link TriggerBuilder}
* under test.
*/
- private final ExecutableTrigger<W> executableTrigger;
+ private final ExecutableTrigger executableTrigger;
/**
* A map from a window and trigger to whether that trigger is finished for the window.
@@ -117,7 +117,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
private final Map<W, FinishedTriggers> finishedSets;
public static <W extends BoundedWindow> SimpleTriggerTester<W> forTrigger(
- TriggerBuilder<W> trigger, WindowFn<Object, W> windowFn)
+ TriggerBuilder trigger, WindowFn<Object, W> windowFn)
throws Exception {
WindowingStrategy<Object, W> windowingStrategy =
WindowingStrategy.of(windowFn).withTrigger(trigger.buildTrigger())
@@ -132,7 +132,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
}
public static <InputT, W extends BoundedWindow> TriggerTester<InputT, W> forAdvancedTrigger(
- TriggerBuilder<W> trigger, WindowFn<Object, W> windowFn) throws Exception {
+ TriggerBuilder trigger, WindowFn<Object, W> windowFn) throws Exception {
WindowingStrategy<Object, W> strategy =
WindowingStrategy.of(windowFn).withTrigger(trigger.buildTrigger())
// Merging requires accumulation mode or early firings can break up a session.
@@ -265,7 +265,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
@SuppressWarnings("unchecked")
W window = activeWindows.mergeResultWindow((W) untypedWindow);
- Trigger<W>.OnElementContext context = contextFactory.createOnElementContext(window,
+ Trigger.OnElementContext context = contextFactory.createOnElementContext(window,
new TestTimers(windowNamespace(window)), windowedValue.getTimestamp(),
executableTrigger, getFinishedSet(window));
@@ -277,7 +277,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
}
public boolean shouldFire(W window) throws Exception {
- Trigger<W>.TriggerContext context = contextFactory.base(
+ Trigger.TriggerContext context = contextFactory.base(
window,
new TestTimers(windowNamespace(window)),
executableTrigger, getFinishedSet(window));
@@ -286,7 +286,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
}
public void fireIfShouldFire(W window) throws Exception {
- Trigger<W>.TriggerContext context = contextFactory.base(
+ Trigger.TriggerContext context = contextFactory.base(
window,
new TestTimers(windowNamespace(window)),
executableTrigger, getFinishedSet(window));
@@ -296,7 +296,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
executableTrigger.getSpec().prefetchOnFire(context.state());
executableTrigger.invokeOnFire(context);
if (context.trigger().isFinished()) {
- activeWindows.remove(context.window());
+ activeWindows.remove(window);
executableTrigger.invokeClear(context);
}
}
@@ -337,7 +337,7 @@ public class TriggerTester<InputT, W extends BoundedWindow> {
private FinishedTriggers getFinishedSet(W window) {
FinishedTriggers finishedSet = finishedSets.get(window);
if (finishedSet == null) {
- finishedSet = FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger<?>>());
+ finishedSet = FinishedTriggersSet.fromSet(new HashSet<ExecutableTrigger>());
finishedSets.put(window, finishedSet);
}
return finishedSet;
[2/3] incubator-beam git commit: Remove extraneous type parameter
from Trigger, etc
Posted by ke...@apache.org.
Remove extraneous type parameter from Trigger, etc
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/de9f10c4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/de9f10c4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/de9f10c4
Branch: refs/heads/master
Commit: de9f10c4b23912b590dd66e42a4fc4e0513fe6fd
Parents: ab7bca7
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon Apr 11 14:49:21 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Mon Apr 11 15:12:52 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/windowing/AfterAll.java | 30 +++---
.../windowing/AfterDelayFromFirstElement.java | 28 ++---
.../sdk/transforms/windowing/AfterEach.java | 29 +++---
.../sdk/transforms/windowing/AfterFirst.java | 33 +++---
.../sdk/transforms/windowing/AfterPane.java | 23 ++---
.../windowing/AfterProcessingTime.java | 22 ++--
.../AfterSynchronizedProcessingTime.java | 11 +-
.../transforms/windowing/AfterWatermark.java | 102 +++++++++----------
.../transforms/windowing/DefaultTrigger.java | 20 ++--
.../transforms/windowing/OrFinallyTrigger.java | 24 ++---
.../sdk/transforms/windowing/Repeatedly.java | 21 ++--
.../sdk/transforms/windowing/Trigger.java | 85 +++++++---------
.../transforms/windowing/TriggerBuilder.java | 6 +-
.../sdk/transforms/windowing/Window.java | 20 ++--
.../dataflow/sdk/util/ExecutableTrigger.java | 58 +++++------
.../dataflow/sdk/util/FinishedTriggers.java | 6 +-
.../sdk/util/FinishedTriggersBitSet.java | 6 +-
.../dataflow/sdk/util/FinishedTriggersSet.java | 16 +--
.../dataflow/sdk/util/ReshuffleTrigger.java | 14 +--
.../sdk/util/TriggerContextFactory.java | 80 +++++++--------
.../cloud/dataflow/sdk/util/TriggerRunner.java | 12 +--
.../dataflow/sdk/util/WindowingStrategy.java | 12 +--
.../sdk/transforms/windowing/AfterAllTest.java | 6 +-
.../sdk/transforms/windowing/AfterEachTest.java | 6 +-
.../transforms/windowing/AfterFirstTest.java | 14 +--
.../windowing/AfterProcessingTimeTest.java | 8 +-
.../AfterSynchronizedProcessingTimeTest.java | 3 +-
.../windowing/AfterWatermarkTest.java | 14 +--
.../windowing/OrFinallyTriggerTest.java | 14 +--
.../transforms/windowing/RepeatedlyTest.java | 14 +--
.../sdk/transforms/windowing/TriggerTest.java | 44 ++++----
.../sdk/transforms/windowing/WindowTest.java | 8 +-
.../sdk/util/ExecutableTriggerTest.java | 18 ++--
.../sdk/util/FinishedTriggersProperties.java | 18 ++--
.../sdk/util/FinishedTriggersSetTest.java | 6 +-
.../dataflow/sdk/util/ReduceFnRunnerTest.java | 21 ++--
.../cloud/dataflow/sdk/util/ReduceFnTester.java | 8 +-
.../cloud/dataflow/sdk/util/TriggerTester.java | 16 +--
38 files changed, 419 insertions(+), 457 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java
index a315692..97f5616 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterAll.java
@@ -29,14 +29,11 @@ import java.util.List;
/**
* Create a {@link Trigger} that fires and finishes once after all of its sub-triggers have fired.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code Trigger}
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
+public class AfterAll extends OnceTrigger {
- private AfterAll(List<Trigger<W>> subTriggers) {
+ private AfterAll(List<Trigger> subTriggers) {
super(subTriggers);
Preconditions.checkArgument(subTriggers.size() > 1);
}
@@ -45,14 +42,13 @@ public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
* Returns an {@code AfterAll} {@code Trigger} with the given subtriggers.
*/
@SafeVarargs
- public static <W extends BoundedWindow> OnceTrigger<W> of(
- OnceTrigger<W>... triggers) {
- return new AfterAll<W>(Arrays.<Trigger<W>>asList(triggers));
+ public static <W extends BoundedWindow> OnceTrigger of(OnceTrigger... triggers) {
+ return new AfterAll(Arrays.<Trigger>asList(triggers));
}
@Override
public void onElement(OnElementContext c) throws Exception {
- for (ExecutableTrigger<W> subTrigger : c.trigger().unfinishedSubTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().unfinishedSubTriggers()) {
// Since subTriggers are all OnceTriggers, they must either CONTINUE or FIRE_AND_FINISH.
// invokeElement will automatically mark the finish bit if they return FIRE_AND_FINISH.
subTrigger.invokeOnElement(c);
@@ -61,21 +57,21 @@ public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
@Override
public void onMerge(OnMergeContext c) throws Exception {
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
subTrigger.invokeOnMerge(c);
}
boolean allFinished = true;
- for (ExecutableTrigger<W> subTrigger1 : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger1 : c.trigger().subTriggers()) {
allFinished &= c.forTrigger(subTrigger1).trigger().isFinished();
}
c.trigger().setFinished(allFinished);
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// This trigger will fire after the latest of its sub-triggers.
Instant deadline = BoundedWindow.TIMESTAMP_MIN_VALUE;
- for (Trigger<W> subTrigger : subTriggers) {
+ for (Trigger subTrigger : subTriggers) {
Instant subDeadline = subTrigger.getWatermarkThatGuaranteesFiring(window);
if (deadline.isBefore(subDeadline)) {
deadline = subDeadline;
@@ -85,8 +81,8 @@ public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
}
@Override
- public OnceTrigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
- return new AfterAll<W>(continuationTriggers);
+ public OnceTrigger getContinuationTrigger(List<Trigger> continuationTriggers) {
+ return new AfterAll(continuationTriggers);
}
/**
@@ -96,7 +92,7 @@ public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
*/
@Override
public boolean shouldFire(TriggerContext context) throws Exception {
- for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+ for (ExecutableTrigger subtrigger : context.trigger().subTriggers()) {
if (!context.forTrigger(subtrigger).trigger().isFinished()
&& !subtrigger.invokeShouldFire(context)) {
return false;
@@ -111,7 +107,7 @@ public class AfterAll<W extends BoundedWindow> extends OnceTrigger<W> {
*/
@Override
public void onOnlyFiring(TriggerContext context) throws Exception {
- for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+ for (ExecutableTrigger subtrigger : context.trigger().subTriggers()) {
subtrigger.invokeOnFire(context);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java
index bc626b2..2143b0c 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterDelayFromFirstElement.java
@@ -48,7 +48,7 @@ import javax.annotation.Nullable;
* <p>This class is for internal use only and may change at any time.
*/
@Experimental(Experimental.Kind.TRIGGER)
-public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extends OnceTrigger<W> {
+public abstract class AfterDelayFromFirstElement extends OnceTrigger {
protected static final List<SerializableFunction<Instant, Instant>> IDENTITY =
ImmutableList.<SerializableFunction<Instant, Instant>>of();
@@ -62,14 +62,14 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
* To complete an implementation, return the desired time from the TriggerContext.
*/
@Nullable
- public abstract Instant getCurrentTime(Trigger<W>.TriggerContext context);
+ public abstract Instant getCurrentTime(Trigger.TriggerContext context);
/**
* To complete an implementation, return a new instance like this one, but incorporating
* the provided timestamp mapping functions. Generally should be used by calling the
* constructor of this class from the constructor of the subclass.
*/
- protected abstract AfterDelayFromFirstElement<W> newWith(
+ protected abstract AfterDelayFromFirstElement newWith(
List<SerializableFunction<Instant, Instant>> transform);
/**
@@ -100,7 +100,7 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
* <p>TODO: Consider sharing this with FixedWindows, and bring over the equivalent of
* CalendarWindows.
*/
- public AfterDelayFromFirstElement<W> alignedTo(final Duration size, final Instant offset) {
+ public AfterDelayFromFirstElement alignedTo(final Duration size, final Instant offset) {
return newWith(new AlignFn(size, offset));
}
@@ -108,7 +108,7 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
* Aligns the time to be the smallest multiple of {@code size} greater than the timestamp
* since the epoch.
*/
- public AfterDelayFromFirstElement<W> alignedTo(final Duration size) {
+ public AfterDelayFromFirstElement alignedTo(final Duration size) {
return alignedTo(size, new Instant(0));
}
@@ -118,7 +118,7 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
* @param delay the delay to add
* @return An updated time trigger that will wait the additional time before firing.
*/
- public AfterDelayFromFirstElement<W> plusDelayOf(final Duration delay) {
+ public AfterDelayFromFirstElement plusDelayOf(final Duration delay) {
return newWith(new DelayFn(delay));
}
@@ -127,22 +127,22 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
* {@link #plusDelayOf} and {@link #alignedTo}.
*/
@Deprecated
- public OnceTrigger<W> mappedTo(SerializableFunction<Instant, Instant> timestampMapper) {
+ public OnceTrigger mappedTo(SerializableFunction<Instant, Instant> timestampMapper) {
return newWith(timestampMapper);
}
@Override
- public boolean isCompatible(Trigger<?> other) {
+ public boolean isCompatible(Trigger other) {
if (!getClass().equals(other.getClass())) {
return false;
}
- AfterDelayFromFirstElement<?> that = (AfterDelayFromFirstElement<?>) other;
+ AfterDelayFromFirstElement that = (AfterDelayFromFirstElement) other;
return this.timestampMappers.equals(that.timestampMappers);
}
- private AfterDelayFromFirstElement<W> newWith(
+ private AfterDelayFromFirstElement newWith(
SerializableFunction<Instant, Instant> timestampMapper) {
return newWith(
ImmutableList.<SerializableFunction<Instant, Instant>>builder()
@@ -173,7 +173,7 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
}
@Override
- public void prefetchOnMerge(MergingStateAccessor<?, W> state) {
+ public void prefetchOnMerge(MergingStateAccessor<?, ?> state) {
super.prefetchOnMerge(state);
StateMerging.prefetchCombiningValues(state, DELAYED_UNTIL_TAG);
}
@@ -218,12 +218,12 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
Instant delayedUntil = context.state().access(DELAYED_UNTIL_TAG).read();
return delayedUntil != null
&& getCurrentTime(context) != null
@@ -231,7 +231,7 @@ public abstract class AfterDelayFromFirstElement<W extends BoundedWindow> extend
}
@Override
- protected void onOnlyFiring(Trigger<W>.TriggerContext context) throws Exception {
+ protected void onOnlyFiring(Trigger.TriggerContext context) throws Exception {
clear(context);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java
index 164cd87..e35cc15 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterEach.java
@@ -42,14 +42,11 @@ import java.util.List;
* <li> {@code AfterEach.inOrder(Repeatedly.forever(a), b)} behaves the same as
* {@code Repeatedly.forever(a)}, since the repeated trigger never finishes.
* </ul>
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code Trigger}
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
+public class AfterEach extends Trigger {
- private AfterEach(List<Trigger<W>> subTriggers) {
+ private AfterEach(List<Trigger> subTriggers) {
super(subTriggers);
checkArgument(subTriggers.size() > 1);
}
@@ -58,8 +55,8 @@ public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
* Returns an {@code AfterEach} {@code Trigger} with the given subtriggers.
*/
@SafeVarargs
- public static <W extends BoundedWindow> Trigger<W> inOrder(Trigger<W>... triggers) {
- return new AfterEach<W>(Arrays.<Trigger<W>>asList(triggers));
+ public static <W extends BoundedWindow> Trigger inOrder(Trigger... triggers) {
+ return new AfterEach(Arrays.<Trigger>asList(triggers));
}
@Override
@@ -69,7 +66,7 @@ public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
c.trigger().firstUnfinishedSubTrigger().invokeOnElement(c);
} else {
// If merges are possible, we need to run all subtriggers in parallel
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
// Even if the subTrigger is done, it may be revived via merging and must have
// adequate state.
subTrigger.invokeOnElement(c);
@@ -86,7 +83,7 @@ public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
// also automatic because they are cleared whenever this trigger
// fires.
boolean priorTriggersAllFinished = true;
- for (ExecutableTrigger<W> subTrigger : context.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : context.trigger().subTriggers()) {
if (priorTriggersAllFinished) {
subTrigger.invokeOnMerge(context);
priorTriggersAllFinished &= context.forTrigger(subTrigger).trigger().isFinished();
@@ -98,31 +95,31 @@ public class AfterEach<W extends BoundedWindow> extends Trigger<W> {
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// This trigger will fire at least once when the first trigger in the sequence
// fires at least once.
return subTriggers.get(0).getWatermarkThatGuaranteesFiring(window);
}
@Override
- public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
- return Repeatedly.forever(new AfterFirst<W>(continuationTriggers));
+ public Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
+ return Repeatedly.forever(new AfterFirst(continuationTriggers));
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
- ExecutableTrigger<W> firstUnfinished = context.trigger().firstUnfinishedSubTrigger();
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
+ ExecutableTrigger firstUnfinished = context.trigger().firstUnfinishedSubTrigger();
return firstUnfinished.invokeShouldFire(context);
}
@Override
- public void onFire(Trigger<W>.TriggerContext context) throws Exception {
+ public void onFire(Trigger.TriggerContext context) throws Exception {
context.trigger().firstUnfinishedSubTrigger().invokeOnFire(context);
// Reset all subtriggers if in a merging context; any may be revived by merging so they are
// all run in parallel for each pending pane.
if (context.trigger().isMerging()) {
- for (ExecutableTrigger<W> subTrigger : context.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : context.trigger().subTriggers()) {
subTrigger.invokeClear(context);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirst.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirst.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirst.java
index d6817ea..b7d491e 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirst.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterFirst.java
@@ -30,14 +30,11 @@ import java.util.List;
/**
* Create a composite {@link Trigger} that fires once after at least one of its sub-triggers have
* fired.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code Trigger}
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterFirst<W extends BoundedWindow> extends OnceTrigger<W> {
+public class AfterFirst extends OnceTrigger {
- AfterFirst(List<Trigger<W>> subTriggers) {
+ AfterFirst(List<Trigger> subTriggers) {
super(subTriggers);
Preconditions.checkArgument(subTriggers.size() > 1);
}
@@ -46,31 +43,31 @@ public class AfterFirst<W extends BoundedWindow> extends OnceTrigger<W> {
* Returns an {@code AfterFirst} {@code Trigger} with the given subtriggers.
*/
@SafeVarargs
- public static <W extends BoundedWindow> OnceTrigger<W> of(
- OnceTrigger<W>... triggers) {
- return new AfterFirst<W>(Arrays.<Trigger<W>>asList(triggers));
+ public static <W extends BoundedWindow> OnceTrigger of(
+ OnceTrigger... triggers) {
+ return new AfterFirst(Arrays.<Trigger>asList(triggers));
}
@Override
public void onElement(OnElementContext c) throws Exception {
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
subTrigger.invokeOnElement(c);
}
}
@Override
public void onMerge(OnMergeContext c) throws Exception {
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
subTrigger.invokeOnMerge(c);
}
updateFinishedStatus(c);
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// This trigger will fire after the earliest of its sub-triggers.
Instant deadline = BoundedWindow.TIMESTAMP_MAX_VALUE;
- for (Trigger<W> subTrigger : subTriggers) {
+ for (Trigger subTrigger : subTriggers) {
Instant subDeadline = subTrigger.getWatermarkThatGuaranteesFiring(window);
if (deadline.isAfter(subDeadline)) {
deadline = subDeadline;
@@ -80,13 +77,13 @@ public class AfterFirst<W extends BoundedWindow> extends OnceTrigger<W> {
}
@Override
- public OnceTrigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
- return new AfterFirst<W>(continuationTriggers);
+ public OnceTrigger getContinuationTrigger(List<Trigger> continuationTriggers) {
+ return new AfterFirst(continuationTriggers);
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
- for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
+ for (ExecutableTrigger subtrigger : context.trigger().subTriggers()) {
if (context.forTrigger(subtrigger).trigger().isFinished()
|| subtrigger.invokeShouldFire(context)) {
return true;
@@ -97,7 +94,7 @@ public class AfterFirst<W extends BoundedWindow> extends OnceTrigger<W> {
@Override
protected void onOnlyFiring(TriggerContext context) throws Exception {
- for (ExecutableTrigger<W> subtrigger : context.trigger().subTriggers()) {
+ for (ExecutableTrigger 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.
@@ -112,7 +109,7 @@ public class AfterFirst<W extends BoundedWindow> extends OnceTrigger<W> {
private void updateFinishedStatus(TriggerContext c) {
boolean anyFinished = false;
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
anyFinished |= c.forTrigger(subTrigger).trigger().isFinished();
}
c.trigger().setFinished(anyFinished);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterPane.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterPane.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterPane.java
index 94f43ac..76136f7 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterPane.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterPane.java
@@ -35,12 +35,9 @@ import java.util.Objects;
/**
* {@link Trigger}s that fire based on properties of the elements in the current pane.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@link Trigger}
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterPane<W extends BoundedWindow> extends OnceTrigger<W>{
+public class AfterPane extends OnceTrigger {
private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Long>>
ELEMENTS_IN_PANE_TAG =
@@ -57,8 +54,8 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
/**
* Creates a trigger that fires when the pane contains at least {@code countElems} elements.
*/
- public static <W extends BoundedWindow> AfterPane<W> elementCountAtLeast(int countElems) {
- return new AfterPane<>(countElems);
+ public static AfterPane elementCountAtLeast(int countElems) {
+ return new AfterPane(countElems);
}
@Override
@@ -67,7 +64,7 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
}
@Override
- public void prefetchOnMerge(MergingStateAccessor<?, W> state) {
+ public void prefetchOnMerge(MergingStateAccessor<?, ?> state) {
super.prefetchOnMerge(state);
StateMerging.prefetchCombiningValues(state, ELEMENTS_IN_PANE_TAG);
}
@@ -92,7 +89,7 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
long count = context.state().access(ELEMENTS_IN_PANE_TAG).read();
return count >= countElems;
}
@@ -103,17 +100,17 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
}
@Override
- public boolean isCompatible(Trigger<?> other) {
+ public boolean isCompatible(Trigger other) {
return this.equals(other);
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- public OnceTrigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ public OnceTrigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return AfterPane.elementCountAtLeast(1);
}
@@ -130,7 +127,7 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
if (!(obj instanceof AfterPane)) {
return false;
}
- AfterPane<?> that = (AfterPane<?>) obj;
+ AfterPane that = (AfterPane) obj;
return this.countElems == that.countElems;
}
@@ -140,7 +137,7 @@ private static final StateTag<Object, AccumulatorCombiningState<Long, long[], Lo
}
@Override
- protected void onOnlyFiring(Trigger<W>.TriggerContext context) throws Exception {
+ protected void onOnlyFiring(Trigger.TriggerContext context) throws Exception {
clear(context);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTime.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTime.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTime.java
index 6af6001..7a82d9c 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTime.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterProcessingTime.java
@@ -34,15 +34,13 @@ import javax.annotation.Nullable;
*
* <p>The time at which to fire the timer can be adjusted via the methods in {@link TimeTrigger},
* such as {@link TimeTrigger#plusDelayOf} or {@link TimeTrigger#alignedTo}.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterProcessingTime<W extends BoundedWindow> extends AfterDelayFromFirstElement<W> {
+public class AfterProcessingTime extends AfterDelayFromFirstElement {
@Override
@Nullable
- public Instant getCurrentTime(Trigger<W>.TriggerContext context) {
+ public Instant getCurrentTime(Trigger.TriggerContext context) {
return context.currentProcessingTime();
}
@@ -54,24 +52,24 @@ public class AfterProcessingTime<W extends BoundedWindow> extends AfterDelayFrom
* Creates a trigger that fires when the current processing time passes the processing time
* at which this trigger saw the first element in a pane.
*/
- public static <W extends BoundedWindow> AfterProcessingTime<W> pastFirstElementInPane() {
- return new AfterProcessingTime<W>(IDENTITY);
+ public static <W extends BoundedWindow> AfterProcessingTime pastFirstElementInPane() {
+ return new AfterProcessingTime(IDENTITY);
}
@Override
- protected AfterProcessingTime<W> newWith(
+ protected AfterProcessingTime newWith(
List<SerializableFunction<Instant, Instant>> transforms) {
- return new AfterProcessingTime<W>(transforms);
+ return new AfterProcessingTime(transforms);
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- protected Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
- return new AfterSynchronizedProcessingTime<W>();
+ protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
+ return new AfterSynchronizedProcessingTime();
}
@Override
@@ -87,7 +85,7 @@ public class AfterProcessingTime<W extends BoundedWindow> extends AfterDelayFrom
if (!(obj instanceof AfterProcessingTime)) {
return false;
}
- AfterProcessingTime<?> that = (AfterProcessingTime<?>) obj;
+ AfterProcessingTime that = (AfterProcessingTime) obj;
return Objects.equals(this.timestampMappers, that.timestampMappers);
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
index 37d6247..4ccba04 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
@@ -28,12 +28,11 @@ import java.util.List;
import javax.annotation.Nullable;
-class AfterSynchronizedProcessingTime<W extends BoundedWindow>
- extends AfterDelayFromFirstElement<W> {
+class AfterSynchronizedProcessingTime extends AfterDelayFromFirstElement {
@Override
@Nullable
- public Instant getCurrentTime(Trigger<W>.TriggerContext context) {
+ public Instant getCurrentTime(Trigger.TriggerContext context) {
return context.currentSynchronizedProcessingTime();
}
@@ -43,12 +42,12 @@ class AfterSynchronizedProcessingTime<W extends BoundedWindow>
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- protected Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
@@ -68,7 +67,7 @@ class AfterSynchronizedProcessingTime<W extends BoundedWindow>
}
@Override
- protected AfterSynchronizedProcessingTime<W>
+ protected AfterSynchronizedProcessingTime
newWith(List<SerializableFunction<Instant, Instant>> transforms) {
// ignore transforms
return this;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermark.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermark.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermark.java
index 249a12f..9ef8186 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermark.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/AfterWatermark.java
@@ -59,11 +59,9 @@ import java.util.Objects;
* Additionaly firings before or after the watermark can be requested by calling
* {@code AfterWatermark.pastEndOfWindow.withEarlyFirings(OnceTrigger)} or
* {@code AfterWatermark.pastEndOfWindow.withEarlyFirings(OnceTrigger)}.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used.
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class AfterWatermark<W extends BoundedWindow> {
+public class AfterWatermark {
// Static factory class.
private AfterWatermark() {}
@@ -71,37 +69,37 @@ public class AfterWatermark<W extends BoundedWindow> {
/**
* Creates a trigger that fires when the watermark passes the end of the window.
*/
- public static <W extends BoundedWindow> FromEndOfWindow<W> pastEndOfWindow() {
- return new FromEndOfWindow<W>();
+ public static FromEndOfWindow pastEndOfWindow() {
+ return new FromEndOfWindow();
}
/**
* Interface for building an AfterWatermarkTrigger with early firings already filled in.
*/
- public interface AfterWatermarkEarly<W extends BoundedWindow> extends TriggerBuilder<W> {
+ public interface AfterWatermarkEarly extends TriggerBuilder {
/**
* 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.
*/
- TriggerBuilder<W> withLateFirings(OnceTrigger<W> lateTrigger);
+ TriggerBuilder withLateFirings(OnceTrigger lateTrigger);
}
/**
* Interface for building an AfterWatermarkTrigger with late firings already filled in.
*/
- public interface AfterWatermarkLate<W extends BoundedWindow> extends TriggerBuilder<W> {
+ public interface AfterWatermarkLate extends TriggerBuilder {
/**
* 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.
*/
- TriggerBuilder<W> withEarlyFirings(OnceTrigger<W> earlyTrigger);
+ TriggerBuilder withEarlyFirings(OnceTrigger earlyTrigger);
}
/**
* A trigger which never fires. Used for the "early" trigger when only a late trigger was
* specified.
*/
- private static class NeverTrigger<W extends BoundedWindow> extends OnceTrigger<W> {
+ private static class NeverTrigger extends OnceTrigger {
protected NeverTrigger() {
super(null);
@@ -114,54 +112,54 @@ public class AfterWatermark<W extends BoundedWindow> {
public void onMerge(OnMergeContext c) throws Exception { }
@Override
- protected Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return BoundedWindow.TIMESTAMP_MAX_VALUE;
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return false;
}
@Override
- protected void onOnlyFiring(Trigger<W>.TriggerContext context) throws Exception {
+ protected void onOnlyFiring(Trigger.TriggerContext context) throws Exception {
throw new UnsupportedOperationException(
String.format("%s should never fire", getClass().getSimpleName()));
}
}
- private static class AfterWatermarkEarlyAndLate<W extends BoundedWindow>
- extends Trigger<W>
- implements TriggerBuilder<W>, AfterWatermarkEarly<W>, AfterWatermarkLate<W> {
+ private static class AfterWatermarkEarlyAndLate
+ extends Trigger
+ implements TriggerBuilder, AfterWatermarkEarly, AfterWatermarkLate {
private static final int EARLY_INDEX = 0;
private static final int LATE_INDEX = 1;
- private final OnceTrigger<W> earlyTrigger;
- private final OnceTrigger<W> lateTrigger;
+ private final OnceTrigger earlyTrigger;
+ private final OnceTrigger lateTrigger;
@SuppressWarnings("unchecked")
- private AfterWatermarkEarlyAndLate(OnceTrigger<W> earlyTrigger, OnceTrigger<W> lateTrigger) {
+ private AfterWatermarkEarlyAndLate(OnceTrigger earlyTrigger, OnceTrigger lateTrigger) {
super(lateTrigger == null
- ? ImmutableList.<Trigger<W>>of(earlyTrigger)
- : ImmutableList.<Trigger<W>>of(earlyTrigger, lateTrigger));
+ ? ImmutableList.<Trigger>of(earlyTrigger)
+ : ImmutableList.<Trigger>of(earlyTrigger, lateTrigger));
this.earlyTrigger = checkNotNull(earlyTrigger, "earlyTrigger should not be null");
this.lateTrigger = lateTrigger;
}
@Override
- public TriggerBuilder<W> withEarlyFirings(OnceTrigger<W> earlyTrigger) {
- return new AfterWatermarkEarlyAndLate<W>(earlyTrigger, lateTrigger);
+ public TriggerBuilder withEarlyFirings(OnceTrigger earlyTrigger) {
+ return new AfterWatermarkEarlyAndLate(earlyTrigger, lateTrigger);
}
@Override
- public TriggerBuilder<W> withLateFirings(OnceTrigger<W> lateTrigger) {
- return new AfterWatermarkEarlyAndLate<W>(earlyTrigger, lateTrigger);
+ public TriggerBuilder withLateFirings(OnceTrigger lateTrigger) {
+ return new AfterWatermarkEarlyAndLate(earlyTrigger, lateTrigger);
}
@Override
@@ -172,7 +170,7 @@ public class AfterWatermark<W extends BoundedWindow> {
} else {
// If merges can happen, we run for all subtriggers because they might be
// de-activated or re-activated
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
subTrigger.invokeOnElement(c);
}
}
@@ -183,7 +181,7 @@ public class AfterWatermark<W extends BoundedWindow> {
// NOTE that the ReduceFnRunner will delete all end-of-window timers for the
// merged-away windows.
- ExecutableTrigger<W> earlySubtrigger = c.trigger().subTrigger(EARLY_INDEX);
+ ExecutableTrigger earlySubtrigger = c.trigger().subTrigger(EARLY_INDEX);
// We check the early trigger to determine if we are still processing it or
// if the end of window has transitioned us to the late trigger
OnMergeContext earlyContext = c.forTrigger(earlySubtrigger);
@@ -194,7 +192,7 @@ public class AfterWatermark<W extends BoundedWindow> {
if (!earlyContext.trigger().finishedInAllMergingWindows() || !endOfWindowReached(c)) {
earlyContext.trigger().setFinished(false);
if (lateTrigger != null) {
- ExecutableTrigger<W> lateSubtrigger = c.trigger().subTrigger(LATE_INDEX);
+ ExecutableTrigger lateSubtrigger = c.trigger().subTrigger(LATE_INDEX);
OnMergeContext lateContext = c.forTrigger(lateSubtrigger);
lateContext.trigger().setFinished(false);
lateSubtrigger.invokeClear(lateContext);
@@ -209,31 +207,31 @@ public class AfterWatermark<W extends BoundedWindow> {
}
@Override
- public Trigger<W> getContinuationTrigger() {
- return new AfterWatermarkEarlyAndLate<W>(
+ public Trigger getContinuationTrigger() {
+ return new AfterWatermarkEarlyAndLate(
earlyTrigger.getContinuationTrigger(),
lateTrigger == null ? null : lateTrigger.getContinuationTrigger());
}
@Override
- protected Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
throw new UnsupportedOperationException(
- "Should not call getContinuationTrigger(List<Trigger<W>>)");
+ "Should not call getContinuationTrigger(List<Trigger>)");
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// Even without an early or late trigger, we'll still produce a firing at the watermark.
return window.maxTimestamp();
}
- private boolean endOfWindowReached(Trigger<W>.TriggerContext context) {
+ private boolean endOfWindowReached(Trigger.TriggerContext context) {
return context.currentEventTime() != null
&& context.currentEventTime().isAfter(context.window().maxTimestamp());
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
if (!context.trigger().isFinished(EARLY_INDEX)) {
// We have not yet transitioned to late firings.
// We should fire if either the trigger is ready or we reach the end of the window.
@@ -248,7 +246,7 @@ public class AfterWatermark<W extends BoundedWindow> {
}
@Override
- public void onFire(Trigger<W>.TriggerContext context) throws Exception {
+ public void onFire(Trigger.TriggerContext context) throws Exception {
if (!context.forTrigger(context.trigger().subTrigger(EARLY_INDEX)).trigger().isFinished()) {
onNonLateFiring(context);
} else if (lateTrigger != null) {
@@ -259,10 +257,10 @@ public class AfterWatermark<W extends BoundedWindow> {
}
}
- private void onNonLateFiring(Trigger<W>.TriggerContext context) throws Exception {
+ private void onNonLateFiring(Trigger.TriggerContext context) throws Exception {
// We have not yet transitioned to late firings.
- ExecutableTrigger<W> earlySubtrigger = context.trigger().subTrigger(EARLY_INDEX);
- Trigger<W>.TriggerContext earlyContext = context.forTrigger(earlySubtrigger);
+ ExecutableTrigger earlySubtrigger = context.trigger().subTrigger(EARLY_INDEX);
+ Trigger.TriggerContext earlyContext = context.forTrigger(earlySubtrigger);
if (!endOfWindowReached(context)) {
// This is an early firing, since we have not arrived at the end of the window
@@ -291,9 +289,9 @@ public class AfterWatermark<W extends BoundedWindow> {
}
- private void onLateFiring(Trigger<W>.TriggerContext context) throws Exception {
+ private void onLateFiring(Trigger.TriggerContext context) throws Exception {
// We are firing the late trigger, with implicit repeat
- ExecutableTrigger<W> lateSubtrigger = context.trigger().subTrigger(LATE_INDEX);
+ ExecutableTrigger lateSubtrigger = context.trigger().subTrigger(LATE_INDEX);
lateSubtrigger.invokeOnFire(context);
// It is a OnceTrigger, so it must have finished; unfinished it and clear it
lateSubtrigger.invokeClear(context);
@@ -304,7 +302,7 @@ public class AfterWatermark<W extends BoundedWindow> {
/**
* A watermark trigger targeted relative to the end of the window.
*/
- public static class FromEndOfWindow<W extends BoundedWindow> extends OnceTrigger<W> {
+ public static class FromEndOfWindow extends OnceTrigger {
private FromEndOfWindow() {
super(null);
@@ -314,20 +312,20 @@ public class AfterWatermark<W extends BoundedWindow> {
* 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 AfterWatermarkEarly<W> withEarlyFirings(OnceTrigger<W> earlyFirings) {
+ public AfterWatermarkEarly withEarlyFirings(OnceTrigger earlyFirings) {
Preconditions.checkNotNull(earlyFirings,
"Must specify the trigger to use for early firings");
- return new AfterWatermarkEarlyAndLate<W>(earlyFirings, null);
+ return new AfterWatermarkEarlyAndLate(earlyFirings, null);
}
/**
* 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 AfterWatermarkLate<W> withLateFirings(OnceTrigger<W> lateFirings) {
+ public AfterWatermarkLate withLateFirings(OnceTrigger lateFirings) {
Preconditions.checkNotNull(lateFirings,
"Must specify the trigger to use for late firings");
- return new AfterWatermarkEarlyAndLate<W>(new NeverTrigger<W>(), lateFirings);
+ return new AfterWatermarkEarlyAndLate(new NeverTrigger(), lateFirings);
}
@Override
@@ -358,12 +356,12 @@ public class AfterWatermark<W extends BoundedWindow> {
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return window.maxTimestamp();
}
@Override
- public FromEndOfWindow<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ public FromEndOfWindow getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
@@ -383,16 +381,16 @@ public class AfterWatermark<W extends BoundedWindow> {
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return endOfWindowReached(context);
}
- private boolean endOfWindowReached(Trigger<W>.TriggerContext context) {
+ private boolean endOfWindowReached(Trigger.TriggerContext context) {
return context.currentEventTime() != null
&& context.currentEventTime().isAfter(context.window().maxTimestamp());
}
@Override
- protected void onOnlyFiring(Trigger<W>.TriggerContext context) throws Exception { }
+ protected void onOnlyFiring(Trigger.TriggerContext context) throws Exception { }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/DefaultTrigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/DefaultTrigger.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/DefaultTrigger.java
index ccb3a87..0e6b5ce 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/DefaultTrigger.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/DefaultTrigger.java
@@ -27,11 +27,9 @@ import java.util.List;
/**
* A trigger that is equivalent to {@code Repeatedly.forever(AfterWatermark.pastEndOfWindow())}.
* See {@link Repeatedly#forever} and {@link AfterWatermark#pastEndOfWindow} for more details.
- *
- * @param <W> The type of windows being triggered/encoded.
*/
@Experimental(Experimental.Kind.TRIGGER)
-public class DefaultTrigger<W extends BoundedWindow> extends Trigger<W>{
+public class DefaultTrigger extends Trigger{
private DefaultTrigger() {
super(null);
@@ -40,8 +38,8 @@ public class DefaultTrigger<W extends BoundedWindow> extends Trigger<W>{
/**
* Returns the default trigger.
*/
- public static <W extends BoundedWindow> DefaultTrigger<W> of() {
- return new DefaultTrigger<W>();
+ public static <W extends BoundedWindow> DefaultTrigger of() {
+ return new DefaultTrigger();
}
@Override
@@ -66,31 +64,31 @@ public class DefaultTrigger<W extends BoundedWindow> extends Trigger<W>{
public void clear(TriggerContext c) throws Exception { }
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
return window.maxTimestamp();
}
@Override
- public boolean isCompatible(Trigger<?> other) {
+ public boolean isCompatible(Trigger other) {
// Semantically, all default triggers are identical
return other instanceof DefaultTrigger;
}
@Override
- public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ public Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return endOfWindowReached(context);
}
- private boolean endOfWindowReached(Trigger<W>.TriggerContext context) {
+ private boolean endOfWindowReached(Trigger.TriggerContext context) {
return context.currentEventTime() != null
&& context.currentEventTime().isAfter(context.window().maxTimestamp());
}
@Override
- public void onFire(Trigger<W>.TriggerContext context) throws Exception { }
+ public void onFire(Trigger.TriggerContext context) throws Exception { }
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java
index b8abf0f..68b9083 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/OrFinallyTrigger.java
@@ -28,12 +28,12 @@ import java.util.List;
/**
* Executes the {@code actual} trigger until it finishes or until the {@code until} trigger fires.
*/
-class OrFinallyTrigger<W extends BoundedWindow> extends Trigger<W> {
+class OrFinallyTrigger extends Trigger {
private static final int ACTUAL = 0;
private static final int UNTIL = 1;
- @VisibleForTesting OrFinallyTrigger(Trigger<W> actual, Trigger.OnceTrigger<W> until) {
+ @VisibleForTesting OrFinallyTrigger(Trigger actual, Trigger.OnceTrigger until) {
super(Arrays.asList(actual, until));
}
@@ -45,14 +45,14 @@ class OrFinallyTrigger<W extends BoundedWindow> extends Trigger<W> {
@Override
public void onMerge(OnMergeContext c) throws Exception {
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
subTrigger.invokeOnMerge(c);
}
updateFinishedState(c);
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// This trigger fires once either the trigger or the until trigger fires.
Instant actualDeadline = subTriggers.get(ACTUAL).getWatermarkThatGuaranteesFiring(window);
Instant untilDeadline = subTriggers.get(UNTIL).getWatermarkThatGuaranteesFiring(window);
@@ -60,25 +60,25 @@ class OrFinallyTrigger<W extends BoundedWindow> extends Trigger<W> {
}
@Override
- public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ public Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
// Use OrFinallyTrigger instead of AfterFirst because the continuation of ACTUAL
// may not be a OnceTrigger.
return Repeatedly.forever(
- new OrFinallyTrigger<W>(
+ new OrFinallyTrigger(
continuationTriggers.get(ACTUAL),
- (Trigger.OnceTrigger<W>) continuationTriggers.get(UNTIL)));
+ (Trigger.OnceTrigger) continuationTriggers.get(UNTIL)));
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return context.trigger().subTrigger(ACTUAL).invokeShouldFire(context)
|| context.trigger().subTrigger(UNTIL).invokeShouldFire(context);
}
@Override
- public void onFire(Trigger<W>.TriggerContext context) throws Exception {
- ExecutableTrigger<W> actualSubtrigger = context.trigger().subTrigger(ACTUAL);
- ExecutableTrigger<W> untilSubtrigger = context.trigger().subTrigger(UNTIL);
+ public void onFire(Trigger.TriggerContext context) throws Exception {
+ ExecutableTrigger actualSubtrigger = context.trigger().subTrigger(ACTUAL);
+ ExecutableTrigger untilSubtrigger = context.trigger().subTrigger(UNTIL);
if (untilSubtrigger.invokeShouldFire(context)) {
untilSubtrigger.invokeOnFire(context);
@@ -94,7 +94,7 @@ class OrFinallyTrigger<W extends BoundedWindow> extends Trigger<W> {
private void updateFinishedState(TriggerContext c) throws Exception {
boolean anyStillFinished = false;
- for (ExecutableTrigger<W> subTrigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger subTrigger : c.trigger().subTriggers()) {
anyStillFinished |= c.forTrigger(subTrigger).trigger().isFinished();
}
c.trigger().setFinished(anyStillFinished);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
index 9be0259..988f0a4 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Repeatedly.java
@@ -34,11 +34,8 @@ import java.util.List;
*
* <p>{@code Repeatedly.forever(someTrigger)} behaves like an infinite
* {@code AfterEach.inOrder(someTrigger, someTrigger, someTrigger, ...)}.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code Trigger}
*/
-public class Repeatedly<W extends BoundedWindow> extends Trigger<W> {
+public class Repeatedly extends Trigger {
private static final int REPEATED = 0;
@@ -50,11 +47,11 @@ public class Repeatedly<W extends BoundedWindow> extends Trigger<W> {
*
* @param repeated the trigger to execute repeatedly.
*/
- public static <W extends BoundedWindow> Repeatedly<W> forever(Trigger<W> repeated) {
- return new Repeatedly<W>(repeated);
+ public static <W extends BoundedWindow> Repeatedly forever(Trigger repeated) {
+ return new Repeatedly(repeated);
}
- private Repeatedly(Trigger<W> repeated) {
+ private Repeatedly(Trigger repeated) {
super(Arrays.asList(repeated));
}
@@ -70,18 +67,18 @@ public class Repeatedly<W extends BoundedWindow> extends Trigger<W> {
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
// This trigger fires once the repeated trigger fires.
return subTriggers.get(REPEATED).getWatermarkThatGuaranteesFiring(window);
}
@Override
- public Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
- return new Repeatedly<W>(continuationTriggers.get(REPEATED));
+ public Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
+ return new Repeatedly(continuationTriggers.get(REPEATED));
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return getRepeated(context).invokeShouldFire(context);
}
@@ -95,7 +92,7 @@ public class Repeatedly<W extends BoundedWindow> extends Trigger<W> {
}
}
- private ExecutableTrigger<W> getRepeated(TriggerContext context) {
+ private ExecutableTrigger getRepeated(TriggerContext context) {
return context.trigger().subTrigger(REPEATED);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Trigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Trigger.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Trigger.java
index 8b1a8bc..fde8ca4 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Trigger.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Trigger.java
@@ -90,18 +90,15 @@ import javax.annotation.Nullable;
* <p>Triggers should not build up any state internally since they may be recreated
* between invocations of the callbacks. All important values should be persisted using
* state before the callback returns.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code Trigger}
*/
@Experimental(Experimental.Kind.TRIGGER)
-public abstract class Trigger<W extends BoundedWindow> implements Serializable, TriggerBuilder<W> {
+public abstract class Trigger implements Serializable, TriggerBuilder {
/**
* Interface for accessing information about the trigger being executed and other triggers in the
* same tree.
*/
- public interface TriggerInfo<W extends BoundedWindow> {
+ public interface TriggerInfo {
/**
* Returns true if the windowing strategy of the current {@code PCollection} is a merging
@@ -114,12 +111,12 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
/**
* Access the executable versions of the sub-triggers of the current trigger.
*/
- Iterable<ExecutableTrigger<W>> subTriggers();
+ Iterable<ExecutableTrigger> subTriggers();
/**
* Access the executable version of the specified sub-trigger.
*/
- ExecutableTrigger<W> subTrigger(int subtriggerIndex);
+ ExecutableTrigger subTrigger(int subtriggerIndex);
/**
* Returns true if the current trigger is marked finished.
@@ -139,12 +136,12 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
/**
* Returns an iterable over the unfinished sub-triggers of the current trigger.
*/
- Iterable<ExecutableTrigger<W>> unfinishedSubTriggers();
+ Iterable<ExecutableTrigger> unfinishedSubTriggers();
/**
* Returns the first unfinished sub-trigger.
*/
- ExecutableTrigger<W> firstUnfinishedSubTrigger();
+ ExecutableTrigger firstUnfinishedSubTrigger();
/**
* Clears all keyed state for triggers in the current sub-tree and unsets all the associated
@@ -167,7 +164,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* Interact with properties of the trigger being executed, with extensions to deal with the
* merging windows.
*/
- public interface MergingTriggerInfo<W extends BoundedWindow> extends TriggerInfo<W> {
+ public interface MergingTriggerInfo extends TriggerInfo {
/** Return true if the trigger is finished in any window being merged. */
public abstract boolean finishedInAnyMergingWindow();
@@ -176,7 +173,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
public abstract boolean finishedInAllMergingWindows();
/** Return the merging windows in which the trigger is finished. */
- public abstract Iterable<W> getFinishedMergingWindows();
+ public abstract Iterable<? extends BoundedWindow> getFinishedMergingWindows();
}
/**
@@ -188,16 +185,16 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
public abstract class TriggerContext {
/** Returns the interface for accessing trigger info. */
- public abstract TriggerInfo<W> trigger();
+ public abstract TriggerInfo trigger();
/** Returns the interface for accessing persistent state. */
public abstract StateAccessor<?> state();
/** The window that the current context is executing in. */
- public abstract W window();
+ public abstract BoundedWindow window();
/** Create a sub-context for the given sub-trigger. */
- public abstract TriggerContext forTrigger(ExecutableTrigger<W> trigger);
+ public abstract TriggerContext forTrigger(ExecutableTrigger trigger);
/**
* Removes the timer set in this trigger context for the given {@link Instant}
@@ -240,7 +237,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
/** Create an {@code OnElementContext} for executing the given trigger. */
@Override
- public abstract OnElementContext forTrigger(ExecutableTrigger<W> trigger);
+ public abstract OnElementContext forTrigger(ExecutableTrigger trigger);
}
/**
@@ -263,19 +260,19 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
/** Create an {@code OnMergeContext} for executing the given trigger. */
@Override
- public abstract OnMergeContext forTrigger(ExecutableTrigger<W> trigger);
+ public abstract OnMergeContext forTrigger(ExecutableTrigger trigger);
@Override
- public abstract MergingStateAccessor<?, W> state();
+ public abstract MergingStateAccessor<?, ?> state();
@Override
- public abstract MergingTriggerInfo<W> trigger();
+ public abstract MergingTriggerInfo trigger();
}
@Nullable
- protected final List<Trigger<W>> subTriggers;
+ protected final List<Trigger> subTriggers;
- protected Trigger(@Nullable List<Trigger<W>> subTriggers) {
+ protected Trigger(@Nullable List<Trigger> subTriggers) {
this.subTriggers = subTriggers;
}
@@ -322,7 +319,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
*/
public void prefetchOnElement(StateAccessor<?> state) {
if (subTriggers != null) {
- for (Trigger<W> trigger : subTriggers) {
+ for (Trigger trigger : subTriggers) {
trigger.prefetchOnElement(state);
}
}
@@ -332,9 +329,9 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* Called to allow the trigger to prefetch any state it will likely need to read from during
* an {@link #onMerge} call.
*/
- public void prefetchOnMerge(MergingStateAccessor<?, W> state) {
+ public void prefetchOnMerge(MergingStateAccessor<?, ?> state) {
if (subTriggers != null) {
- for (Trigger<W> trigger : subTriggers) {
+ for (Trigger trigger : subTriggers) {
trigger.prefetchOnMerge(state);
}
}
@@ -346,7 +343,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
*/
public void prefetchShouldFire(StateAccessor<?> state) {
if (subTriggers != null) {
- for (Trigger<W> trigger : subTriggers) {
+ for (Trigger trigger : subTriggers) {
trigger.prefetchShouldFire(state);
}
}
@@ -358,7 +355,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
*/
public void prefetchOnFire(StateAccessor<?> state) {
if (subTriggers != null) {
- for (Trigger<W> trigger : subTriggers) {
+ for (Trigger trigger : subTriggers) {
trigger.prefetchOnFire(state);
}
}
@@ -373,13 +370,13 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
*/
public void clear(TriggerContext c) throws Exception {
if (subTriggers != null) {
- for (ExecutableTrigger<W> trigger : c.trigger().subTriggers()) {
+ for (ExecutableTrigger trigger : c.trigger().subTriggers()) {
trigger.invokeClear(c);
}
}
}
- public Iterable<Trigger<W>> subTriggers() {
+ public Iterable<Trigger> subTriggers() {
return subTriggers;
}
@@ -390,13 +387,13 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* speculative results. Triggers that fire once (or multiple times) should
* continue firing once (or multiple times).
*/
- public Trigger<W> getContinuationTrigger() {
+ public Trigger getContinuationTrigger() {
if (subTriggers == null) {
return getContinuationTrigger(null);
}
- List<Trigger<W>> subTriggerContinuations = new ArrayList<>();
- for (Trigger<W> subTrigger : subTriggers) {
+ List<Trigger> subTriggerContinuations = new ArrayList<>();
+ for (Trigger subTrigger : subTriggers) {
subTriggerContinuations.add(subTrigger.getContinuationTrigger());
}
return getContinuationTrigger(subTriggerContinuations);
@@ -406,7 +403,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* Return the {@link #getContinuationTrigger} of this {@code Trigger}. For convenience, this
* is provided the continuation trigger of each of the sub-triggers.
*/
- protected abstract Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers);
+ protected abstract Trigger getContinuationTrigger(List<Trigger> continuationTriggers);
/**
* Returns a bound in watermark time by which this trigger would have fired at least once
@@ -419,12 +416,12 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* <p>This estimate is used to determine that there are no elements in a side-input window, which
* causes the default value to be used instead.
*/
- public abstract Instant getWatermarkThatGuaranteesFiring(W window);
+ public abstract Instant getWatermarkThatGuaranteesFiring(BoundedWindow window);
/**
* Returns whether this performs the same triggering as the given {@code Trigger}.
*/
- public boolean isCompatible(Trigger<?> other) {
+ public boolean isCompatible(Trigger other) {
if (!getClass().equals(other.getClass())) {
return false;
}
@@ -467,8 +464,7 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
if (!(obj instanceof Trigger)) {
return false;
}
- @SuppressWarnings("unchecked")
- Trigger<W> that = (Trigger<W>) obj;
+ Trigger that = (Trigger) obj;
return Objects.equals(getClass(), that.getClass())
&& Objects.equals(subTriggers, that.subTriggers);
}
@@ -497,34 +493,31 @@ public abstract class Trigger<W extends BoundedWindow> implements Serializable,
* <p>Note that if {@code t1} is {@link OnceTrigger}, then {@code t1.orFinally(t2)} is the same
* as {@code AfterFirst.of(t1, t2)}.
*/
- public Trigger<W> orFinally(OnceTrigger<W> until) {
- return new OrFinallyTrigger<W>(this, until);
+ public Trigger orFinally(OnceTrigger until) {
+ return new OrFinallyTrigger(this, until);
}
@Override
- public Trigger<W> buildTrigger() {
+ public Trigger buildTrigger() {
return this;
}
/**
* {@link Trigger}s that are guaranteed to fire at most once should extend from this, rather
* than the general {@link Trigger} class to indicate that behavior.
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used by this
- * {@code AtMostOnceTrigger}
*/
- public abstract static class OnceTrigger<W extends BoundedWindow> extends Trigger<W> {
- protected OnceTrigger(List<Trigger<W>> subTriggers) {
+ public abstract static class OnceTrigger extends Trigger {
+ protected OnceTrigger(List<Trigger> subTriggers) {
super(subTriggers);
}
@Override
- public final OnceTrigger<W> getContinuationTrigger() {
- Trigger<W> continuation = super.getContinuationTrigger();
+ public final OnceTrigger getContinuationTrigger() {
+ Trigger continuation = super.getContinuationTrigger();
if (!(continuation instanceof OnceTrigger)) {
throw new IllegalStateException("Continuation of a OnceTrigger must be a OnceTrigger");
}
- return (OnceTrigger<W>) continuation;
+ return (OnceTrigger) continuation;
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerBuilder.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerBuilder.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerBuilder.java
index dcc042c..017fd87 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerBuilder.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/TriggerBuilder.java
@@ -22,10 +22,8 @@ package com.google.cloud.dataflow.sdk.transforms.windowing;
*
* <p>This includes {@code Trigger}s (which can return themselves) and any "enhanced" syntax for
* constructing a trigger.
- *
- * @param <W> The type of windows the built trigger will operate on.
*/
-public interface TriggerBuilder<W extends BoundedWindow> {
+public interface TriggerBuilder {
/** Return the {@code Trigger} built by this builder. */
- Trigger<W> buildTrigger();
+ Trigger buildTrigger();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Window.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Window.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Window.java
index 20b3ed5..aef61b2 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Window.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/transforms/windowing/Window.java
@@ -194,7 +194,7 @@ public class Window {
* mode using either {@link #discardingFiredPanes()} or {@link #accumulatingFiredPanes()}.
*/
@Experimental(Kind.TRIGGER)
- public static <T> Bound<T> triggering(TriggerBuilder<?> trigger) {
+ public static <T> Bound<T> triggering(TriggerBuilder trigger) {
return new Unbound().triggering(trigger);
}
@@ -289,7 +289,7 @@ public class Window {
* mode using either {@link #discardingFiredPanes()} or {@link #accumulatingFiredPanes()}.
*/
@Experimental(Kind.TRIGGER)
- public <T> Bound<T> triggering(TriggerBuilder<?> trigger) {
+ public <T> Bound<T> triggering(TriggerBuilder trigger) {
return new Bound<T>(name).triggering(trigger);
}
@@ -361,16 +361,20 @@ public class Window {
@Nullable private final WindowFn<? super T, ?> windowFn;
- @Nullable private final Trigger<?> trigger;
+ @Nullable private final Trigger trigger;
@Nullable private final AccumulationMode mode;
@Nullable private final Duration allowedLateness;
@Nullable private final ClosingBehavior closingBehavior;
@Nullable private final OutputTimeFn<?> outputTimeFn;
- private Bound(String name,
- @Nullable WindowFn<? super T, ?> windowFn, @Nullable Trigger<?> trigger,
- @Nullable AccumulationMode mode, @Nullable Duration allowedLateness,
- ClosingBehavior behavior, @Nullable OutputTimeFn<?> outputTimeFn) {
+ private Bound(
+ String name,
+ @Nullable WindowFn<? super T, ?> windowFn,
+ @Nullable Trigger trigger,
+ @Nullable AccumulationMode mode,
+ @Nullable Duration allowedLateness,
+ ClosingBehavior behavior,
+ @Nullable OutputTimeFn<?> outputTimeFn) {
super(name);
this.windowFn = windowFn;
this.trigger = trigger;
@@ -428,7 +432,7 @@ public class Window {
* mode using either {@link #discardingFiredPanes()} or {@link #accumulatingFiredPanes()}.
*/
@Experimental(Kind.TRIGGER)
- public Bound<T> triggering(TriggerBuilder<?> trigger) {
+ public Bound<T> triggering(TriggerBuilder trigger) {
return new Bound<T>(
name,
windowFn,
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ExecutableTrigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ExecutableTrigger.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ExecutableTrigger.java
index 25a630a..71c8237 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ExecutableTrigger.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ExecutableTrigger.java
@@ -30,42 +30,40 @@ import java.util.List;
* A wrapper around a trigger used during execution. While an actual trigger may appear multiple
* times (both in the same trigger expression and in other trigger expressions), the
* {@code ExecutableTrigger} wrapped around them forms a tree (only one occurrence).
- *
- * @param <W> {@link BoundedWindow} subclass used to represent the windows used.
*/
-public class ExecutableTrigger<W extends BoundedWindow> implements Serializable {
+public class ExecutableTrigger implements Serializable {
/** Store the index assigned to this trigger. */
private final int triggerIndex;
private final int firstIndexAfterSubtree;
- private final List<ExecutableTrigger<W>> subTriggers = new ArrayList<>();
- private final Trigger<W> trigger;
+ private final List<ExecutableTrigger> subTriggers = new ArrayList<>();
+ private final Trigger trigger;
- public static <W extends BoundedWindow> ExecutableTrigger<W> create(Trigger<W> trigger) {
+ public static <W extends BoundedWindow> ExecutableTrigger create(Trigger trigger) {
return create(trigger, 0);
}
- private static <W extends BoundedWindow> ExecutableTrigger<W> create(
- Trigger<W> trigger, int nextUnusedIndex) {
+ private static <W extends BoundedWindow> ExecutableTrigger create(
+ Trigger trigger, int nextUnusedIndex) {
if (trigger instanceof OnceTrigger) {
- return new ExecutableOnceTrigger<W>((OnceTrigger<W>) trigger, nextUnusedIndex);
+ return new ExecutableOnceTrigger((OnceTrigger) trigger, nextUnusedIndex);
} else {
- return new ExecutableTrigger<W>(trigger, nextUnusedIndex);
+ return new ExecutableTrigger(trigger, nextUnusedIndex);
}
}
- public static <W extends BoundedWindow> ExecutableTrigger<W> createForOnceTrigger(
- OnceTrigger<W> trigger, int nextUnusedIndex) {
- return new ExecutableOnceTrigger<W>(trigger, nextUnusedIndex);
+ public static <W extends BoundedWindow> ExecutableTrigger createForOnceTrigger(
+ OnceTrigger trigger, int nextUnusedIndex) {
+ return new ExecutableOnceTrigger(trigger, nextUnusedIndex);
}
- private ExecutableTrigger(Trigger<W> trigger, int nextUnusedIndex) {
+ private ExecutableTrigger(Trigger trigger, int nextUnusedIndex) {
this.trigger = Preconditions.checkNotNull(trigger, "trigger must not be null");
this.triggerIndex = nextUnusedIndex++;
if (trigger.subTriggers() != null) {
- for (Trigger<W> subTrigger : trigger.subTriggers()) {
- ExecutableTrigger<W> subExecutable = create(subTrigger, nextUnusedIndex);
+ for (Trigger subTrigger : trigger.subTriggers()) {
+ ExecutableTrigger subExecutable = create(subTrigger, nextUnusedIndex);
subTriggers.add(subExecutable);
nextUnusedIndex = subExecutable.firstIndexAfterSubtree;
}
@@ -73,7 +71,7 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
firstIndexAfterSubtree = nextUnusedIndex;
}
- public List<ExecutableTrigger<W>> subTriggers() {
+ public List<ExecutableTrigger> subTriggers() {
return subTriggers;
}
@@ -85,7 +83,7 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
/**
* Return the underlying trigger specification corresponding to this {@code ExecutableTrigger}.
*/
- public Trigger<W> getSpec() {
+ public Trigger getSpec() {
return trigger;
}
@@ -97,16 +95,16 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
return firstIndexAfterSubtree;
}
- public boolean isCompatible(ExecutableTrigger<W> other) {
+ public boolean isCompatible(ExecutableTrigger other) {
return trigger.isCompatible(other.trigger);
}
- public ExecutableTrigger<W> getSubTriggerContaining(int index) {
+ public ExecutableTrigger getSubTriggerContaining(int index) {
Preconditions.checkNotNull(subTriggers);
Preconditions.checkState(index > triggerIndex && index < firstIndexAfterSubtree,
"Cannot find sub-trigger containing index not in this tree.");
- ExecutableTrigger<W> previous = null;
- for (ExecutableTrigger<W> subTrigger : subTriggers) {
+ ExecutableTrigger previous = null;
+ for (ExecutableTrigger subTrigger : subTriggers) {
if (index < subTrigger.triggerIndex) {
return previous;
}
@@ -119,7 +117,7 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
* Invoke the {@link Trigger#onElement} method for this trigger, ensuring that the bits are
* properly updated if the trigger finishes.
*/
- public void invokeOnElement(Trigger<W>.OnElementContext c) throws Exception {
+ public void invokeOnElement(Trigger.OnElementContext c) throws Exception {
trigger.onElement(c.forTrigger(this));
}
@@ -127,23 +125,23 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
* Invoke the {@link Trigger#onMerge} method for this trigger, ensuring that the bits are properly
* updated.
*/
- public void invokeOnMerge(Trigger<W>.OnMergeContext c) throws Exception {
- Trigger<W>.OnMergeContext subContext = c.forTrigger(this);
+ public void invokeOnMerge(Trigger.OnMergeContext c) throws Exception {
+ Trigger.OnMergeContext subContext = c.forTrigger(this);
trigger.onMerge(subContext);
}
- public boolean invokeShouldFire(Trigger<W>.TriggerContext c) throws Exception {
+ public boolean invokeShouldFire(Trigger.TriggerContext c) throws Exception {
return trigger.shouldFire(c.forTrigger(this));
}
- public void invokeOnFire(Trigger<W>.TriggerContext c) throws Exception {
+ public void invokeOnFire(Trigger.TriggerContext c) throws Exception {
trigger.onFire(c.forTrigger(this));
}
/**
* Invoke clear for the current this trigger.
*/
- public void invokeClear(Trigger<W>.TriggerContext c) throws Exception {
+ public void invokeClear(Trigger.TriggerContext c) throws Exception {
trigger.clear(c.forTrigger(this));
}
@@ -151,9 +149,9 @@ public class ExecutableTrigger<W extends BoundedWindow> implements Serializable
* {@link ExecutableTrigger} that enforces the fact that the trigger should always FIRE_AND_FINISH
* and never just FIRE.
*/
- private static class ExecutableOnceTrigger<W extends BoundedWindow> extends ExecutableTrigger<W> {
+ private static class ExecutableOnceTrigger extends ExecutableTrigger {
- public ExecutableOnceTrigger(OnceTrigger<W> trigger, int nextUnusedIndex) {
+ public ExecutableOnceTrigger(OnceTrigger trigger, int nextUnusedIndex) {
super(trigger, nextUnusedIndex);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggers.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggers.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggers.java
index 76b54e5..e0f14de 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggers.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggers.java
@@ -25,17 +25,17 @@ public interface FinishedTriggers {
/**
* Returns {@code true} if the trigger is finished.
*/
- public boolean isFinished(ExecutableTrigger<?> trigger);
+ public boolean isFinished(ExecutableTrigger trigger);
/**
* Sets the fact that the trigger is finished.
*/
- public void setFinished(ExecutableTrigger<?> trigger, boolean value);
+ public void setFinished(ExecutableTrigger trigger, boolean value);
/**
* Sets the trigger and all of its subtriggers to unfinished.
*/
- public void clearRecursively(ExecutableTrigger<?> trigger);
+ public void clearRecursively(ExecutableTrigger trigger);
/**
* Create an independent copy of this mutable {@link FinishedTriggers}.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersBitSet.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersBitSet.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersBitSet.java
index f31e64d..69e1b84 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersBitSet.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersBitSet.java
@@ -46,17 +46,17 @@ public class FinishedTriggersBitSet implements FinishedTriggers {
}
@Override
- public boolean isFinished(ExecutableTrigger<?> trigger) {
+ public boolean isFinished(ExecutableTrigger trigger) {
return bitSet.get(trigger.getTriggerIndex());
}
@Override
- public void setFinished(ExecutableTrigger<?> trigger, boolean value) {
+ public void setFinished(ExecutableTrigger trigger, boolean value) {
bitSet.set(trigger.getTriggerIndex(), value);
}
@Override
- public void clearRecursively(ExecutableTrigger<?> trigger) {
+ public void clearRecursively(ExecutableTrigger trigger) {
bitSet.clear(trigger.getTriggerIndex(), trigger.getFirstIndexAfterSubtree());
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSet.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSet.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSet.java
index d0d8329..c6ec7a1 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSet.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/FinishedTriggersSet.java
@@ -26,30 +26,30 @@ import java.util.Set;
*/
public class FinishedTriggersSet implements FinishedTriggers {
- private final Set<ExecutableTrigger<?>> finishedTriggers;
+ private final Set<ExecutableTrigger> finishedTriggers;
- private FinishedTriggersSet(Set<ExecutableTrigger<?>> finishedTriggers) {
+ private FinishedTriggersSet(Set<ExecutableTrigger> finishedTriggers) {
this.finishedTriggers = finishedTriggers;
}
- public static FinishedTriggersSet fromSet(Set<ExecutableTrigger<?>> finishedTriggers) {
+ public static FinishedTriggersSet fromSet(Set<ExecutableTrigger> finishedTriggers) {
return new FinishedTriggersSet(finishedTriggers);
}
/**
* Returns a mutable {@link Set} of the underlying triggers that are finished.
*/
- public Set<ExecutableTrigger<?>> getFinishedTriggers() {
+ public Set<ExecutableTrigger> getFinishedTriggers() {
return finishedTriggers;
}
@Override
- public boolean isFinished(ExecutableTrigger<?> trigger) {
+ public boolean isFinished(ExecutableTrigger trigger) {
return finishedTriggers.contains(trigger);
}
@Override
- public void setFinished(ExecutableTrigger<?> trigger, boolean value) {
+ public void setFinished(ExecutableTrigger trigger, boolean value) {
if (value) {
finishedTriggers.add(trigger);
} else {
@@ -58,9 +58,9 @@ public class FinishedTriggersSet implements FinishedTriggers {
}
@Override
- public void clearRecursively(ExecutableTrigger<?> trigger) {
+ public void clearRecursively(ExecutableTrigger trigger) {
finishedTriggers.remove(trigger);
- for (ExecutableTrigger<?> subTrigger : trigger.subTriggers()) {
+ for (ExecutableTrigger subTrigger : trigger.subTriggers()) {
clearRecursively(subTrigger);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/de9f10c4/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ReshuffleTrigger.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ReshuffleTrigger.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ReshuffleTrigger.java
index 98a9d94..d772566 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ReshuffleTrigger.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/ReshuffleTrigger.java
@@ -30,34 +30,34 @@ import java.util.List;
*
* @param <W> The kind of window that is being reshuffled.
*/
-public class ReshuffleTrigger<W extends BoundedWindow> extends Trigger<W> {
+public class ReshuffleTrigger<W extends BoundedWindow> extends Trigger {
ReshuffleTrigger() {
super(null);
}
@Override
- public void onElement(Trigger<W>.OnElementContext c) { }
+ public void onElement(Trigger.OnElementContext c) { }
@Override
- public void onMerge(Trigger<W>.OnMergeContext c) { }
+ public void onMerge(Trigger.OnMergeContext c) { }
@Override
- protected Trigger<W> getContinuationTrigger(List<Trigger<W>> continuationTriggers) {
+ protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
return this;
}
@Override
- public Instant getWatermarkThatGuaranteesFiring(W window) {
+ public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
throw new UnsupportedOperationException(
"ReshuffleTrigger should not be used outside of Reshuffle");
}
@Override
- public boolean shouldFire(Trigger<W>.TriggerContext context) throws Exception {
+ public boolean shouldFire(Trigger.TriggerContext context) throws Exception {
return true;
}
@Override
- public void onFire(Trigger<W>.TriggerContext context) throws Exception { }
+ public void onFire(Trigger.TriggerContext context) throws Exception { }
}
[3/3] incubator-beam git commit: This closes #168
Posted by ke...@apache.org.
This closes #168
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c558a64c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c558a64c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c558a64c
Branch: refs/heads/master
Commit: c558a64c3892a1d00af9f99927e8fbe277523e3a
Parents: af1e575 de9f10c
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Apr 12 09:31:33 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Apr 12 09:31:33 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/windowing/AfterAll.java | 30 +++---
.../windowing/AfterDelayFromFirstElement.java | 28 ++---
.../sdk/transforms/windowing/AfterEach.java | 29 +++---
.../sdk/transforms/windowing/AfterFirst.java | 33 +++---
.../sdk/transforms/windowing/AfterPane.java | 23 ++---
.../windowing/AfterProcessingTime.java | 22 ++--
.../AfterSynchronizedProcessingTime.java | 11 +-
.../transforms/windowing/AfterWatermark.java | 102 +++++++++----------
.../transforms/windowing/DefaultTrigger.java | 20 ++--
.../transforms/windowing/OrFinallyTrigger.java | 24 ++---
.../sdk/transforms/windowing/Repeatedly.java | 21 ++--
.../sdk/transforms/windowing/Trigger.java | 85 +++++++---------
.../transforms/windowing/TriggerBuilder.java | 6 +-
.../sdk/transforms/windowing/Window.java | 20 ++--
.../dataflow/sdk/util/ExecutableTrigger.java | 58 +++++------
.../dataflow/sdk/util/FinishedTriggers.java | 6 +-
.../sdk/util/FinishedTriggersBitSet.java | 6 +-
.../dataflow/sdk/util/FinishedTriggersSet.java | 16 +--
.../dataflow/sdk/util/ReshuffleTrigger.java | 14 +--
.../sdk/util/TriggerContextFactory.java | 80 +++++++--------
.../cloud/dataflow/sdk/util/TriggerRunner.java | 12 +--
.../dataflow/sdk/util/WindowingStrategy.java | 12 +--
.../sdk/transforms/windowing/AfterAllTest.java | 6 +-
.../sdk/transforms/windowing/AfterEachTest.java | 6 +-
.../transforms/windowing/AfterFirstTest.java | 14 +--
.../windowing/AfterProcessingTimeTest.java | 8 +-
.../AfterSynchronizedProcessingTimeTest.java | 3 +-
.../windowing/AfterWatermarkTest.java | 14 +--
.../windowing/OrFinallyTriggerTest.java | 14 +--
.../transforms/windowing/RepeatedlyTest.java | 14 +--
.../sdk/transforms/windowing/TriggerTest.java | 44 ++++----
.../sdk/transforms/windowing/WindowTest.java | 8 +-
.../sdk/util/ExecutableTriggerTest.java | 18 ++--
.../sdk/util/FinishedTriggersProperties.java | 18 ++--
.../sdk/util/FinishedTriggersSetTest.java | 6 +-
.../dataflow/sdk/util/ReduceFnRunnerTest.java | 21 ++--
.../cloud/dataflow/sdk/util/ReduceFnTester.java | 8 +-
.../cloud/dataflow/sdk/util/TriggerTester.java | 16 +--
38 files changed, 419 insertions(+), 457 deletions(-)
----------------------------------------------------------------------