public static <T> Window<T> into(WindowFn<? super T, ?> fn) {
    try {
      fn.windowCoder().verifyDeterministic();
    } catch (NonDeterministicException e) {
      throw new IllegalArgumentException("Window coders must be deterministic.", e);
    }
    return Window.<T>configure().withWindowFn(fn);
  }
@Test
  public void testWindowIntoWindowFnAssign() {
    pipeline
        .apply(Create.of(1, 2, 3))
        .apply(
            Window.into(FixedWindows.of(Duration.standardMinutes(11L).plus(Duration.millis(1L)))));

    final AtomicBoolean foundAssign = new AtomicBoolean(false);
    pipeline.traverseTopologically(
        new PipelineVisitor.Defaults() {
          @Override
          public void visitPrimitiveTransform(TransformHierarchy.Node node) {
            if (node.getTransform() instanceof Window.Assign) {
              foundAssign.set(true);
            }
          }
        });
    assertThat(foundAssign.get(), is(true));
  }