static <T, W extends BoundedWindow>
      ThrowingFunction<KV<T, Iterable<W>>, KV<T, KV<Iterable<W>, Iterable<KV<W, Iterable<W>>>>>>
          createMapFunctionForPTransform(String ptransformId, PTransform ptransform)
              throws IOException {
    RunnerApi.FunctionSpec payload =
        RunnerApi.FunctionSpec.parseFrom(ptransform.getSpec().getPayload());

    WindowFn<?, W> windowFn =
        (WindowFn<?, W>) WindowingStrategyTranslation.windowFnFromProto(payload);
    return WindowMergingFnRunner.<T, W>create(windowFn)::mergeWindows;
  }
@Test
  public void testWindowMergingWithNonMergingWindowFn() throws Exception {
    ThrowingFunction<
            KV<Object, Iterable<BoundedWindow>>,
            KV<
                Object,
                KV<Iterable<BoundedWindow>, Iterable<KV<BoundedWindow, Iterable<BoundedWindow>>>>>>
        mapFunction =
            WindowMergingFnRunner.createMapFunctionForPTransform(
                "ptransformId", createMergeTransformForWindowFn(new GlobalWindows()));

    KV<Object, Iterable<BoundedWindow>> input =
        KV.of(
            "abc",
            ImmutableList.of(new IntervalWindow(Instant.now(), Duration.standardMinutes(1))));

    assertEquals(
        KV.of(input.getKey(), KV.of(input.getValue(), Collections.emptyList())),
        mapFunction.apply(input));
  }