@Override
  public ParDoFn create(
      PipelineOptions options,
      CloudObject cloudUserFn,
      @Nullable List<SideInputInfo> sideInputInfos,
      TupleTag<?> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext)
      throws Exception {

    DoFnInstanceManager instanceManager =
        fnCache.get(
            operationContext.nameContext().systemName(),
            () ->
                DoFnInstanceManagers.cloningPool(doFnExtractor.getDoFnInfo(cloudUserFn), options));

    DoFnInfo<?, ?> doFnInfo = instanceManager.peek();

    DataflowExecutionContext.DataflowStepContext stepContext =
        executionContext.getStepContext(operationContext);

    Iterable<PCollectionView<?>> sideInputViews = doFnInfo.getSideInputViews();
    SideInputReader sideInputReader =
        executionContext.getSideInputReader(sideInputInfos, sideInputViews, operationContext);

    if (doFnInfo.getDoFn() instanceof BatchStatefulParDoOverrides.BatchStatefulDoFn) {
      // HACK: BatchStatefulDoFn is a class from DataflowRunner's overrides
      // that just instructs the worker to execute it differently. This will
      // be replaced by metadata in the Runner API payload
      BatchStatefulParDoOverrides.BatchStatefulDoFn fn =
          (BatchStatefulParDoOverrides.BatchStatefulDoFn) doFnInfo.getDoFn();
      DoFn underlyingFn = fn.getUnderlyingDoFn();

      return new BatchModeUngroupingParDoFn(
          (BatchModeExecutionContext.StepContext) stepContext,
          new SimpleParDoFn(
              options,
              DoFnInstanceManagers.singleInstance(doFnInfo.withFn(underlyingFn)),
              sideInputReader,
              doFnInfo.getMainOutput(),
              outputTupleTagsToReceiverIndices,
              stepContext,
              operationContext,
              doFnInfo.getDoFnSchemaInformation(),
              doFnInfo.getSideInputMapping(),
              runnerFactory));

    } else if (doFnInfo.getDoFn() instanceof StreamingPCollectionViewWriterFn) {
      // HACK: StreamingPCollectionViewWriterFn is a class from
      // DataflowPipelineTranslator. Using the class as an indicator is a migration path
      // to simply having an indicator string.

      checkArgument(
          stepContext instanceof StreamingModeExecutionContext.StreamingModeStepContext,
          "stepContext must be a StreamingModeStepContext to use StreamingPCollectionViewWriterFn");
      DataflowRunner.StreamingPCollectionViewWriterFn<Object> writerFn =
          (StreamingPCollectionViewWriterFn<Object>) doFnInfo.getDoFn();
      return new StreamingPCollectionViewWriterParDoFn(
          (StreamingModeExecutionContext.StreamingModeStepContext) stepContext,
          writerFn.getView().getTagInternal(),
          writerFn.getDataCoder(),
          (Coder<BoundedWindow>) doFnInfo.getWindowingStrategy().getWindowFn().windowCoder());
    } else {
      return new SimpleParDoFn(
          options,
          instanceManager,
          sideInputReader,
          doFnInfo.getMainOutput(),
          outputTupleTagsToReceiverIndices,
          stepContext,
          operationContext,
          doFnInfo.getDoFnSchemaInformation(),
          doFnInfo.getSideInputMapping(),
          runnerFactory);
    }
  }
@Test
  public void testFactoryDoesNotReuseAfterAborted() throws Exception {
    PipelineOptions options = PipelineOptionsFactory.create();
    CounterSet counters = new CounterSet();
    TestDoFn initialFn = new TestDoFn(Collections.<TupleTag<String>>emptyList());
    CloudObject cloudObject = getCloudObject(initialFn);
    ParDoFn parDoFn =
        factory.create(
            options,
            cloudObject,
            null,
            MAIN_OUTPUT,
            ImmutableMap.<TupleTag<?>, Integer>of(MAIN_OUTPUT, 0),
            BatchModeExecutionContext.forTesting(options, "testStage"),
            TestOperationContext.create(counters));

    Receiver rcvr = new OutputReceiver();

    parDoFn.startBundle(rcvr);
    parDoFn.processElement(WindowedValue.valueInGlobalWindow("foo"));
    TestDoFn fn = (TestDoFn) ((SimpleParDoFn) parDoFn).getDoFnInfo().getDoFn();

    parDoFn.abort();
    assertThat(fn.state, equalTo(TestDoFn.State.TORN_DOWN));

    // The fn should not be torn down here
    ParDoFn secondParDoFn =
        factory.create(
            options,
            cloudObject.clone(),
            null,
            MAIN_OUTPUT,
            ImmutableMap.<TupleTag<?>, Integer>of(MAIN_OUTPUT, 0),
            BatchModeExecutionContext.forTesting(options, "testStage"),
            TestOperationContext.create(counters));

    secondParDoFn.startBundle(rcvr);
    secondParDoFn.processElement(WindowedValue.valueInGlobalWindow("foo"));
    TestDoFn secondFn = (TestDoFn) ((SimpleParDoFn) secondParDoFn).getDoFnInfo().getDoFn();

    assertThat(secondFn, not(theInstance(fn)));
    assertThat(fn.state, equalTo(TestDoFn.State.TORN_DOWN));
    assertThat(secondFn.state, equalTo(TestDoFn.State.PROCESSING));
  }