@Override
    public SourceReader<T, MySqlSplit> createReader(SourceReaderContext readerContext)
            throws Exception {
        final Method metricGroupMethod = readerContext.getClass().getMethod("metricGroup");
        metricGroupMethod.setAccessible(true);
        final MetricGroup metricGroup = (MetricGroup) metricGroupMethod.invoke(readerContext);
        final MySqlSourceReaderMetrics sourceReaderMetrics =
                new MySqlSourceReaderMetrics(metricGroup);
        // create source config for the given subtask (e.g. unique server id)
        MySqlSourceConfig sourceConfig =
                configFactory.createConfig(readerContext.getIndexOfSubtask());
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(sourceConfig.getInlongMetric())
                .withAuditAddress(sourceConfig.getInlongAudit())
                .withAuditKeys(sourceConfig.getAuditKyes())
                .withRegisterMetric(RegisteredMetric.ALL)
                .build();
        sourceReaderMetrics.registerMetrics(metricOption);
        MySqlSourceReaderContext mySqlSourceReaderContext =
                new MySqlSourceReaderContext(readerContext);

        FutureCompletingBlockingQueue<RecordsWithSplitIds<SourceRecord>> elementsQueue =
                new FutureCompletingBlockingQueue<>();
        Supplier<MySqlSplitReader> splitReaderSupplier =
                () -> new MySqlSplitReader(
                        sourceConfig,
                        readerContext.getIndexOfSubtask(),
                        mySqlSourceReaderContext);
        return new MySqlSourceReader<>(
                elementsQueue,
                splitReaderSupplier,
                new MySqlRecordEmitter<>(
                        deserializationSchema,
                        sourceReaderMetrics,
                        sourceConfig),
                readerContext.getConfiguration(),
                mySqlSourceReaderContext,
                sourceConfig, sourceReaderMetrics);
    }
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