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 testFetchGlobalDataCacheOverflow() throws Exception {
    Coder<List<String>> coder = ListCoder.of(StringUtf8Coder.of());

    ByteStringOutputStream stream = new ByteStringOutputStream();
    coder.encode(Collections.singletonList("data1"), stream, Coder.Context.OUTER);
    ByteString encodedIterable1 = stream.toByteStringAndReset();
    coder.encode(Collections.singletonList("data2"), stream, Coder.Context.OUTER);
    ByteString encodedIterable2 = stream.toByteString();

    Cache<SideInputCache.Key<?>, SideInput<?>> cache = CacheBuilder.newBuilder().build();

    SideInputStateFetcher fetcher =
        new SideInputStateFetcher(server::getSideInputData, new SideInputCache(cache));

    PCollectionView<String> view1 =
        TestPipeline.create().apply(Create.empty(StringUtf8Coder.of())).apply(View.asSingleton());

    PCollectionView<String> view2 =
        TestPipeline.create().apply(Create.empty(StringUtf8Coder.of())).apply(View.asSingleton());

    String tag1 = view1.getTagInternal().getId();
    String tag2 = view2.getTagInternal().getId();

    // Test four calls in a row. First, fetch view1, then view2 (which evicts view1 from the cache),
    // then view 1 again twice.
    when(server.getSideInputData(any(Windmill.GlobalDataRequest.class)))
        .thenReturn(
            buildGlobalDataResponse(tag1, true, encodedIterable1),
            buildGlobalDataResponse(tag2, true, encodedIterable2),
            buildGlobalDataResponse(tag1, true, encodedIterable1));

    assertEquals(
        "data1",
        fetcher
            .fetchSideInput(
                view1,
                GlobalWindow.INSTANCE,
                STATE_FAMILY,
                SideInputState.UNKNOWN,
                readStateSupplier)
            .value()
            .orElse(null));
    assertEquals(
        "data2",
        fetcher
            .fetchSideInput(
                view2,
                GlobalWindow.INSTANCE,
                STATE_FAMILY,
                SideInputState.UNKNOWN,
                readStateSupplier)
            .value()
            .orElse(null));
    cache.invalidateAll();
    assertEquals(
        "data1",
        fetcher
            .fetchSideInput(
                view1,
                GlobalWindow.INSTANCE,
                STATE_FAMILY,
                SideInputState.UNKNOWN,
                readStateSupplier)
            .value()
            .orElse(null));
    assertEquals(
        "data1",
        fetcher
            .fetchSideInput(
                view1,
                GlobalWindow.INSTANCE,
                STATE_FAMILY,
                SideInputState.UNKNOWN,
                readStateSupplier)
            .value()
            .orElse(null));

    ArgumentCaptor<Windmill.GlobalDataRequest> captor =
        ArgumentCaptor.forClass(Windmill.GlobalDataRequest.class);

    verify(server, times(3)).getSideInputData(captor.capture());
    verifyNoMoreInteractions(server);

    assertThat(
        captor.getAllValues(),
        contains(
            buildGlobalDataRequest(tag1),
            buildGlobalDataRequest(tag2),
            buildGlobalDataRequest(tag1)));
  }