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
  @Category({ValidatesRunner.class, UsesCustomWindowMerging.class})
  public void testMergingCustomWindows() {
    Instant startInstant = new Instant(0L);
    PCollection<String> inputCollection =
        pipeline.apply(
            Create.timestamped(
                TimestampedValue.of("big", startInstant.plus(Duration.standardSeconds(10))),
                TimestampedValue.of("small1", startInstant.plus(Duration.standardSeconds(20))),
                // This one will be outside of bigWindow thus not merged
                TimestampedValue.of("small2", startInstant.plus(Duration.standardSeconds(39)))));
    PCollection<String> windowedCollection =
        inputCollection.apply(Window.into(new CustomWindowFn<>()));
    PCollection<Long> count =
        windowedCollection.apply(Combine.globally(Count.<String>combineFn()).withoutDefaults());
    // "small1" and "big" elements merged into bigWindow "small2" not merged
    // because timestamp is not in bigWindow
    PAssert.that("Wrong number of elements in output collection", count).containsInAnyOrder(2L, 1L);
    pipeline.run();
  }