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 ) ; }
Creates and returns the KvState service .