@Override
    public void deserialize(SourceRecord record, Collector<T> out) throws Exception {
        if (find) {
            serializer.deserialize(record, out);
            return;
        }

        if (filtered == 0) {
            LOG.info("Begin to seek binlog to the specific timestamp {}.", startupTimestampMillis);
        }

        Struct value = (Struct) record.value();
        Struct source = value.getStruct(Envelope.FieldName.SOURCE);
        Long ts = source.getInt64(Envelope.FieldName.TIMESTAMP);
        if (ts != null && ts >= startupTimestampMillis) {
            serializer.deserialize(record, out);
            find = true;
            LOG.info(
                    "Successfully seek to the specific timestamp {} with filtered {} change events.",
                    startupTimestampMillis,
                    filtered);
        } else {
            filtered++;
            if (filtered % 10000 == 0) {
                LOG.info(
                        "Seeking binlog to specific timestamp with filtered {} change events.",
                        filtered);
            }
        }
    }
A {@link DebeziumDeserializationSchema} which wraps a real {@link DebeziumDeserializationSchema}
to seek binlog to the specific timestamp.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/SeekBinlogToTimestampFilter.java