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 testMergingCustomWindowsWithoutCustomWindowTypes() {
    Instant startInstant = new Instant(0L);
    PCollection<KV<String, Integer>> inputCollection =
        pipeline.apply(
            Create.timestamped(
                TimestampedValue.of(KV.of("a", 1), startInstant.plus(Duration.standardSeconds(1))),
                TimestampedValue.of(KV.of("a", 2), startInstant.plus(Duration.standardSeconds(2))),
                TimestampedValue.of(KV.of("a", 3), startInstant.plus(Duration.standardSeconds(3))),
                TimestampedValue.of(KV.of("a", 4), startInstant.plus(Duration.standardSeconds(4))),
                TimestampedValue.of(
                    KV.of("a", 5), startInstant.plus(Duration.standardSeconds(5)))));
    PCollection<KV<String, Integer>> windowedCollection =
        inputCollection.apply(Window.into(new WindowOddEvenMergingBuckets<>()));
    PCollection<String> result =
        windowedCollection
            .apply(GroupByKey.create())
            .apply(
                ParDo.of(
                    new DoFn<KV<String, Iterable<Integer>>, String>() {
                      @ProcessElement
                      public void processElement(ProcessContext c, BoundedWindow window) {
                        List<Integer> elements = Lists.newArrayList();
                        c.element().getValue().forEach(elements::add);
                        Collections.sort(elements);
                        c.output(elements.toString());
                      }
                    }));
    PAssert.that("Wrong output collection", result).containsInAnyOrder("[2, 4]", "[1, 3, 5]");
    pipeline.run();
  }