How to achieve data synchronization from MySQL to MySQL using Flink CDC?

[imsuperli]

I want to achieve MySQL to MySQL synchronization through Flink CDC, but when I look at the official documentation, I find that it only supports MySQL as a source, not as a sink. After various searches, I found: https://github.com/apache/flink-connector-jdbc. But how should I use and configure it? Is flink-connector-jdbc used for Flink CDC?

cdc pipeline config:

source:
type: mysql
name: MySQL Source
hostname: 172.31.203.9
port: 3306
username: tjptdev
password: ""
tables: FlinkCdCSource.\.*
server-id: 5401-5404
scan.startup.mode: initial
  schema-change.enabled: true

sink:
  type: jdbc
name: MySQL Sink
hostname: 172.31.203.9
port: 3306
username: tjptdev
password: ""
  database: FlinkCdCSink

pipeline:
  name: MySQL to MySQL Pipeline
parallelism: 4

How to configure sink.type? I configured it as mysql, jdbc, and myql-cdc, and all reported the following errors:

[root@tjptdevinteg flink]# env JAVA_HOME=/data/server/flink/jdk-17.0.15+6 ./flink-cdc-3.4.0/bin/flink-cdc.sh piplelines/mysql2mysql0010.yaml
Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "jdbc" in the classpath.

Available factory classes are:

org.apache.flink.cdc.connectors.mysql.factory.MySqlDataSourceFactory
at org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:65)
at org.apache.flink.cdc.composer.flink.translator.DataSinkTranslator.createDataSink(DataSinkTranslator.java:67)
at org.apache.flink.cdc.composer.flink.FlinkPipelineComposer.translate(FlinkPipelineComposer.java:142)
at org.apache.flink.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:100)
at org.apache.flink.cdc.cli.CliExecutor.deployWithComposer(CliExecutor.java:101)
at org.apache.flink.cdc.cli.CliExecutor.run(CliExecutor.java:82)
at org.apache.flink.cdc.cli.CliFrontend.main(CliFrontend.java:81)

    Who can tell me how to do this? Thanks ！