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 testSinglePollWithManyResults() {
    // More than the default 100 elements per checkpoint for direct runner.
    final long numResults = 3000;
    PCollection<KV<String, Integer>> res =
        p.apply(Create.of("a"))
            .apply(
                Watch.growthOf(
                        new PollFn<String, KV<String, Integer>>() {
                          @Override
                          public PollResult<KV<String, Integer>> apply(String element, Context c)
                              throws Exception {
                            String pollId = UUID.randomUUID().toString();
                            List<KV<String, Integer>> output = Lists.newArrayList();
                            for (int i = 0; i < numResults; ++i) {
                              output.add(KV.of(pollId, i));
                            }
                            return PollResult.complete(Instant.now(), output);
                          }
                        })
                    .withTerminationPerInput(Growth.afterTotalOf(standardSeconds(1)))
                    .withPollInterval(Duration.millis(1))
                    .withOutputCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of())))
            .apply("Drop input", Values.create());

    PAssert.that("Poll called only once", res.apply(Keys.create()))
        .satisfies(
            pollIds -> {
              assertEquals(1, Sets.newHashSet(pollIds).size());
              return null;
            });
    PAssert.that("Yields all expected results", res.apply("Drop poll id", Values.create()))
        .satisfies(
            input -> {
              assertEquals(
                  "Total number of results mismatches",
                  numResults,
                  Lists.newArrayList(input).size());
              assertEquals("Results are not unique", numResults, Sets.newHashSet(input).size());
              return null;
            });

    p.run();
  }