AfterProcessingTimeStateMachine extends AfterDelayFromFirstElementStateMachine { public static AfterProcessingTimeStateMachine pastFirstElementInPane() { return new AfterProcessingTimeStateMachine(IDENTITY); } private  AfterProcessingTimeStateMachine(List<SerializableFunction<Instant, Instant>> transforms); @Override @Nullable Instant getCurrentTime(TriggerStateMachine.TriggerContext context); static AfterProcessingTimeStateMachine pastFirstElementInPane(); @Override String toString(); @Override boolean equals(@Nullable Object obj); @Override int hashCode();  }
@Test public void testAfterProcessingTimeFixedWindows() throws Exception { Duration windowDuration = Duration.millis(10); SimpleTriggerStateMachineTester<IntervalWindow> tester = TriggerStateMachineTester.forTrigger( AfterProcessingTimeStateMachine.pastFirstElementInPane() .plusDelayOf(Duration.millis(5)), FixedWindows.of(windowDuration)); tester.advanceProcessingTime(new Instant(10)); tester.injectElements(1); IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(10)); tester.advanceProcessingTime(new Instant(12)); assertFalse(tester.shouldFire(firstWindow)); tester.injectElements(11, 12, 13); IntervalWindow secondWindow = new IntervalWindow(new Instant(10), new Instant(20)); assertFalse(tester.shouldFire(secondWindow)); tester.advanceProcessingTime(new Instant(14)); assertFalse(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.injectElements(2, 3); tester.advanceProcessingTime(new Instant(16)); assertTrue(tester.shouldFire(firstWindow)); assertFalse(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(firstWindow); assertTrue(tester.isMarkedFinished(firstWindow)); tester.advanceProcessingTime(new Instant(18)); assertTrue(tester.shouldFire(secondWindow)); tester.fireIfShouldFire(secondWindow); assertTrue(tester.isMarkedFinished(secondWindow)); }
@Test public void testClear() throws Exception { SimpleTriggerStateMachineTester<IntervalWindow> tester = TriggerStateMachineTester.forTrigger( AfterProcessingTimeStateMachine.pastFirstElementInPane() .plusDelayOf(Duration.millis(5)), FixedWindows.of(Duration.millis(10))); tester.injectElements(1, 2, 3); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(10)); tester.clearState(window); tester.assertCleared(window); }
@Test public void testAfterProcessingTimeWithMergingWindow() throws Exception { SimpleTriggerStateMachineTester<IntervalWindow> tester = TriggerStateMachineTester.forTrigger( AfterProcessingTimeStateMachine.pastFirstElementInPane() .plusDelayOf(Duration.millis(5)), Sessions.withGapDuration(Duration.millis(10))); tester.advanceProcessingTime(new Instant(10)); tester.injectElements(1); IntervalWindow firstWindow = new IntervalWindow(new Instant(1), new Instant(11)); assertFalse(tester.shouldFire(firstWindow)); tester.advanceProcessingTime(new Instant(12)); tester.injectElements(3); IntervalWindow secondWindow = new IntervalWindow(new Instant(3), new Instant(13)); assertFalse(tester.shouldFire(secondWindow)); tester.mergeWindows(); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(13)); tester.advanceProcessingTime(new Instant(16)); assertTrue(tester.shouldFire(mergedWindow)); }
@Test public void testCompatibilityIdentical() throws Exception { TriggerStateMachine t1 = AfterProcessingTimeStateMachine.pastFirstElementInPane() .plusDelayOf(Duration.standardMinutes(1L)); TriggerStateMachine t2 = AfterProcessingTimeStateMachine.pastFirstElementInPane() .plusDelayOf(Duration.standardMinutes(1L)); assertTrue(t1.isCompatible(t2)); }