@Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
    }
@Test
    void testWatermarkBeforeRecords() {
        final TestingWatermarkOutput output = new TestingWatermarkOutput();
        final BoundedOutOfOrdernessWatermarks<Object> watermarks =
                new BoundedOutOfOrdernessWatermarks<>(Duration.ofMillis(10));

        watermarks.onPeriodicEmit(output);

        assertThat(output.lastWatermark()).isNotNull();
        assertThat(output.lastWatermark().getTimestamp()).isEqualTo(Long.MIN_VALUE);
    }