code: public static KvStateService fromConfiguration(TaskManagerServicesConfiguration taskManagerServicesConfiguration) {
  		KvStateRegistry kvStateRegistry = new KvStateRegistry();
  
  		QueryableStateConfiguration qsConfig = taskManagerServicesConfiguration.getQueryableStateConfig();
  
  		KvStateClientProxy kvClientProxy = null;
  		KvStateServer kvStateServer = null;
  
  		if (qsConfig != null) {
  			int numProxyServerNetworkThreads = qsConfig.numProxyServerThreads() == 0 ?
  				taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numProxyServerThreads();
  			int numProxyServerQueryThreads = qsConfig.numProxyQueryThreads() == 0 ?
  				taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numProxyQueryThreads();
  			kvClientProxy = QueryableStateUtils.createKvStateClientProxy(
  				taskManagerServicesConfiguration.getTaskManagerAddress(),
  				qsConfig.getProxyPortRange(),
  				numProxyServerNetworkThreads,
  				numProxyServerQueryThreads,
  				new DisabledKvStateRequestStats());
  
  			int numStateServerNetworkThreads = qsConfig.numStateServerThreads() == 0 ?
  				taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numStateServerThreads();
  			int numStateServerQueryThreads = qsConfig.numStateQueryThreads() == 0 ?
  				taskManagerServicesConfiguration.getNumberOfSlots() : qsConfig.numStateQueryThreads();
  			kvStateServer = QueryableStateUtils.createKvStateServer(
  				taskManagerServicesConfiguration.getTaskManagerAddress(),
  				qsConfig.getStateServerPortRange(),
  				numStateServerNetworkThreads,
  				numStateServerQueryThreads,
  				kvStateRegistry,
  				new DisabledKvStateRequestStats());
  		}
  
  		return new KvStateService(kvStateRegistry, kvStateServer, kvClientProxy);
  	}
docs: Creates and returns the KvState service.
  
  @param taskManagerServicesConfiguration task manager configuration
  @return service for kvState related components