public static ComposeCombineFnBuilder compose() {
    return new ComposeCombineFnBuilder();
  }
@Test
  @Category({ValidatesRunner.class, UsesSideInputs.class})
  public void testComposedCombine() {
    p.getCoderRegistry().registerCoderForClass(UserString.class, UserStringCoder.of());

    PCollection<KV<String, KV<Integer, UserString>>> perKeyInput =
        p.apply(
            Create.timestamped(
                    Arrays.asList(
                        KV.of("a", KV.of(1, UserString.of("1"))),
                        KV.of("a", KV.of(1, UserString.of("1"))),
                        KV.of("a", KV.of(4, UserString.of("4"))),
                        KV.of("b", KV.of(1, UserString.of("1"))),
                        KV.of("b", KV.of(13, UserString.of("13")))),
                    Arrays.asList(0L, 4L, 7L, 10L, 16L))
                .withCoder(
                    KvCoder.of(
                        StringUtf8Coder.of(),
                        KvCoder.of(BigEndianIntegerCoder.of(), UserStringCoder.of()))));

    TupleTag<Integer> maxIntTag = new TupleTag<>();
    TupleTag<UserString> concatStringTag = new TupleTag<>();
    PCollection<KV<String, KV<Integer, String>>> combineGlobally =
        perKeyInput
            .apply(Values.create())
            .apply(
                Combine.globally(
                    CombineFns.compose()
                        .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag)
                        .with(new GetUserStringFunction(), new ConcatString(), concatStringTag)))
            .apply(WithKeys.of("global"))
            .apply(
                "ExtractGloballyResult",
                ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag)));

    PCollection<KV<String, KV<Integer, String>>> combinePerKey =
        perKeyInput
            .apply(
                Combine.perKey(
                    CombineFns.compose()
                        .with(new GetIntegerFunction(), Max.ofIntegers(), maxIntTag)
                        .with(new GetUserStringFunction(), new ConcatString(), concatStringTag)))
            .apply(
                "ExtractPerKeyResult", ParDo.of(new ExtractResultDoFn(maxIntTag, concatStringTag)));
    PAssert.that(combineGlobally).containsInAnyOrder(KV.of("global", KV.of(13, "111134")));
    PAssert.that(combinePerKey)
        .containsInAnyOrder(KV.of("a", KV.of(4, "114")), KV.of("b", KV.of(13, "113")));
    p.run();
  }