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;
  }
@ 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
Creates an instance of SDC and adds to pool