public CustomTimestampPolicyWithLimitedDelay(
      SerializableFunction<KafkaRecord<K, V>, Instant> timestampFunction,
      Duration maxDelay,
      Optional<Instant> previousWatermark) {
    this.maxDelay = maxDelay;
    this.timestampFunction = timestampFunction;

    // 'previousWatermark' is not the same as maxEventTimestamp (e.g. it could have been in future).
    // Initialize it such that watermark before reading any event same as previousWatermark.
    maxEventTimestamp = previousWatermark.orElse(BoundedWindow.TIMESTAMP_MIN_VALUE).plus(maxDelay);
  }
@Test
  public void testCustomTimestampPolicyWithLimitedDelay() {
    // Verifies that max delay is applies appropriately for reporting watermark

    Duration maxDelay = Duration.standardSeconds(60);

    CustomTimestampPolicyWithLimitedDelay<String, String> policy =
        new CustomTimestampPolicyWithLimitedDelay<>(
            record -> new Instant(record.getTimestamp()), maxDelay, Optional.empty());

    Instant now = Instant.now();

    TimestampPolicy.PartitionContext ctx = mock(TimestampPolicy.PartitionContext.class);
    when(ctx.getMessageBacklog()).thenReturn(100L);
    when(ctx.getBacklogCheckTime()).thenReturn(now);

    assertThat(policy.getWatermark(ctx), is(BoundedWindow.TIMESTAMP_MIN_VALUE));

    // (1) Test simple case : watermark == max_timesatmp - max_delay

    List<Long> input =
        ImmutableList.of(
            -200_000L, -150_000L, -120_000L, -140_000L, -100_000L, // <<< Max timestamp
            -110_000L);
    assertThat(getTimestampsForRecords(policy, now, input), is(input));

    // Watermark should be max_timestamp - maxDelay
    assertThat(
        policy.getWatermark(ctx), is(now.minus(Duration.standardSeconds(100)).minus(maxDelay)));

    // (2) Verify future timestamps

    input =
        ImmutableList.of(
            -200_000L, -150_000L, -120_000L, -140_000L, 100_000L, // <<< timestamp is in future
            -100_000L, -110_000L);

    assertThat(getTimestampsForRecords(policy, now, input), is(input));

    // Watermark should be now - max_delay (backlog in context still non zero)
    assertThat(policy.getWatermark(ctx, now), is(now.minus(maxDelay)));

    // (3) Verify that Watermark advances when there is no backlog

    // advance current time by 5 minutes
    now = now.plus(Duration.standardMinutes(5));
    Instant backlogCheckTime = now.minus(Duration.standardSeconds(10));

    when(ctx.getMessageBacklog()).thenReturn(0L);
    when(ctx.getBacklogCheckTime()).thenReturn(backlogCheckTime);

    assertThat(policy.getWatermark(ctx, now), is(backlogCheckTime.minus(maxDelay)));
  }