SimplePushbackSideInputDoFnRunner implements PushbackSideInputDoFnRunner<InputT, OutputT> { @Override public Iterable<WindowedValue<InputT>> processElementInReadyWindows(WindowedValue<InputT> elem) { if (views.isEmpty()) { underlying.processElement(elem); return Collections.emptyList(); } ImmutableList.Builder<WindowedValue<InputT>> pushedBack = ImmutableList.builder(); for (WindowedValue<InputT> windowElem : elem.explodeWindows()) { BoundedWindow mainInputWindow = Iterables.getOnlyElement(windowElem.getWindows()); if (isReady(mainInputWindow)) { underlying.processElement(windowElem); } else { notReadyWindows.add(mainInputWindow); pushedBack.add(windowElem); } } return pushedBack.build(); } 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 processElementNoSideInputs() { SimplePushbackSideInputDoFnRunner<Integer, Integer> runner = createRunner(ImmutableList.of()); WindowedValue<Integer> multiWindow = WindowedValue.of( 2, new Instant(-2), ImmutableList.of( new IntervalWindow(new Instant(-500L), new Instant(0L)), new IntervalWindow(BoundedWindow.TIMESTAMP_MIN_VALUE, new Instant(250L)), GlobalWindow.INSTANCE), PaneInfo.ON_TIME_AND_ONLY_FIRING); Iterable<WindowedValue<Integer>> multiWindowPushback = runner.processElementInReadyWindows(multiWindow); assertThat(multiWindowPushback, emptyIterable()); assertThat(underlying.inputElems, containsInAnyOrder(multiWindow)); }