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

    String className = cloudUserFn.getClassName();
    ParDoFnFactory factory = defaultFactories.get(className);

    if (factory == null) {
      throw new Exception("No known ParDoFnFactory for " + className);
    }

    return factory.create(
        options,
        cloudUserFn,
        sideInputInfos,
        mainOutputTag,
        outputTupleTagsToReceiverIndices,
        executionContext,
        operationContext);
  }
@Test
  public void testCreateUnknownParDoFn() throws Exception {
    // A bogus serialized DoFn
    CloudObject cloudUserFn = CloudObject.forClassName("UnknownKindOfDoFn");
    try {
      DEFAULT_FACTORY.create(
          DEFAULT_OPTIONS,
          cloudUserFn,
          null,
          MAIN_OUTPUT,
          ImmutableMap.<TupleTag<?>, Integer>of(MAIN_OUTPUT, 0),
          DEFAULT_EXECUTION_CONTEXT,
          TestOperationContext.create(counterSet));
      fail("should have thrown an exception");
    } catch (Exception exn) {
      assertThat(exn.toString(), Matchers.containsString("No known ParDoFnFactory"));
    }
  }