private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(
            int version, int splitVersion, DataInputDeserializer in) throws IOException {
        List<TableId> alreadyProcessedTables = readTableIds(in);
        List<SnapshotSplit> remainingSplits = readSnapshotSplits(splitVersion, in);
        Map<String, SnapshotSplit> assignedSnapshotSplits =
                readAssignedSnapshotSplits(splitVersion, in);
        Map<String, Offset> finishedOffsets = readFinishedOffsets(splitVersion, in);
        boolean isAssignerFinished = in.readBoolean();
        List<TableId> remainingTableIds = readTableIds(in);
        boolean isTableIdCaseSensitive = in.readBoolean();
        final List<SchemalessSnapshotSplit> remainingSchemalessSplits = new ArrayList<>();
        final Map<String, SchemalessSnapshotSplit> assignedSchemalessSnapshotSplits =
                new HashMap<>();
        final Map<TableId, TableChanges.TableChange> tableSchemas = new HashMap<>();
        remainingSplits.forEach(
                split -> {
                    tableSchemas.putAll(split.getTableSchemas());
                    remainingSchemalessSplits.add(split.toSchemalessSnapshotSplit());
                });
        assignedSnapshotSplits
                .entrySet()
                .forEach(
                        entry -> {
                            tableSchemas.putAll(entry.getValue().getTableSchemas());
                            assignedSchemalessSnapshotSplits.put(
                                    entry.getKey(), entry.getValue().toSchemalessSnapshotSplit());
                        });
        if (version >= 4) {
            tableSchemas.putAll(readTableSchemas(splitVersion, in));
        }
        return new SnapshotPendingSplitsState(
                alreadyProcessedTables,
                remainingSchemalessSplits,
                assignedSchemalessSnapshotSplits,
                tableSchemas,
                finishedOffsets,
                isAssignerFinished,
                remainingTableIds,
                isTableIdCaseSensitive,
                true);
    }
The {@link SimpleVersionedSerializer Serializer} for the {@link PendingSplitsState}.
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/assigner/state/PendingSplitsStateSerializer.java