docs: @SuppressWarnings("unchecked")
  	public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) {
  
  		if (typeInfo == null) {
  			if (function instanceof ResultTypeQueryable) {
  				typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
  			} else {
  				try {
  					typeInfo = TypeExtractor.createTypeInfo(
  							SourceFunction.class,
  							function.getClass(), 0, null, null);
  				} catch (final InvalidTypesException e) {
  					typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
  				}
  			}
  		}
  
  		boolean isParallel = function instanceof ParallelSourceFunction;
  
  		clean(function);
  
  		final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);
  		return new DataStreamSource<>(this, typeInfo, sourceOperator, isParallel, sourceName);
  	}