public static SourceRecord create(
            Map<String, ?> sourcePartition,
            String topic,
            String splitId,
            WatermarkKind watermarkKind,
            Offset watermark) {
        return new SourceRecord(
                sourcePartition,
                watermark.getOffset(),
                topic,
                SIGNAL_EVENT_KEY_SCHEMA,
                signalRecordKey(splitId),
                SIGNAL_EVENT_VALUE_SCHEMA,
                signalRecordValue(splitId, watermarkKind));
    }
Utility class to deal Watermark event.
Copy from com.ververica:flink-cdc-base:2.3.0.
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/wartermark/WatermarkEvent.java