public static <
          K extends @Nullable Object,
          InputT extends @Nullable Object,
          AccumT extends @Nullable Object,
          OutputT extends @Nullable Object>
      MultiStepCombine<K, InputT, AccumT, OutputT> of(
          CombineFn<InputT, AccumT, OutputT> combineFn, Coder<KV<K, OutputT>> outputCoder) {
    return new MultiStepCombine<>(combineFn, outputCoder);
  }
@Test
  public void testMultiStepCombineWindowed() {
    SlidingWindows windowFn = SlidingWindows.of(Duration.millis(6L)).every(Duration.millis(3L));
    PCollection<KV<String, Long>> combined =
        pipeline
            .apply(
                Create.timestamped(
                    TimestampedValue.of(KV.of("foo", 1L), new Instant(1L)),
                    TimestampedValue.of(KV.of("bar", 2L), new Instant(2L)),
                    TimestampedValue.of(KV.of("bizzle", 3L), new Instant(3L)),
                    TimestampedValue.of(KV.of("bar", 4L), new Instant(4L)),
                    TimestampedValue.of(KV.of("bizzle", 11L), new Instant(11L))))
            .apply(Window.into(windowFn))
            .apply(Combine.perKey(new MultiStepCombineFn()));

    PAssert.that("Windows should combine only elements in their windows", combined)
        .inWindow(new IntervalWindow(new Instant(0L), Duration.millis(6L)))
        .containsInAnyOrder(KV.of("foo", 1L), KV.of("bar", 6L), KV.of("bizzle", 3L));
    PAssert.that("Elements should appear in all the windows they are assigned to", combined)
        .inWindow(new IntervalWindow(new Instant(-3L), Duration.millis(6L)))
        .containsInAnyOrder(KV.of("foo", 1L), KV.of("bar", 2L));
    PAssert.that(combined)
        .inWindow(new IntervalWindow(new Instant(6L), Duration.millis(6L)))
        .containsInAnyOrder(KV.of("bizzle", 11L));
    PAssert.that(combined)
        .containsInAnyOrder(
            KV.of("foo", 1L),
            KV.of("foo", 1L),
            KV.of("bar", 6L),
            KV.of("bar", 2L),
            KV.of("bar", 4L),
            KV.of("bizzle", 11L),
            KV.of("bizzle", 11L),
            KV.of("bizzle", 3L),
            KV.of("bizzle", 3L));
    pipeline.run();
  }