code: private <S extends State, V> S createKeyedState(StateDescriptor<S, V> stateDescriptor) throws Exception {
  		checkNotNull(keyedStateBackend.getKeySerializer(),
  			"State key serializer has not been configured in the config. " +
  			"This operation cannot use partitioned state.");
  		if (!stateDescriptor.isSerializerInitialized()) {
  			stateDescriptor.initializeSerializerUnlessSet(ctx.getExecutionConfig());
  		}
  		return keyedStateBackend.createInternalState(windowSerializer, stateDescriptor);
  	}
docs: Delegate the state registration to {@link KeyedStateBackend#createInternalState(TypeSerializer, StateDescriptor)}.
  We can't use {@link KeyedStateBackend#getOrCreateKeyedState(TypeSerializer, StateDescriptor)}
  here because it returns the same state instance not always a new instance.