MapFnRunners { public static <InputT, OutputT> PTransformRunnerFactory<?> forWindowedValueMapFnFactory( WindowedValueMapFnFactory<InputT, OutputT> fnFactory) { return new Factory<>(new ExplodedWindowedValueMapperFactory<>(fnFactory)); }  static PTransformRunnerFactory<?> forValueMapFnFactory(
      ValueMapFnFactory<InputT, OutputT> fnFactory); static PTransformRunnerFactory<?> forWindowedValueMapFnFactory(
      WindowedValueMapFnFactory<InputT, OutputT> fnFactory);  }
@Test public void testFullWindowedValueMapping() 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<>(); MapFnRunners.forWindowedValueMapFnFactory(this::createMapFunctionForPTransform) .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 testFullWindowedValueMappingWithCompressedWindow() throws Exception { List<WindowedValue<?>> outputConsumer = new ArrayList<>(); PCollectionConsumerRegistry consumers = new PCollectionConsumerRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class)); consumers.register("outputPC", "pTransformId", outputConsumer::add); PTransformFunctionRegistry startFunctionRegistry = new PTransformFunctionRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start"); PTransformFunctionRegistry finishFunctionRegistry = new PTransformFunctionRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish"); List<ThrowingRunnable> teardownFunctions = new ArrayList<>(); MapFnRunners.forWindowedValueMapFnFactory(this::createMapFunctionForPTransform) .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")); IntervalWindow firstWindow = new IntervalWindow(new Instant(0L), Duration.standardMinutes(10L)); IntervalWindow secondWindow = new IntervalWindow(new Instant(-10L), Duration.standardSeconds(22L)); consumers .getMultiplexingConsumer("inputPC") .accept( WindowedValue.of( "abc", new Instant(12), ImmutableSet.of(firstWindow, GlobalWindow.INSTANCE, secondWindow), PaneInfo.NO_FIRING)); assertThat( outputConsumer, containsInAnyOrder( WindowedValue.timestampedValueInGlobalWindow("ABC", new Instant(12)), WindowedValue.of("ABC", new Instant(12), secondWindow, PaneInfo.NO_FIRING), WindowedValue.of("ABC", new Instant(12), firstWindow, PaneInfo.NO_FIRING))); }
@Test public void testFullWindowedValueMapping() 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<>(); MapFnRunners.forWindowedValueMapFnFactory(this::createMapFunctionForPTransform) .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"))); }
@Test public void testFullWindowedValueMappingWithCompressedWindow() throws Exception { List<WindowedValue<?>> outputConsumer = new ArrayList<>(); PCollectionConsumerRegistry consumers = new PCollectionConsumerRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class)); consumers.register("outputPC", "pTransformId", outputConsumer::add); PTransformFunctionRegistry startFunctionRegistry = new PTransformFunctionRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start"); PTransformFunctionRegistry finishFunctionRegistry = new PTransformFunctionRegistry( mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish"); List<ThrowingRunnable> teardownFunctions = new ArrayList<>(); MapFnRunners.forWindowedValueMapFnFactory(this::createMapFunctionForPTransform) .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")); IntervalWindow firstWindow = new IntervalWindow(new Instant(0L), Duration.standardMinutes(10L)); IntervalWindow secondWindow = new IntervalWindow(new Instant(-10L), Duration.standardSeconds(22L)); consumers .getMultiplexingConsumer("inputPC") .accept( WindowedValue.of( "abc", new Instant(12), ImmutableSet.of(firstWindow, GlobalWindow.INSTANCE, secondWindow), PaneInfo.NO_FIRING)); assertThat( outputConsumer, containsInAnyOrder( WindowedValue.timestampedValueInGlobalWindow("ABC", new Instant(12)), WindowedValue.of("ABC", new Instant(12), secondWindow, PaneInfo.NO_FIRING), WindowedValue.of("ABC", new Instant(12), firstWindow, PaneInfo.NO_FIRING))); }