SimplePushbackSideInputDoFnRunner implements PushbackSideInputDoFnRunner<InputT, OutputT> { @Override public <KeyT> void onTimer( String timerId, String timerFamilyId, KeyT key, BoundedWindow window, Instant timestamp, Instant outputTimestamp, TimeDomain timeDomain) { underlying.onTimer(timerId, timerFamilyId, key, window, timestamp, outputTimestamp, timeDomain); } private  SimplePushbackSideInputDoFnRunner(
      DoFnRunner<InputT, OutputT> underlying,
      Collection<PCollectionView<?>> views,
      ReadyCheckingSideInputReader sideInputReader); static SimplePushbackSideInputDoFnRunner<InputT, OutputT> create(
      DoFnRunner<InputT, OutputT> underlying,
      Collection<PCollectionView<?>> views,
      ReadyCheckingSideInputReader sideInputReader); @Override DoFn<InputT, OutputT> getFn(); @Override void startBundle(); @Override Iterable<WindowedValue<InputT>> processElementInReadyWindows(WindowedValue<InputT> elem); @Override void onTimer(
      String timerId,
      String timerFamilyId,
      KeyT key,
      BoundedWindow window,
      Instant timestamp,
      Instant outputTimestamp,
      TimeDomain timeDomain); @Override void finishBundle();  }
@Test public void testOnTimerCalled() { PushbackSideInputDoFnRunner<Integer, Integer> runner = createRunner(ImmutableList.of()); String timerId = "fooTimer"; IntervalWindow window = new IntervalWindow(new Instant(4), new Instant(16)); Instant timestamp = new Instant(72); runner.onTimer( timerId, "", null, window, new Instant(timestamp), new Instant(timestamp), TimeDomain.EVENT_TIME); assertThat( underlying.firedTimers, contains( TimerData.of( timerId, StateNamespaces.window(IntervalWindow.getCoder(), window), timestamp, timestamp, TimeDomain.EVENT_TIME))); }