private EmbeddedSDC fastForward(int id) throws Exception {
    if (id == sparkProcessorCount) {
      return null;
    }
    LOG.info("No SDC was found at ID: " + id + ". Fast-forwarding..");
    EmbeddedSDC sdc;
    // If there are not SDCs that are just idling, create a new one, else return one from the not started pool.
    if (notStarted.isEmpty()) {
      sdc = create();
    } else {
      sdc = getNotStartedSDC();
    }

    Class<?> clusterFunctionClass = Class.forName("com.streamsets.pipeline.cluster.ClusterFunctionImpl");
    Method getBatch = clusterFunctionClass.getMethod("getNextBatch", int.class, EmbeddedSDC.class);
    Method forward =
        clusterFunctionClass.getMethod("doForward", Iterator.class, int.class, EmbeddedSDC.class);
    sdc.getSource().put(Collections.emptyList());
    getBatch.invoke(null, 0, sdc);
    for (int i = 0; i <= id - 1; i++) {
      forward.invoke(null, Collections.emptyIterator(), i, sdc);
    }
    return sdc;
  }
private EmbeddedSDC fastForward(int id) throws Exception {
    if (id == sparkProcessorCount) {
      return null;
    }
    LOG.info("No SDC was found at ID: " + id + ". Fast-forwarding..");
    EmbeddedSDC sdc;
    
    if (notStarted.isEmpty()) {
      sdc = create();
    } else {
      sdc = getNotStartedSDC();
    }

    Class<?> clusterFunctionClass = Class.forName("com.streamsets.pipeline.cluster.ClusterFunctionImpl");
    Method getBatch = clusterFunctionClass.getMethod("getNextBatch", int.class, EmbeddedSDC.class);
    Method forward =
        clusterFunctionClass.getMethod("doForward", Iterator.class, int.class, EmbeddedSDC.class);
    sdc.getSource().put(Collections.emptyList());
    getBatch.invoke(null, 0, sdc);
    for (int i = 0; i <= id - 1; i++) {
      forward.invoke(null, Collections.emptyIterator(), i, sdc);
    }
    return sdc;
  }
private EmbeddedSDC fastForward ( int id ) throws Exception { if ( id == sparkProcessorCount ) { return null ; } LOG . info ( "No SDC was found at ID: " + id + ". Fast-forwarding.." ) ; EmbeddedSDC sdc ; // If there are not SDCs that are just idling, create a new one, else return one from the not started pool. if ( notStarted . isEmpty ( ) ) { sdc = create ( ) ; } else { sdc = getNotStartedSDC ( ) ; } Class < ? > clusterFunctionClass = Class . forName ( "com.streamsets.pipeline.cluster.ClusterFunctionImpl" ) ; Method getBatch = clusterFunctionClass . getMethod ( "getNextBatch" , int . class , EmbeddedSDC . class ) ; Method forward = clusterFunctionClass . getMethod ( "doForward" , Iterator . class , int . class , EmbeddedSDC . class ) ; sdc . getSource ( ) . put ( Collections . emptyList ( ) ) ; getBatch . invoke ( null , 0 , sdc ) ; for ( int i = 0 ; i <= id - 1 ; i ++ ) { forward . invoke ( null , Collections . emptyIterator ( ) , i , sdc ) ; } return sdc ; }
What is this?!
Spark could end up scheduling transformers to run on machines that may actually not have received the data, due to:
- Non-local reads: Spark Locality is time-bound, so if local executor where the data is located is not available,
Spark will tell another one to pick up the task.
- Shuffle in the transformer: After a shuffle, it is possible the next task passing the data to the next batch could
end up on a different executor.

In both these cases, it is possible that there is no SDC at that id. So we create a new SDC and fast-forward to
that id. Since a read can never happen from a fast-forwarded id, we read the fake batch we passed till there, and
then return the sdc so it is ready for writes/errors.
What is this?! Spark could end up scheduling transformers to run on machines that may actually not have received the data due to : - Non - local reads : Spark Locality is time - bound so if local executor where the data is located is not available Spark will tell another one to pick up the task . - Shuffle in the transformer : After a shuffle it is possible the next task passing the data to the next batch could end up on a different executor .