StreamingSideInputFetcher { public boolean storeIfBlocked(WindowedValue<InputT> elem) { @SuppressWarnings("unchecked") W window = (W) Iterables.getOnlyElement(elem.getWindows()); Set<Windmill.GlobalDataRequest> blocked = blockedMap().get(window); if (blocked == null) { for (PCollectionView<?> view : sideInputViews.values()) { if (!stepContext.issueSideInputFetch(view, window, StateFetcher.SideInputState.UNKNOWN)) { if (blocked == null) { blocked = new HashSet<>(); blockedMap().put(window, blocked); } blocked.add(buildGlobalDataRequest(view, window)); } } } if (blocked != null) { elementBag(window).add(elem); watermarkHold(window).add(elem.getTimestamp()); stepContext.addBlockingSideInputs(blocked); return true; } else { return false; } } StreamingSideInputFetcher(
      Iterable<PCollectionView<?>> views,
      Coder<InputT> inputCoder,
      WindowingStrategy<?, W> windowingStrategy,
      StreamingModeExecutionContext.StreamingModeStepContext stepContext); Set<W> getReadyWindows(); Iterable<BagState<WindowedValue<InputT>>> prefetchElements(Iterable<W> readyWindows); void releaseBlockedWindows(Iterable<W> windows); Iterable<BagState<TimerData>> prefetchTimers(Iterable<W> readyWindows); boolean storeIfBlocked(WindowedValue<InputT> elem); boolean storeIfBlocked(TimerData timer); void persist();  }
@Test public void testStoreIfBlocked() throws Exception { PCollectionView<String> view = createView(); IntervalWindow readyWindow = createWindow(0); Windmill.GlobalDataId id = Windmill.GlobalDataId.newBuilder() .setTag(view.getTagInternal().getId()) .setVersion( ByteString.copyFrom( CoderUtils.encodeToByteArray(IntervalWindow.getCoder(), readyWindow))) .build(); when(stepContext.getSideInputNotifications()) .thenReturn(Arrays.<Windmill.GlobalDataId>asList(id)); when(stepContext.issueSideInputFetch( eq(view), any(BoundedWindow.class), eq(SideInputState.UNKNOWN))) .thenReturn(false); when(stepContext.issueSideInputFetch( eq(view), any(BoundedWindow.class), eq(SideInputState.KNOWN_READY))) .thenReturn(true); StreamingSideInputFetcher<String, IntervalWindow> fetcher = createFetcher(Arrays.asList(view)); WindowedValue<String> datum1 = createDatum("e1", 0); assertTrue(fetcher.storeIfBlocked(datum1)); assertThat(fetcher.getBlockedWindows(), Matchers.contains(createWindow(0))); WindowedValue<String> datum2 = createDatum("e2", 0); assertTrue(fetcher.storeIfBlocked(datum2)); assertThat(fetcher.getBlockedWindows(), Matchers.contains(createWindow(0))); WindowedValue<String> datum3 = createDatum("e3", 10); assertTrue(fetcher.storeIfBlocked(datum3)); assertThat( fetcher.getBlockedWindows(), Matchers.containsInAnyOrder(createWindow(0), createWindow(10))); TimerData timer1 = createTimer(0); assertTrue(fetcher.storeIfBlocked(timer1)); TimerData timer2 = createTimer(15); assertTrue(fetcher.storeIfBlocked(timer2)); assertThat(fetcher.getReadyWindows(), Matchers.contains(readyWindow)); Set<WindowedValue<String>> actualElements = Sets.newHashSet(); for (BagState<WindowedValue<String>> bag : fetcher.prefetchElements(ImmutableList.of(readyWindow))) { for (WindowedValue<String> elem : bag.read()) { actualElements.add(elem); } bag.clear(); } assertThat(actualElements, Matchers.containsInAnyOrder(datum1, datum2)); Set<TimerData> actualTimers = Sets.newHashSet(); for (BagState<TimerData> bag : fetcher.prefetchTimers(ImmutableList.of(readyWindow))) { for (TimerData timer : bag.read()) { actualTimers.add(timer); } bag.clear(); } assertThat(actualTimers, Matchers.contains(timer1)); fetcher.releaseBlockedWindows(ImmutableList.of(readyWindow)); assertThat(fetcher.getBlockedWindows(), Matchers.contains(createWindow(10))); Set<WindowedValue<String>> restElements = Sets.newHashSet(); for (BagState<WindowedValue<String>> bag : fetcher.prefetchElements(ImmutableList.of(createWindow(10), createWindow(15)))) { for (WindowedValue<String> elem : bag.read()) { restElements.add(elem); } } assertThat(restElements, Matchers.contains(datum3)); Set<TimerData> restTimers = Sets.newHashSet(); for (BagState<TimerData> bag : fetcher.prefetchTimers(ImmutableList.of(createWindow(10), createWindow(15)))) { for (TimerData timer : bag.read()) { restTimers.add(timer); } } assertThat(restTimers, Matchers.contains(timer2)); }