static <T, W extends BoundedWindow> AssignWindowsRunner<T, W> create(
      WindowFn<? super T, W> windowFn) {
    // Safe contravariant cast
    WindowFn<T, W> typedWindowFn = (WindowFn<T, W>) windowFn;
    return new AssignWindowsRunner<>(typedWindowFn);
  }
@Test
  public void factoryCreatesFromJavaWindowFn() throws Exception {
    SdkComponents components = SdkComponents.create();
    components.registerEnvironment(Environments.createDockerEnvironment("java"));
    PTransform windowPTransform =
        PTransform.newBuilder()
            .putInputs("in", "input")
            .putOutputs("out", "output")
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.ASSIGN_WINDOWS_TRANSFORM_URN)
                    .setPayload(
                        WindowIntoPayload.newBuilder()
                            .setWindowFn(
                                WindowingStrategyTranslation.toProto(
                                    new TestWindowFn(), components))
                            .build()
                            .toByteString())
                    .build())
            .build();

    ThrowingFunction<WindowedValue<?>, WindowedValue<?>> fn =
        (ThrowingFunction) factory.forPTransform("transform", windowPTransform);

    assertThat(
        fn.apply(
            WindowedValue.of(
                22L,
                new Instant(5),
                new IntervalWindow(new Instant(0L), new Instant(20027L)),
                PaneInfo.ON_TIME_AND_ONLY_FIRING)),
        equalTo(
            WindowedValue.of(
                22L,
                new Instant(5),
                new TestWindowFn().assignWindow(new Instant(5)),
                PaneInfo.ON_TIME_AND_ONLY_FIRING)));
  }