cluster-common/src/main/java/com/streamsets/pipeline/EmbeddedSDCPool.java
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 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.