public static <InputT, OutputT> Growth<InputT, OutputT, OutputT> growthOf(
      Growth.PollFn<InputT, OutputT> pollFn, Requirements requirements) {
    return new AutoValue_Watch_Growth.Builder<InputT, OutputT, OutputT>()
        .setTerminationPerInput(Growth.never())
        .setPollFn(Contextful.of(pollFn, requirements))
        // use null as a signal that this is the identity function and output coder can be
        // reused as key coder
        .setOutputKeyFn(null)
        .build();
  }
@Test
  @Category({NeedsRunner.class, UsesUnboundedSplittableParDo.class})
  public void testMultiplePollsWithKeyExtractor() {
    List<KV<Integer, String>> polls =
        Arrays.asList(
            KV.of(0, "0"),
            KV.of(10, "10"),
            KV.of(20, "20"),
            KV.of(30, "30"),
            KV.of(40, "40"),
            KV.of(40, "40.1"),
            KV.of(20, "20.1"),
            KV.of(50, "50"),
            KV.of(10, "10.1"),
            KV.of(10, "10.2"),
            KV.of(60, "60"),
            KV.of(70, "70"),
            KV.of(60, "60.1"),
            KV.of(80, "80"),
            KV.of(40, "40.2"),
            KV.of(90, "90"),
            KV.of(90, "90.1"));

    List<Integer> expected = Arrays.asList(0, 10, 20, 30, 40, 50, 60, 70, 80, 90);

    PCollection<Integer> res =
        p.apply(Create.of("a"))
            .apply(
                Watch.growthOf(
                        Contextful.of(
                            new TimedPollFn<String, KV<Integer, String>>(
                                polls,
                                standardSeconds(1) /* timeToOutputEverything */,
                                standardSeconds(3) /* timeToDeclareOutputFinal */,
                                standardSeconds(30) /* timeToFail */),
                            Requirements.empty()),
                        KV::getKey)
                    .withTerminationPerInput(Growth.afterTotalOf(standardSeconds(5)))
                    .withPollInterval(Duration.millis(100))
                    .withOutputCoder(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of()))
                    .withOutputKeyCoder(VarIntCoder.of()))
            .apply("Drop input", Values.create())
            .apply("Drop auxiliary string", Keys.create());

    PAssert.that(res).containsInAnyOrder(expected);

    p.run();
  }