code: @Override
  	@PublicEvolving
  	public <R> SingleOutputStreamOperator<R> transform(String operatorName,
  			TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
  
  		SingleOutputStreamOperator<R> returnStream = super.transform(operatorName, outTypeInfo, operator);
  
  		// inject the key selector and key type
  		OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation();
  		transform.setStateKeySelector(keySelector);
  		transform.setStateKeyType(keyType);
  
  		return returnStream;
  	}
docs: ------------------------------------------------------------------------