@Override
    public final void collect(T record) {
        collect(record, TimestampAssigner.NO_TIMESTAMP);
    }
@Test
    void eventsAreBeforeWatermarks() {
        final CollectingDataOutput<Integer> dataOutput = new CollectingDataOutput<>();
        final SourceOutputWithWatermarks<Integer> out =
                createWithSameOutputs(
                        dataOutput,
                        new RecordTimestampAssigner<>(),
                        new TestWatermarkGenerator<>());

        out.collect(42, 12345L);

        assertThat(dataOutput.events)
                .contains(
                        new StreamRecord<>(42, 12345L),
                        new org.apache.flink.streaming.api.watermark.Watermark(12345L));
    }