public KvStateID registerKvState ( JobID jobId , JobVertexID jobVertexId , KeyGroupRange keyGroupRange , String registrationName , InternalKvState < ? , ? , ? > kvState ) { KvStateID kvStateId = new KvStateID ( ) ; if ( registeredKvStates . putIfAbsent ( kvStateId , new KvStateEntry <> ( kvState ) ) == null ) { final KvStateRegistryListener listener = getKvStateRegistryListener ( jobId ) ; if ( listener != null ) { listener . notifyKvStateRegistered ( jobId , jobVertexId , keyGroupRange , registrationName , kvStateId ) ; } return kvStateId ; } else { throw new IllegalStateException ( "State \"" + registrationName + " \"(id=" + kvStateId + ") appears registered although it should not." ) ; } }
Registers the KvState instance and returns the assigned ID .