default FinishedSnapshotSplitInfo deserialize(byte[] serialized) {
        try {
            final DataInputDeserializer in = new DataInputDeserializer(serialized);
            String tableIdStr = in.readUTF();
            String splitId = in.readUTF();
            Object[] splitStart = serializedStringToRow(in.readUTF());
            Object[] splitEnd = serializedStringToRow(in.readUTF());
            OffsetFactory offsetFactory = (OffsetFactory) serializedStringToObject(in.readUTF());
            Offset highWatermark = readOffsetPosition(in);
            boolean useCatalogBeforeSchema = true;
            if (in.available() > 0) {
                useCatalogBeforeSchema = in.readBoolean();
            }
            TableId tableId = TableId.parse(tableIdStr, useCatalogBeforeSchema);
            in.releaseArrays();

            return new FinishedSnapshotSplitInfo(
                    tableId, splitId, splitStart, splitEnd, highWatermark, offsetFactory);
        } catch (IOException e) {
            throw new FlinkRuntimeException(e);
        }
    }
read {@link Offset} from input stream and write {@link Offset} to output stream.
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/offset/OffsetDeserializerSerializer.java