private MultiStepCombine(
      CombineFn<InputT, AccumT, OutputT> combineFn, Coder<KV<K, OutputT>> outputCoder) {
    this.combineFn = combineFn;
    this.outputCoder = outputCoder;
  }
@Test
  public void testMultiStepCombine() {
    PCollection<KV<String, Long>> combined =
        pipeline
            .apply(
                Create.of(
                    KV.of("foo", 1L),
                    KV.of("bar", 2L),
                    KV.of("bizzle", 3L),
                    KV.of("bar", 4L),
                    KV.of("bizzle", 11L)))
            .apply(Combine.perKey(new MultiStepCombineFn()));

    PAssert.that(combined)
        .containsInAnyOrder(KV.of("foo", 1L), KV.of("bar", 6L), KV.of("bizzle", 14L));
    pipeline.run();
  }