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

        out.collect(17);

        final Object event = dataOutput.events.get(0);
        assertThat(event).isInstanceOf(StreamRecord.class);
        assertThat(((StreamRecord<?>) event).getTimestamp())
                .isEqualTo(TimestampAssigner.NO_TIMESTAMP);
    }