ReduceFnRunner { @VisibleForTesting boolean hasNoActiveWindows() { return activeWindows.getActiveAndNewWindows().isEmpty(); } ReduceFnRunner(
      K key,
      WindowingStrategy<?, W> windowingStrategy,
      ExecutableTriggerStateMachine triggerStateMachine,
      StateInternals stateInternals,
      TimerInternals timerInternals,
      OutputWindowedValue<KV<K, OutputT>> outputter,
      @Nullable SideInputReader sideInputReader,
      ReduceFn<K, InputT, OutputT, W> reduceFn,
      @Nullable PipelineOptions options); void processElements(Iterable<WindowedValue<InputT>> values); void persist(); void onTimers(Iterable<TimerData> timers); static final String DROPPED_DUE_TO_CLOSED_WINDOW; }
@Test public void testMergingWithReusedWindow() throws Exception { Duration allowedLateness = Duration.millis(50); ReduceFnTester<Integer, Iterable<Integer>, IntervalWindow> tester = ReduceFnTester.nonCombining( Sessions.withGapDuration(Duration.millis(10)), mockTriggerStateMachine, AccumulationMode.DISCARDING_FIRED_PANES, allowedLateness, ClosingBehavior.FIRE_IF_NON_EMPTY); IntervalWindow mergedWindow = new IntervalWindow(new Instant(1), new Instant(11)); tester.injectElements(TimestampedValue.of(1, new Instant(1))); when(mockTriggerStateMachine.shouldFire(anyTriggerContext())).thenReturn(true); triggerShouldFinish(mockTriggerStateMachine); tester.advanceInputWatermark(new Instant(15)); tester.fireTimer(mergedWindow, mergedWindow.maxTimestamp(), TimeDomain.EVENT_TIME); tester.injectElements(TimestampedValue.of(1, new Instant(1))); assertTrue(tester.hasNoActiveWindows()); tester.advanceInputWatermark(new Instant(100)); List<WindowedValue<Iterable<Integer>>> output = tester.extractOutput(); assertThat(output.size(), equalTo(1)); assertThat( output.get(0), isSingleWindowedValue( containsInAnyOrder(1), equalTo(new Instant(1)), equalTo((BoundedWindow) mergedWindow))); assertThat( output.get(0).getPane(), equalTo(PaneInfo.createPane(true, true, Timing.ON_TIME, 0, 0))); }