code: public void notifyKvStateRegistered(
  			JobVertexID jobVertexId,
  			KeyGroupRange keyGroupRange,
  			String registrationName,
  			KvStateID kvStateId,
  			InetSocketAddress kvStateServerAddress) {
  
  		KvStateLocation location = lookupTable.get(registrationName);
  
  		if (location == null) {
  			// First registration for this operator, create the location info
  			ExecutionJobVertex vertex = jobVertices.get(jobVertexId);
  
  			if (vertex != null) {
  				int parallelism = vertex.getMaxParallelism();
  				location = new KvStateLocation(jobId, jobVertexId, parallelism, registrationName);
  				lookupTable.put(registrationName, location);
  			} else {
  				throw new IllegalArgumentException("Unknown JobVertexID " + jobVertexId);
  			}
  		}
  
  		// Duplicated name if vertex IDs don't match
  		if (!location.getJobVertexId().equals(jobVertexId)) {
  			IllegalStateException duplicate = new IllegalStateException(
  					"Registration name clash. KvState with name '" + registrationName +
  							"' has already been registered by another operator (" +
  							location.getJobVertexId() + ").");
  
  			ExecutionJobVertex vertex = jobVertices.get(jobVertexId);
  			if (vertex != null) {
  				vertex.fail(new SuppressRestartsException(duplicate));
  			}
  
  			throw duplicate;
  		}
  		location.registerKvState(keyGroupRange, kvStateId, kvStateServerAddress);
  	}
docs: Notifies the registry about a registered KvState instance.
  
  @param jobVertexId JobVertexID the KvState instance belongs to
  @param keyGroupRange Key group range the KvState instance belongs to
  @param registrationName Name under which the KvState has been registered
  @param kvStateId ID of the registered KvState instance
  @param kvStateServerAddress Server address where to find the KvState instance
  
  @throws IllegalArgumentException If JobVertexID does not belong to job
  @throws IllegalArgumentException If state has been registered with same
  name by another operator.
  @throws IndexOutOfBoundsException If key group index is out of bounds.