StreamingKeyedWorkItemSideInputDoFnRunner implements DoFnRunner<KeyedWorkItem<K, InputT>, OutputT> { @Override public void startBundle() { simpleDoFnRunner.startBundle(); 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); } StreamingKeyedWorkItemSideInputDoFnRunner(
      DoFnRunner<KeyedWorkItem<K, InputT>, OutputT> simpleDoFnRunner,
      Coder<K> keyCoder,
      StreamingSideInputFetcher<InputT, W> sideInputFetcher,
      StepContext stepContext); @Override void startBundle(); @Override void processElement(WindowedValue<KeyedWorkItem<K, InputT>> elem); @Override void onTimer(
      String timerId,
      String timerFamilyId,
      KeyT key,
      BoundedWindow window,
      Instant timestamp,
      Instant outputTimestamp,
      TimeDomain timeDomain); @Override void finishBundle(); @Override void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key); @Override DoFn<KeyedWorkItem<K, InputT>, OutputT> getFn();  }
@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()); }