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 testSinglePollMultipleInputsWithSideInput() {
    final PCollectionView<String> moo =
        p.apply("moo", Create.of("moo")).apply("moo singleton", View.asSingleton());
    final PCollectionView<String> zoo =
        p.apply("zoo", Create.of("zoo")).apply("zoo singleton", View.asSingleton());
    PCollection<KV<String, String>> res =
        p.apply("input", Create.of("a", "b"))
            .apply(
                Watch.growthOf(
                        new PollFn<String, String>() {
                          @Override
                          public PollResult<String> apply(String element, Context c)
                              throws Exception {
                            return PollResult.complete(
                                Instant.now(),
                                Arrays.asList(
                                    element + " " + c.sideInput(moo) + " " + c.sideInput(zoo)));
                          }
                        },
                        requiresSideInputs(moo, zoo))
                    .withPollInterval(Duration.ZERO));

    PAssert.that(res)
        .containsInAnyOrder(Arrays.asList(KV.of("a", "a moo zoo"), KV.of("b", "b moo zoo")));

    p.run();
  }