@Override
    public SplitEnumerator<MySqlSplit, PendingSplitsState> restoreEnumerator(
            SplitEnumeratorContext<MySqlSplit> enumContext, PendingSplitsState checkpoint) {
        MySqlSourceConfig sourceConfig = configFactory.createConfig(0);

        final MySqlSplitAssigner splitAssigner;
        if (checkpoint instanceof HybridPendingSplitsState) {
            try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
                final List<TableId> capturedTables = discoverCapturedTables(jdbc, sourceConfig);
                SnapshotPendingSplitsState splitsState =
                        ((HybridPendingSplitsState) checkpoint).getSnapshotPendingSplits();
                List<TableId> tables = new ArrayList<>(splitsState.getAlreadyProcessedTables());
                tables.addAll(splitsState.getRemainingTables());
                LOG.info("Checkpoint tables: {}", Arrays.deepToString(tables.toArray()));
                tables.removeAll(capturedTables);
                if (tables.size() > 0) {
                    LOG.info("Debezium doesn't capture {} tables, remove them from checkpoint",
                            Arrays.deepToString(tables.toArray()));
                    splitsState.getRemainingTables().removeAll(tables);
                    splitsState.getAlreadyProcessedTables().addAll(tables);
                    List<MySqlSchemalessSnapshotSplit> snapshotSplits = splitsState.getRemainingSplits().stream()
                            .filter(it -> tables.contains(it.getTableId())).collect(Collectors.toList());
                    splitsState.getRemainingSplits().removeAll(snapshotSplits);
                    LOG.info("Remaining splits: {}", Arrays.deepToString(splitsState.getRemainingSplits().toArray()));
                }
            } catch (Exception e) {
                throw new FlinkRuntimeException("Failed to discover captured tables for enumerator", e);
            }

            splitAssigner =
                    new MySqlHybridSplitAssigner(
                            sourceConfig,
                            enumContext.currentParallelism(),
                            (HybridPendingSplitsState) checkpoint);
        } else if (checkpoint instanceof BinlogPendingSplitsState) {
            splitAssigner =
                    new MySqlBinlogSplitAssigner(
                            sourceConfig, (BinlogPendingSplitsState) checkpoint);
        } else {
            throw new UnsupportedOperationException(
                    "Unsupported restored PendingSplitsState: " + checkpoint);
        }
        return new MySqlSourceEnumerator(enumContext, sourceConfig, splitAssigner);
    }
Get a MySqlParallelSourceBuilder to build a {@link MySqlSource}.

@return a MySql parallel source builder.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/MySqlSource.java