cluster-common/src/main/java/com/streamsets/pipeline/EmbeddedSDCPool.java
@SuppressWarnings("unchecked")
  protected EmbeddedSDC create() throws Exception {
    Utils.checkState(open, "Not open");
    final EmbeddedSDC embeddedSDC = new EmbeddedSDC();
    Object source;
    // post-batch runnable
    Object pipelineStartResult = BootstrapCluster.startPipeline(() -> LOG.debug("Batch completed"));
    source = pipelineStartResult.getClass().getDeclaredField("source").get(pipelineStartResult);
    if (source instanceof DSource) {
      long startTime = System.currentTimeMillis();
      long endTime = startTime;
      long diff = 0;
      Source actualSource = ((DSource) source).getSource();
      while (actualSource == null && diff < 60000) {
        Thread.sleep(100);
        actualSource = ((DSource) source).getSource();
        endTime = System.currentTimeMillis();
        diff = endTime - startTime;
      }
      if (actualSource == null) {
        throw new IllegalStateException("Actual source is null, pipeline may not have been initialized");
      }
      source = actualSource;
    }
    if (!(source instanceof ClusterSource)) {
        throw new IllegalArgumentException("Source is not of type ClusterSource: " + source.getClass().getName());
    }
    embeddedSDC.setSource((ClusterSource) source);
    embeddedSDC.setSparkProcessors(
        (List<Object>)pipelineStartResult.getClass().getDeclaredField("sparkProcessors").get(pipelineStartResult));
    return embeddedSDC;
  }
@SuppressWarnings("unchecked")
  protected EmbeddedSDC create() throws Exception {
    Utils.checkState(open, "Not open");
    final EmbeddedSDC embeddedSDC = new EmbeddedSDC();
    Object source;
    // post-batch runnable
    Object pipelineStartResult = BootstrapCluster.startPipeline(() -> LOG.debug("Batch completed"));
    source = pipelineStartResult.getClass().getDeclaredField("source").get(pipelineStartResult);
    if (source instanceof DSource) {
      long startTime = System.currentTimeMillis();
      long endTime = startTime;
      long diff = 0;
      Source actualSource = ((DSource) source).getSource();
      while (actualSource == null && diff < 60000) {
        Thread.sleep(100);
        actualSource = ((DSource) source).getSource();
        endTime = System.currentTimeMillis();
        diff = endTime - startTime;
      }
      if (actualSource == null) {
        throw new IllegalStateException("Actual source is null, pipeline may not have been initialized");
      }
      source = actualSource;
    }
    if (!(source instanceof ClusterSource)) {
        throw new IllegalArgumentException("Source is not of type ClusterSource: " + source.getClass().getName());
    }
    embeddedSDC.setSource((ClusterSource) source);
    embeddedSDC.setSparkProcessors(
        (List<Object>)pipelineStartResult.getClass().getDeclaredField("sparkProcessors").get(pipelineStartResult));
    return embeddedSDC;
  }
Creates an instance of SDC and adds to pool
@return EmbeddedSDC
@throws Exception