private static List<SourceRecord> upsertBinlog(
            SourceRecord lowWatermarkEvent,
            SourceRecord highWatermarkEvent,
            Map<Struct, SourceRecord> snapshotRecordsWithKey,
            List<SourceRecord> binlogRecords,
            List<SourceRecord> snapshotRecordsWithoutKey) {
        // upsert binlog events to snapshot events of split
        if (!binlogRecords.isEmpty()) {
            for (SourceRecord binlog : binlogRecords) {
                Struct key = (Struct) binlog.key();
                Struct value = (Struct) binlog.value();
                if (value != null) {
                    Envelope.Operation operation =
                            Envelope.Operation.forCode(
                                    value.getString(Envelope.FieldName.OPERATION));
                    switch (operation) {
                        case CREATE:
                        case UPDATE:
                            Envelope envelope = Envelope.fromSchema(binlog.valueSchema());
                            Struct source = value.getStruct(Envelope.FieldName.SOURCE);
                            Struct after = value.getStruct(Envelope.FieldName.AFTER);
                            Instant fetchTs =
                                    Instant.ofEpochMilli(
                                            (Long) source.get(Envelope.FieldName.TIMESTAMP));
                            SourceRecord record =
                                    new SourceRecord(
                                            binlog.sourcePartition(),
                                            binlog.sourceOffset(),
                                            binlog.topic(),
                                            binlog.kafkaPartition(),
                                            binlog.keySchema(),
                                            binlog.key(),
                                            binlog.valueSchema(),
                                            envelope.read(after, source, fetchTs));
                            snapshotRecordsWithKey.put(key, record);
                            break;
                        case DELETE:
                            snapshotRecordsWithKey.remove(key);
                            break;
                        case READ:
                            throw new IllegalStateException(
                                    String.format(
                                            "Binlog record shouldn't use READ operation, the the record is %s.",
                                            binlog));
                        default:
                            break;
                    }
                }
            }
        }

        final List<SourceRecord> normalizedRecords = new ArrayList<>();
        normalizedRecords.add(lowWatermarkEvent);
        if (!snapshotRecordsWithoutKey.isEmpty()) {
            // for table without key, there is no need for binlog upsert
            // because highWatermark equals to lowWatermark
            normalizedRecords.addAll(formatMessageTimestamp(snapshotRecordsWithoutKey));
        } else {
            normalizedRecords.addAll(formatMessageTimestamp(snapshotRecordsWithKey.values()));
        }
        normalizedRecords.add(highWatermarkEvent);

        return normalizedRecords;
    }
Normalize the records of snapshot split which represents the split records state on high
watermark. data input: [low watermark event] [snapshot events ] [high watermark event]
[binlog events] [binlog-end event] data output: [low watermark event] [normalized events]
[high watermark event]
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/utils/RecordUtils.java