code: @SuppressWarnings({ "rawtypes", "unchecked" })
  	public static <T> SourceSupplier<T> create(Properties executionConfig, String pipelineName, SourceFilter<?> sourceFilter){
  		try {
  			Class<? extends SourceSupplier<?>> sourceSupplierClass;
  			if (executionConfig.containsKey(DStreamConstants.SOURCE_SUPPLIER + pipelineName)){
  				sourceSupplierClass = (Class<? extends SourceSupplier<?>>) Class
  						.forName(executionConfig.getProperty(DStreamConstants.SOURCE_SUPPLIER + pipelineName), false, Thread.currentThread().getContextClassLoader());
  			}
  			else {
  				sourceSupplierClass = UriSourceSupplier.class;
  			}
  
  			SourceSupplier sourceSupplier = ReflectionUtils.newInstance(sourceSupplierClass,
  					new Class[]{Properties.class, String.class}, new Object[]{executionConfig, pipelineName});
  			return sourceSupplier;
  		}
  		catch (Exception e) {
  			throw new IllegalStateException("Failure while creating SurceSupplier.", e);
  		}
  	}
docs: Factory method that constructs the instance of the SourceSupplier by using
  execution configuration properties and the name of the pipeline to which it
  is supplying sources to.
  See {@link DStreamConstants#SOURCE} and {@link DStreamConstants#SOURCE_SUPPLIER}
  properties.