@Override
  public void startBundle() {
    simpleDoFnRunner.startBundle();

    // Find the set of ready windows.
    Set<W> readyWindows = sideInputFetcher.getReadyWindows();

    Iterable<BagState<WindowedValue<InputT>>> elementsBags =
        sideInputFetcher.prefetchElements(readyWindows);
    Iterable<BagState<TimerData>> timersBags = sideInputFetcher.prefetchTimers(readyWindows);

    K key = keyValue().read();
    Iterable<TimerData> timers = Iterables.concat(Iterables.transform(timersBags, BagState::read));

    Iterable<WindowedValue<InputT>> elements =
        Iterables.concat(Iterables.transform(elementsBags, BagState::read));

    if (!isEmpty(timers, elements)) {
      simpleDoFnRunner.processElement(
          new ValueInEmptyWindows<>(KeyedWorkItems.workItem(key, timers, elements)));
      for (BagState<WindowedValue<InputT>> bag : elementsBags) {
        bag.clear();
      }
      for (BagState<TimerData> bag : timersBags) {
        bag.clear();
      }
    }
    sideInputFetcher.releaseBlockedWindows(readyWindows);
  }
@Test
  public void testStartBundle() throws Exception {
    ListOutputManager outputManager = new ListOutputManager();
    StreamingKeyedWorkItemSideInputDoFnRunner<String, Integer, KV<String, Integer>, IntervalWindow>
        runner = createRunner(outputManager);

    runner.keyValue().write("a");
    Set<IntervalWindow> readyWindows = ImmutableSet.of(window(10, 20));
    when(sideInputFetcher.getReadyWindows()).thenReturn(readyWindows);
    when(sideInputFetcher.prefetchElements(readyWindows)).thenReturn(ImmutableList.of(elemsBag));
    when(sideInputFetcher.prefetchTimers(readyWindows)).thenReturn(ImmutableList.of(timersBag));
    when(elemsBag.read()).thenReturn(ImmutableList.of(createDatum(13, 13L), createDatum(18, 18L)));
    when(timersBag.read())
        .thenReturn(
            ImmutableList.of(timerData(window(10, 20), new Instant(19), Timer.Type.WATERMARK)));
    when(mockTimerInternals.currentInputWatermarkTime()).thenReturn(new Instant(20));

    runner.startBundle();

    List<WindowedValue<KV<String, Integer>>> result = outputManager.getOutput(mainOutputTag);
    assertEquals(1, result.size());
    WindowedValue<KV<String, Integer>> item0 = result.get(0);
    assertEquals("a", item0.getValue().getKey());
    assertEquals(31, item0.getValue().getValue().intValue());
  }