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

}