docs: @PublicEvolving
  	public <OUT> DataStreamSource<OUT> createInput(InputFormat<OUT, ?> inputFormat, TypeInformation<OUT> typeInfo) {
  		DataStreamSource<OUT> source;
  
  		if (inputFormat instanceof FileInputFormat) {
  			@SuppressWarnings("unchecked")
  			FileInputFormat<OUT> format = (FileInputFormat<OUT>) inputFormat;
  
  			source = createFileInput(format, typeInfo, "Custom File source",
  					FileProcessingMode.PROCESS_ONCE, -1);
  		} else {
  			source = createInput(inputFormat, typeInfo, "Custom Source");
  		}
  		return source;
  	}