docs: @SuppressWarnings("unchecked")
  	@Override
  	public <N, S extends State> S getPartitionedState(
  			final N namespace,
  			final TypeSerializer<N> namespaceSerializer,
  			final StateDescriptor<S, ?> stateDescriptor) throws Exception {
  
  		checkNotNull(namespace, "Namespace");
  
  		if (lastName != null && lastName.equals(stateDescriptor.getName())) {
  			lastState.setCurrentNamespace(namespace);
  			return (S) lastState;
  		}
  
  		InternalKvState<K, ?, ?> previous = keyValueStatesByName.get(stateDescriptor.getName());
  		if (previous != null) {
  			lastState = previous;
  			lastState.setCurrentNamespace(namespace);
  			lastName = stateDescriptor.getName();
  			return (S) previous;
  		}
  
  		final S state = getOrCreateKeyedState(namespaceSerializer, stateDescriptor);
  		final InternalKvState<K, N, ?> kvState = (InternalKvState<K, N, ?>) state;
  
  		lastName = stateDescriptor.getName();
  		lastState = kvState;
  		kvState.setCurrentNamespace(namespace);
  
  		return state;
  	}