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));
  }