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
  public void testPollingGrowthTrackerUsesElementTimestampIfNoWatermarkProvided() throws Exception {
    Instant now = Instant.now();
    Watch.Growth<String, String, String> growth =
        Watch.growthOf(
                new Watch.Growth.PollFn<String, String>() {

                  @Override
                  public PollResult<String> apply(String element, Context c) throws Exception {
                    // We specifically test an unsorted list.
                    return PollResult.incomplete(
                        Arrays.asList(
                            TimestampedValue.of("d", now.plus(standardSeconds(4))),
                            TimestampedValue.of("c", now.plus(standardSeconds(3))),
                            TimestampedValue.of("a", now.plus(standardSeconds(1))),
                            TimestampedValue.of("b", now.plus(standardSeconds(2)))));
                  }
                })
            .withPollInterval(standardSeconds(10));
    WatchGrowthFn<String, String, String, Integer> growthFn =
        new WatchGrowthFn(
            growth, StringUtf8Coder.of(), SerializableFunctions.identity(), StringUtf8Coder.of());
    GrowthTracker<String, Integer> tracker = newPollingGrowthTracker();
    DoFn.ProcessContext context = mock(DoFn.ProcessContext.class);
    ManualWatermarkEstimator<Instant> watermarkEstimator =
        new WatermarkEstimators.Manual(BoundedWindow.TIMESTAMP_MIN_VALUE);
    ProcessContinuation processContinuation =
        growthFn.process(context, tracker, watermarkEstimator);
    assertEquals(now.plus(standardSeconds(1)), watermarkEstimator.currentWatermark());
    assertTrue(processContinuation.shouldResume());
  }