@Override
    public byte[] serialize(SourceSplitBase split) throws IOException {
        if (split.isSnapshotSplit()) {
            final SnapshotSplit snapshotSplit = split.asSnapshotSplit();
            // optimization: the splits lazily cache their own serialized form
            if (snapshotSplit.serializedFormCache != null) {
                return snapshotSplit.serializedFormCache;
            }

            final DataOutputSerializer out = SERIALIZER_CACHE.get();
            out.writeInt(SNAPSHOT_SPLIT_FLAG);
            boolean useCatalogBeforeSchema = shouldUseCatalogBeforeSchema(snapshotSplit.getTableId());
            out.writeBoolean(useCatalogBeforeSchema);
            out.writeUTF(snapshotSplit.getTableId().toString());
            out.writeUTF(snapshotSplit.splitId());
            out.writeUTF(snapshotSplit.getSplitKeyType().asSerializableString());

            final Object[] splitStart = snapshotSplit.getSplitStart();
            final Object[] splitEnd = snapshotSplit.getSplitEnd();
            // rowToSerializedString deals null case
            out.writeUTF(SerializerUtils.rowToSerializedString(splitStart));
            out.writeUTF(SerializerUtils.rowToSerializedString(splitEnd));
            writeOffsetPosition(snapshotSplit.getHighWatermark(), out);
            writeTableSchemas(snapshotSplit.getTableSchemas(), out);
            final byte[] result = out.getCopyOfBuffer();
            out.clear();
            // optimization: cache the serialized from, so we avoid the byte work during repeated
            // serialization
            snapshotSplit.serializedFormCache = result;
            return result;
        } else if (split.isStreamSplit()) {
            final StreamSplit streamSplit = split.asStreamSplit();
            // optimization: the splits lazily cache their own serialized form
            if (streamSplit.serializedFormCache != null) {
                return streamSplit.serializedFormCache;
            }
            final DataOutputSerializer out = SERIALIZER_CACHE.get();
            out.writeInt(STREAM_SPLIT_FLAG);
            out.writeUTF(streamSplit.splitId());
            out.writeUTF("");
            writeOffsetPosition(streamSplit.getStartingOffset(), out);
            writeOffsetPosition(streamSplit.getEndingOffset(), out);
            writeFinishedSplitsInfo(streamSplit.getFinishedSnapshotSplitInfos(), out);
            writeTableSchemas(streamSplit.getTableSchemas(), out);
            out.writeInt(streamSplit.getTotalFinishedSplitSize());
            final byte[] result = out.getCopyOfBuffer();
            out.clear();
            // optimization: cache the serialized from, so we avoid the byte work during repeated
            // serialization
            streamSplit.serializedFormCache = result;
            return result;
        } else {
            final MetricSplit metricSplit = (MetricSplit) split;
            final DataOutputSerializer out = SERIALIZER_CACHE.get();
            out.writeInt(METRIC_SPLIT_FLAG);
            out.writeLong(metricSplit.getNumBytesIn());
            out.writeLong(metricSplit.getNumRecordsIn());
            writeReadPhaseMetric(metricSplit.getReadPhaseMetricMap(), out);
            writeTableMetrics(metricSplit.getTableMetricMap(), out);
            final byte[] result = out.getCopyOfBuffer();
            out.clear();
            return result;
        }
    }
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