@Override
    public void collectAndOverwriteTimestamp(OUT outputRecord, long timestamp) {
        checkOutputKey(outputRecord);
        collector.collectAndOverwriteTimestamp(outputRecord, timestamp);
    }
@Test
    void testCollectAndOverwriteTimestamp() {
        TestingTimestampCollector.Builder<Integer> builder = TestingTimestampCollector.builder();
        CompletableFuture<Integer> consumeRecord = new CompletableFuture<>();
        CompletableFuture<Long> consumeTimeStamp = new CompletableFuture<>();
        builder.setCollectAndOverwriteTimestampConsumer(
                (data, timeStamp) -> {
                    consumeRecord.complete(data);
                    consumeTimeStamp.complete(timeStamp);
                });

        final int record = 1;
        final long timeStamp = 10L;
        KeyCheckedOutputCollector<Integer, Integer> collector =
                new KeyCheckedOutputCollector<>(builder.build(), (ignore) -> 1, () -> 1);
        collector.collectAndOverwriteTimestamp(record, timeStamp);
        assertThat(consumeRecord).isCompletedWithValue(record);
        assertThat(consumeTimeStamp).isCompletedWithValue(timeStamp);
    }