class WindowMergingFnRunner {

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

}