public static StateRequestHandler forBagUserStateHandlerFactory(
      ExecutableProcessBundleDescriptor processBundleDescriptor,
      BagUserStateHandlerFactory bagUserStateHandlerFactory) {
    return new ByteStringStateRequestHandlerToBagUserStateHandlerFactoryAdapter(
        processBundleDescriptor, bagUserStateHandlerFactory);
  }
@Test
  public void testUserStateCacheTokenGeneration() throws Exception {
    ExecutableStage stage = buildExecutableStage("state1", "state2");
    ProcessBundleDescriptors.ExecutableProcessBundleDescriptor descriptor =
        ProcessBundleDescriptors.fromExecutableStage(
            "id", stage, Endpoints.ApiServiceDescriptor.getDefaultInstance());

    InMemoryBagUserStateFactory inMemoryBagUserStateFactory = new InMemoryBagUserStateFactory<>();
    assertThat(inMemoryBagUserStateFactory.handlers.size(), is(0));

    StateRequestHandler stateRequestHandler =
        StateRequestHandlers.forBagUserStateHandlerFactory(descriptor, inMemoryBagUserStateFactory);
    final BeamFnApi.ProcessBundleRequest.CacheToken cacheToken =
        assertSingleCacheToken(stateRequestHandler);

    sendGetRequest(stateRequestHandler, "state1");
    assertThat(inMemoryBagUserStateFactory.handlers.size(), is(1));
    assertThat(assertSingleCacheToken(stateRequestHandler), is(cacheToken));

    sendGetRequest(stateRequestHandler, "state2");
    assertThat(inMemoryBagUserStateFactory.handlers.size(), is(2));
    assertThat(assertSingleCacheToken(stateRequestHandler), is(cacheToken));
  }