DefaultParDoFnFactory implements ParDoFnFactory { @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); } DefaultParDoFnFactory(); @Override ParDoFn create(
      PipelineOptions options,
      CloudObject cloudUserFn,
      List<SideInputInfo> sideInputInfos,
      TupleTag<?> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext<?> executionContext,
      DataflowOperationContext operationContext);  }
@Test public void testCreateSimpleParDoFn() throws Exception { String stringFieldValue = "some state"; long longFieldValue = 42L; TestDoFn fn = new TestDoFn(stringFieldValue, longFieldValue); String serializedFn = StringUtils.byteArrayToJsonString( SerializableUtils.serializeToByteArray( DoFnInfo.forFn( fn, WindowingStrategy.globalDefault(), null , null , new TupleTag<>("output") , DoFnSchemaInformation.create(), Collections.emptyMap()))); CloudObject cloudUserFn = CloudObject.forClassName("DoFn"); addString(cloudUserFn, "serialized_fn", serializedFn); ParDoFn parDoFn = DEFAULT_FACTORY.create( DEFAULT_OPTIONS, cloudUserFn, null, MAIN_OUTPUT, ImmutableMap.<TupleTag<?>, Integer>of(MAIN_OUTPUT, 0), DEFAULT_EXECUTION_CONTEXT, TestOperationContext.create(counterSet)); assertThat(parDoFn, instanceOf(SimpleParDoFn.class)); SimpleParDoFn simpleParDoFn = (SimpleParDoFn) parDoFn; parDoFn.startBundle(new OutputReceiver()); parDoFn.processElement(WindowedValue.valueInGlobalWindow("foo")); @SuppressWarnings("rawtypes") DoFnInfo doFnInfo = simpleParDoFn.getDoFnInfo(); DoFn innerDoFn = (TestDoFn) doFnInfo.getDoFn(); assertThat(innerDoFn, instanceOf(TestDoFn.class)); assertThat(doFnInfo.getWindowingStrategy().getWindowFn(), instanceOf(GlobalWindows.class)); assertThat(doFnInfo.getWindowingStrategy().getTrigger(), instanceOf(DefaultTrigger.class)); TestDoFn actualTestDoFn = (TestDoFn) innerDoFn; assertEquals(stringFieldValue, actualTestDoFn.stringField); assertEquals(longFieldValue, actualTestDoFn.longField); }
@Test public void testCreateUnknownParDoFn() throws Exception { 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")); } }