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 testMultiStepCombineTimestampCombiner() {
    TimestampCombiner combiner = TimestampCombiner.LATEST;
    PCollection<KV<String, Long>> combined =
        pipeline
            .apply(
                Create.timestamped(
                    TimestampedValue.of(KV.of("foo", 4L), new Instant(1L)),
                    TimestampedValue.of(KV.of("foo", 1L), new Instant(4L)),
                    TimestampedValue.of(KV.of("bazzle", 4L), new Instant(4L)),
                    TimestampedValue.of(KV.of("foo", 12L), new Instant(12L))))
            .apply(
                Window.<KV<String, Long>>into(FixedWindows.of(Duration.millis(5L)))
                    .withTimestampCombiner(combiner))
            .apply(Combine.perKey(new MultiStepCombineFn()));
    PCollection<KV<String, TimestampedValue<Long>>> reified =
        combined.apply(
            ParDo.of(
                new DoFn<KV<String, Long>, KV<String, TimestampedValue<Long>>>() {
                  @ProcessElement
                  public void reifyTimestamp(ProcessContext context) {
                    context.output(
                        KV.of(
                            context.element().getKey(),
                            TimestampedValue.of(
                                context.element().getValue(), context.timestamp())));
                  }
                }));

    PAssert.that(reified)
        .containsInAnyOrder(
            KV.of("foo", TimestampedValue.of(5L, new Instant(4L))),
            KV.of("bazzle", TimestampedValue.of(4L, new Instant(4L))),
            KV.of("foo", TimestampedValue.of(12L, new Instant(12L))));
    pipeline.run();
  }