@VisibleForTesting
  boolean hasNoActiveWindows() {
    return activeWindows.getActiveAndNewWindows().isEmpty();
  }
@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));

    // One elements in one session window.
    tester.injectElements(TimestampedValue.of(1, new Instant(1))); // in [1, 11), gc at 21.

    // Close the trigger, but the gargbage collection timer is still pending.
    when(mockTriggerStateMachine.shouldFire(anyTriggerContext())).thenReturn(true);
    triggerShouldFinish(mockTriggerStateMachine);
    tester.advanceInputWatermark(new Instant(15));
    tester.fireTimer(mergedWindow, mergedWindow.maxTimestamp(), TimeDomain.EVENT_TIME);

    // Another element in the same session window.
    // Should be discarded with 'window closed'.
    tester.injectElements(TimestampedValue.of(1, new Instant(1))); // in [1, 11), gc at 21.

    // And nothing should be left in the active window state.
    assertTrue(tester.hasNoActiveWindows());

    // Now the garbage collection timer will fire, finding the trigger already closed.
    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)), // timestamp
            equalTo((BoundedWindow) mergedWindow)));

    assertThat(
        output.get(0).getPane(), equalTo(PaneInfo.createPane(true, true, Timing.ON_TIME, 0, 0)));
  }