AfterWatermarkStateMachine { public static FromEndOfWindow pastEndOfWindow() { return new FromEndOfWindow(); } private  AfterWatermarkStateMachine(); static FromEndOfWindow pastEndOfWindow();  }
@Test public void testEarlyAndAtWatermark() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow().withEarlyFirings(mockEarly), FixedWindows.of(Duration.millis(100))); injectElements(1); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); testRunningAsTrigger(mockEarly, window); when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(false); tester.advanceInputWatermark(new Instant(100)); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertTrue(tester.isMarkedFinished(window)); }
@Test public void testTimerForEndOfWindow() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow(), FixedWindows.of(Duration.millis(100))); assertThat(tester.getNextTimer(TimeDomain.EVENT_TIME), nullValue()); injectElements(1); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); assertThat(tester.getNextTimer(TimeDomain.EVENT_TIME), equalTo(window.maxTimestamp())); }
@Test public void testTimerForEndOfWindowCompound() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow().withEarlyFirings(NeverStateMachine.ever()), FixedWindows.of(Duration.millis(100))); assertThat(tester.getNextTimer(TimeDomain.EVENT_TIME), nullValue()); injectElements(1); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); assertThat(tester.getNextTimer(TimeDomain.EVENT_TIME), equalTo(window.maxTimestamp())); }
@Test public void testAtWatermarkAndLate() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow().withLateFirings(mockLate), FixedWindows.of(Duration.millis(100))); injectElements(1); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(true); assertFalse(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(false); tester.advanceInputWatermark(new Instant(100)); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); testRunningAsTrigger(mockLate, window); }
@Test public void testEarlyAndAtWatermarkAndLate() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(mockEarly) .withLateFirings(mockLate), FixedWindows.of(Duration.millis(100))); injectElements(1); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100)); testRunningAsTrigger(mockEarly, window); when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(false); tester.advanceInputWatermark(new Instant(100)); assertTrue(tester.shouldFire(window)); tester.fireIfShouldFire(window); assertFalse(tester.isMarkedFinished(window)); testRunningAsTrigger(mockLate, window); }
@Test public void testOnMergeAlreadyFinished() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterEachStateMachine.inOrder( AfterWatermarkStateMachine.pastEndOfWindow(), RepeatedlyStateMachine.forever(AfterPaneStateMachine.elementCountAtLeast(1))), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); tester.injectElements(5); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); IntervalWindow secondWindow = new IntervalWindow(new Instant(5), new Instant(15)); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15)); tester.advanceInputWatermark(new Instant(15)); assertTrue(tester.shouldFire(firstWindow)); assertTrue(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); tester.fireIfShouldFire(secondWindow); assertFalse(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.injectElements(1); tester.injectElements(5); assertTrue(tester.shouldFire(firstWindow)); assertTrue(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); tester.fireIfShouldFire(secondWindow); tester.mergeWindows(); assertFalse(tester.shouldFire(mergedWindow)); tester.injectElements(1); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testOnMergeRewinds() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterEachStateMachine.inOrder( AfterWatermarkStateMachine.pastEndOfWindow(), RepeatedlyStateMachine.forever(AfterPaneStateMachine.elementCountAtLeast(1))), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); tester.injectElements(5); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); IntervalWindow secondWindow = new IntervalWindow(new Instant(5), new Instant(15)); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15)); tester.advanceInputWatermark(new Instant(11)); assertTrue(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); assertFalse(tester.shouldFire(firstWindow)); tester.injectElements(1); assertTrue(tester.shouldFire(firstWindow)); tester.fireIfShouldFire(firstWindow); tester.mergeWindows(); assertFalse(tester.shouldFire(mergedWindow)); tester.injectElements(1); assertFalse(tester.shouldFire(mergedWindow)); tester.advanceInputWatermark(new Instant(15)); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testEarlyAndLateOnMergeAlreadyFinished() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(AfterPaneStateMachine.elementCountAtLeast(100)) .withLateFirings(AfterPaneStateMachine.elementCountAtLeast(1)), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); tester.injectElements(5); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); IntervalWindow secondWindow = new IntervalWindow(new Instant(5), new Instant(15)); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15)); tester.advanceInputWatermark(new Instant(15)); assertTrue(tester.shouldFire(firstWindow)); assertTrue(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); tester.fireIfShouldFire(secondWindow); assertFalse(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.injectElements(1); tester.injectElements(5); assertTrue(tester.shouldFire(firstWindow)); assertTrue(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); tester.fireIfShouldFire(secondWindow); tester.mergeWindows(); assertFalse(tester.shouldFire(mergedWindow)); tester.injectElements(1); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testEarlyAndLateOnMergeSubtriggerMerges() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(mockEarly) .withLateFirings(mockLate), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); tester.injectElements(5); tester.mergeWindows(); verify(mockEarly).onMerge(Mockito.any(OnMergeContext.class)); }
@Test public void testEarlyAndLateOnMergeRewinds() throws Exception { tester = TriggerStateMachineTester.forTrigger( AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(AfterPaneStateMachine.elementCountAtLeast(100)) .withLateFirings(AfterPaneStateMachine.elementCountAtLeast(1)), Sessions.withGapDuration(Duration.millis(10))); tester.injectElements(1); tester.injectElements(5); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); IntervalWindow secondWindow = new IntervalWindow(new Instant(5), new Instant(15)); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(15)); tester.advanceInputWatermark(new Instant(11)); assertTrue(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); assertFalse(tester.shouldFire(firstWindow)); tester.injectElements(1); assertTrue(tester.shouldFire(firstWindow)); tester.fireIfShouldFire(firstWindow); tester.mergeWindows(); assertFalse(tester.shouldFire(mergedWindow)); tester.injectElements(1); assertFalse(tester.shouldFire(mergedWindow)); tester.advanceInputWatermark(new Instant(15)); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testFromEndOfWindowToString() { TriggerStateMachine trigger = AfterWatermarkStateMachine.pastEndOfWindow(); assertEquals("AfterWatermark.pastEndOfWindow()", trigger.toString()); }
@Test public void testEarlyFiringsToString() { TriggerStateMachine trigger = AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(StubTriggerStateMachine.named("t1")); assertEquals("AfterWatermark.pastEndOfWindow().withEarlyFirings(t1)", trigger.toString()); }
@Test public void testLateFiringsToString() { TriggerStateMachine trigger = AfterWatermarkStateMachine.pastEndOfWindow() .withLateFirings(StubTriggerStateMachine.named("t1")); assertEquals("AfterWatermark.pastEndOfWindow().withLateFirings(t1)", trigger.toString()); }
@Test public void testEarlyAndLateFiringsToString() { TriggerStateMachine trigger = AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(StubTriggerStateMachine.named("t1")) .withLateFirings(StubTriggerStateMachine.named("t2")); assertEquals( "AfterWatermark.pastEndOfWindow().withEarlyFirings(t1).withLateFirings(t2)", trigger.toString()); }
@Test public void testToStringExcludesNeverTrigger() { TriggerStateMachine trigger = AfterWatermarkStateMachine.pastEndOfWindow() .withEarlyFirings(NeverStateMachine.ever()) .withLateFirings(NeverStateMachine.ever()); assertEquals("AfterWatermark.pastEndOfWindow()", trigger.toString()); }