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, 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;
    }
  }
@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));

    // Verify storeIfBlocked
    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));

    // Verify ready windows
    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));

    // Verify releaseBlockedWindows
    fetcher.releaseBlockedWindows(ImmutableList.of(readyWindow));
    assertThat(fetcher.getBlockedWindows(), Matchers.contains(createWindow(10)));

    // Verify rest elements and timers
    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));
  }