code: public void registerListener(JobID jobId, KvStateRegistryListener listener) {
  		final KvStateRegistryListener previousValue = listeners.putIfAbsent(jobId, listener);
  
  		if (previousValue != null) {
  			throw new IllegalStateException("Listener already registered under " + jobId + '.');
  		}
  	}
docs: Registers a listener with the registry.
  
  @param jobId identifying the job for which to register a {@link KvStateRegistryListener}
  @param listener The registry listener.
  @throws IllegalStateException If there is a registered listener