@Override
    public void collectAndOverwriteTimestamp(OUT record, long timestamp) {
        setTimestamp(timestamp);
        output.collect(reuse.replace(record));
    }
@Test
    void testCollectAndOverwriteTimestamp() {
        List<StreamElement> list = new ArrayList<>();
        CollectorOutput<Integer> collectorOutput = new CollectorOutput<>(list);
        OutputCollector<Integer> collector = new OutputCollector<>(collectorOutput);
        collector.collectAndOverwriteTimestamp(1, 10L);
        collector.collectAndOverwriteTimestamp(2, 20L);
        assertThat(list).containsExactly(new StreamRecord<>(1, 10L), new StreamRecord<>(2, 20L));
    }