public SourceSplitBase deserializeSplit(int version, byte[] serialized) throws IOException {
        final DataInputDeserializer in = new DataInputDeserializer(serialized);

        int splitKind = in.readInt();
        if (splitKind == SNAPSHOT_SPLIT_FLAG) {
            boolean useCatalogBeforeSchema = in.readBoolean();
            TableId tableId = TableId.parse(in.readUTF(), useCatalogBeforeSchema);
            String splitId = in.readUTF();
            RowType splitKeyType = (RowType) LogicalTypeParser.parse(in.readUTF());
            Object[] splitBoundaryStart = SerializerUtils.serializedStringToRow(in.readUTF());
            Object[] splitBoundaryEnd = SerializerUtils.serializedStringToRow(in.readUTF());
            Offset highWatermark = readOffsetPosition(version, in);
            Map<TableId, TableChange> tableSchemas = readTableSchemas(version, in);

            return new SnapshotSplit(
                    tableId,
                    splitId,
                    splitKeyType,
                    splitBoundaryStart,
                    splitBoundaryEnd,
                    highWatermark,
                    tableSchemas);
        } else if (splitKind == STREAM_SPLIT_FLAG) {
            String splitId = in.readUTF();
            // skip split Key Type
            in.readUTF();
            Offset startingOffset = readOffsetPosition(version, in);
            Offset endingOffset = readOffsetPosition(version, in);
            List<FinishedSnapshotSplitInfo> finishedSplitsInfo =
                    readFinishedSplitsInfo(version, in);
            Map<TableId, TableChange> tableChangeMap = readTableSchemas(version, in);
            int totalFinishedSplitSize = finishedSplitsInfo.size();
            if (version == 3) {
                totalFinishedSplitSize = in.readInt();
            }
            in.releaseArrays();
            return new StreamSplit(
                    splitId,
                    startingOffset,
                    endingOffset,
                    finishedSplitsInfo,
                    tableChangeMap,
                    totalFinishedSplitSize);
        } else if (splitKind == METRIC_SPLIT_FLAG) {
            long numBytesIn = 0L;
            long numRecordsIn = 0L;
            if (in.available() > 0) {
                numBytesIn = in.readLong();
                numRecordsIn = in.readLong();
            }
            Map<String, Long> readPhaseMetricMap = readReadPhaseMetric(in);
            Map<String, TableMetric> tableMetricMap = readTableMetrics(in);
            return new MetricSplit(numBytesIn, numRecordsIn, readPhaseMetricMap, tableMetricMap);
        } else {
            throw new IOException("Unknown split kind: " + splitKind);
        }
    }
A serializer for the {@link SourceSplitBase}.
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/split/SourceSplitSerializer.java