private static Map<String, Long> readReadPhaseMetric(DataInputDeserializer in) throws IOException {
        Map<String, Long> readPhaseMetrics = new HashMap<>();
        if (in.available() > 0) {
            final int size = in.readInt();
            for (int i = 0; i < size; i++) {
                readPhaseMetrics.put(in.readUTF(), in.readLong());
            }
        }
        return readPhaseMetrics;
    }
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