MapFnRunners { public static <InputT, OutputT> PTransformRunnerFactory<?> forValueMapFnFactory( ValueMapFnFactory<InputT, OutputT> fnFactory) { return new Factory<>(new CompressedValueOnlyMapperFactory<>(fnFactory)); }  static PTransformRunnerFactory<?> forValueMapFnFactory(
      ValueMapFnFactory<InputT, OutputT> fnFactory); static PTransformRunnerFactory<?> forWindowedValueMapFnFactory(
      WindowedValueMapFnFactory<InputT, OutputT> fnFactory);  }
@Test public void testValueOnlyMapping() throws Exception { List<WindowedValue<?>> outputConsumer = new ArrayList<>(); MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap(); PCollectionConsumerRegistry consumers = new PCollectionConsumerRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class)); consumers.register("outputPC", EXPECTED_ID, outputConsumer::add); PTransformFunctionRegistry startFunctionRegistry = new PTransformFunctionRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class), "start"); PTransformFunctionRegistry finishFunctionRegistry = new PTransformFunctionRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class), "finish"); List<ThrowingRunnable> teardownFunctions = new ArrayList<>(); ValueMapFnFactory<String, String> factory = (ptId, pt) -> String::toUpperCase; MapFnRunners.forValueMapFnFactory(factory) .createRunnerForPTransform( PipelineOptionsFactory.create(), null , null , null , EXPECTED_ID, EXPECTED_PTRANSFORM, Suppliers.ofInstance("57L")::get, Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), consumers, startFunctionRegistry, finishFunctionRegistry, null , teardownFunctions::add, null , null , null ); assertThat(startFunctionRegistry.getFunctions(), empty()); assertThat(finishFunctionRegistry.getFunctions(), empty()); assertThat(teardownFunctions, empty()); assertThat(consumers.keySet(), containsInAnyOrder("inputPC", "outputPC")); consumers.getMultiplexingConsumer("inputPC").accept(valueInGlobalWindow("abc")); assertThat(outputConsumer, contains(valueInGlobalWindow("ABC"))); }
@Test public void testValueOnlyMapping() throws Exception { List<WindowedValue<?>> outputConsumer = new ArrayList<>(); MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap(); PCollectionConsumerRegistry consumers = new PCollectionConsumerRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class)); consumers.register("outputPC", EXPECTED_ID, outputConsumer::add); PTransformFunctionRegistry startFunctionRegistry = new PTransformFunctionRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class), "start"); PTransformFunctionRegistry finishFunctionRegistry = new PTransformFunctionRegistry( metricsContainerRegistry, mock(ExecutionStateTracker.class), "finish"); List<ThrowingRunnable> teardownFunctions = new ArrayList<>(); ValueMapFnFactory<String, String> factory = (ptId, pt) -> String::toUpperCase; MapFnRunners.forValueMapFnFactory(factory) .createRunnerForPTransform( PipelineOptionsFactory.create(), null , null , EXPECTED_ID, EXPECTED_PTRANSFORM, Suppliers.ofInstance("57L")::get, Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), consumers, startFunctionRegistry, finishFunctionRegistry, teardownFunctions::add, null ); assertThat(startFunctionRegistry.getFunctions(), empty()); assertThat(finishFunctionRegistry.getFunctions(), empty()); assertThat(teardownFunctions, empty()); assertThat(consumers.keySet(), containsInAnyOrder("inputPC", "outputPC")); consumers.getMultiplexingConsumer("inputPC").accept(valueInGlobalWindow("abc")); assertThat(outputConsumer, contains(valueInGlobalWindow("ABC"))); }