StateRequestHandlers { public static StateRequestHandler forBagUserStateHandlerFactory( ExecutableProcessBundleDescriptor processBundleDescriptor, BagUserStateHandlerFactory bagUserStateHandlerFactory) { return new ByteStringStateRequestHandlerToBagUserStateHandlerFactoryAdapter( processBundleDescriptor, bagUserStateHandlerFactory); }  static StateRequestHandler delegateBasedUponType(
      EnumMap<StateKey.TypeCase, StateRequestHandler> handlers); static StateRequestHandler forSideInputHandlerFactory(
      Map<String, Map<String, SideInputSpec>> sideInputSpecs,
      SideInputHandlerFactory sideInputHandlerFactory); static StateRequestHandler forBagUserStateHandlerFactory(
      ExecutableProcessBundleDescriptor processBundleDescriptor,
      BagUserStateHandlerFactory 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)); }