docs: public DataStreamSink<T> addSink(SinkFunction<T> sinkFunction) {
  
  		// read the output type of the input Transform to coax out errors about MissingTypeInfo
  		transformation.getOutputType();
  
  		// configure the type if needed
  		if (sinkFunction instanceof InputTypeConfigurable) {
  			((InputTypeConfigurable) sinkFunction).setInputType(getType(), getExecutionConfig());
  		}
  
  		StreamSink<T> sinkOperator = new StreamSink<>(clean(sinkFunction));
  
  		DataStreamSink<T> sink = new DataStreamSink<>(this, sinkOperator);
  
  		getExecutionEnvironment().addOperator(sink.getTransformation());
  		return sink;
  	}