@ 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