public <T> SideInput<T> fetchSideInput(
      PCollectionView<T> view,
      BoundedWindow sideWindow,
      String stateFamily,
      SideInputState state,
      Supplier<Closeable> scopedReadStateSupplier) {
    Callable<SideInput<T>> loadSideInputFromWindmill =
        () -> loadSideInputFromWindmill(view, sideWindow, stateFamily, scopedReadStateSupplier);
    SideInputCache.Key<T> sideInputCacheKey =
        SideInputCache.Key.create(
            getInternalTag(view), sideWindow, getViewFn(view).getTypeDescriptor());

    try {
      if (state == SideInputState.KNOWN_READY) {
        Optional<SideInput<T>> existingCacheEntry = sideInputCache.get(sideInputCacheKey);
        if (!existingCacheEntry.isPresent()) {
          return sideInputCache.getOrLoad(sideInputCacheKey, loadSideInputFromWindmill);
        }

        if (!existingCacheEntry.get().isReady()) {
          return sideInputCache.invalidateThenLoadNewEntry(
              sideInputCacheKey, loadSideInputFromWindmill);
        }

        return existingCacheEntry.get();
      }

      return sideInputCache.getOrLoad(sideInputCacheKey, loadSideInputFromWindmill);
    } catch (Exception e) {
      LOG.error("Fetch failed: ", e);
      throw new RuntimeException("Exception while fetching side input: ", e);
    }
  }
@Test
  public void testEmptyFetchGlobalData() {
    SideInputStateFetcherFactory factory =
        SideInputStateFetcherFactory.fromOptions(
            PipelineOptionsFactory.as(DataflowStreamingPipelineOptions.class));
    SideInputStateFetcher fetcher = factory.createSideInputStateFetcher(server::getSideInputData);

    ByteString encodedIterable = ByteString.EMPTY;

    PCollectionView<Long> view =
        TestPipeline.create()
            .apply(Create.empty(VarLongCoder.of()))
            .apply(Sum.longsGlobally().asSingletonView());

    String tag = view.getTagInternal().getId();

    // Test three calls in a row. First, data is not ready, then data is ready,
    // then the data is already cached.
    when(server.getSideInputData(any(Windmill.GlobalDataRequest.class)))
        .thenReturn(buildGlobalDataResponse(tag, true, encodedIterable));

    assertEquals(
        0L,
        (long)
            fetcher
                .fetchSideInput(
                    view,
                    GlobalWindow.INSTANCE,
                    STATE_FAMILY,
                    SideInputState.UNKNOWN,
                    readStateSupplier)
                .value()
                .orElse(null));

    verify(server).getSideInputData(buildGlobalDataRequest(tag));
    verifyNoMoreInteractions(server);
  }