CustomTimestampPolicyWithLimitedDelay extends TimestampPolicy<K, V> { public CustomTimestampPolicyWithLimitedDelay( SerializableFunction<KafkaRecord<K, V>, Instant> timestampFunction, Duration maxDelay, Optional<Instant> previousWatermark) { this.maxDelay = maxDelay; this.timestampFunction = timestampFunction; maxEventTimestamp = previousWatermark.orElse(BoundedWindow.TIMESTAMP_MIN_VALUE).plus(maxDelay); } CustomTimestampPolicyWithLimitedDelay(
      SerializableFunction<KafkaRecord<K, V>, Instant> timestampFunction,
      Duration maxDelay,
      Optional<Instant> previousWatermark); @Override Instant getTimestampForRecord(PartitionContext ctx, KafkaRecord<K, V> record); @Override Instant getWatermark(PartitionContext ctx);  }
@Test public void testCustomTimestampPolicyWithLimitedDelay() { 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)); List<Long> input = ImmutableList.of( -200_000L, -150_000L, -120_000L, -140_000L, -100_000L, -110_000L); assertThat(getTimestampsForRecords(policy, now, input), is(input)); assertThat( policy.getWatermark(ctx), is(now.minus(Duration.standardSeconds(100)).minus(maxDelay))); input = ImmutableList.of( -200_000L, -150_000L, -120_000L, -140_000L, 100_000L, -100_000L, -110_000L); assertThat(getTimestampsForRecords(policy, now, input), is(input)); assertThat(policy.getWatermark(ctx, now), is(now.minus(maxDelay))); 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))); }