WindowMappingFnRunner { static <T, W1 extends BoundedWindow, W2 extends BoundedWindow> ThrowingFunction<KV<T, W1>, KV<T, W2>> createMapFunctionForPTransform( String ptransformId, PTransform pTransform) throws IOException { FunctionSpec windowMappingFnPayload = FunctionSpec.parseFrom(pTransform.getSpec().getPayload()); WindowMappingFn<W2> windowMappingFn = (WindowMappingFn<W2>) PCollectionViewTranslation.windowMappingFnFromProto(windowMappingFnPayload); return (KV<T, W1> input) -> KV.of(input.getKey(), windowMappingFn.getSideInputWindow(input.getValue())); }    }
@Test public void testWindowMapping() throws Exception { String pTransformId = "pTransformId"; SdkComponents components = SdkComponents.create(); components.registerEnvironment(Environments.createDockerEnvironment("java")); RunnerApi.FunctionSpec functionSpec = RunnerApi.FunctionSpec.newBuilder() .setUrn(WindowMappingFnRunner.URN) .setPayload( ParDoTranslation.translateWindowMappingFn( new GlobalWindows().getDefaultWindowMappingFn(), components) .toByteString()) .build(); RunnerApi.PTransform pTransform = RunnerApi.PTransform.newBuilder().setSpec(functionSpec).build(); ThrowingFunction<KV<Object, BoundedWindow>, KV<Object, BoundedWindow>> mapFunction = WindowMappingFnRunner.createMapFunctionForPTransform(pTransformId, pTransform); KV<Object, BoundedWindow> input = KV.of("abc", new IntervalWindow(Instant.now(), Duration.standardMinutes(1))); assertEquals(KV.of(input.getKey(), GlobalWindow.INSTANCE), mapFunction.apply(input)); }