CDC Connectors for Apache Flink®
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
Go to file
ice f7dda3b67b
[docs][zh][hotfix] Remove redundant sections from mysql-cdc documentation (#2777)
This closes #2777.
3 days ago
.github [build][hotfix] Use the correct format for labeler.yml (#2758) 5 days ago
.idea [build] Add vcs.xml for IDEA (#2064) 5 months ago
docs [docs][zh][hotfix] Remove redundant sections from mysql-cdc documentation (#2777) 3 days ago
flink-cdc-cli [common] Introduce Configuration and related utilities (#2681) 2 weeks ago
flink-cdc-common [cdc-common] Introduce createFieldGetters method in SchemaUtils to build FieldGetters of given Schema. (#2762) 3 days ago
flink-cdc-composer [cdc-composer] use DataSinkWriterOperatorFactory to replace all SinkWriterOperatorFactory. (#2773) 3 days ago
flink-cdc-connect [mongodb][hotfix] Fix MongoDB's rewriteOutputBuffer to emit +I rather than +U (#2760) (#2760) 3 days ago
flink-cdc-dist [pom] Use ${revision} replace specific version. (#2623) 4 weeks ago
flink-cdc-e2e-tests [hotfix][vitess] Fix invalidate package name 1 week ago
flink-cdc-runtime [cdc-common] Introduce createFieldGetters method in SchemaUtils to build FieldGetters of given Schema. (#2762) 3 days ago
tools [build] Add header copyright check for java files 2 weeks ago
.gitignore [pom] Use ${revision} replace specific version. (#2623) 4 weeks ago
LICENSE [build] Update the copyright year to 2023 (#2205) 5 months ago
NOTICE [build] Update the copyright year to 2023 (#2205) 5 months ago [docs] Remove flink 1.13 from supported flink versions 3 weeks ago
azure-pipelines.yml [oceanbase][ci] Move the OceanBase CI to free azure pipeline (#2506) 2 months ago
pom.xml [build] Add header copyright check for java files 2 weeks ago

CDC Connectors for Apache Flink®

CDC Connectors for Apache Flink® is a set of source connectors for Apache Flink®, ingesting changes from different databases using change data capture (CDC). CDC Connectors for Apache Flink® integrates Debezium as the engine to capture data changes. So it can fully leverage the ability of Debezium. See more about what is Debezium.

This README is meant as a brief walkthrough on the core features of CDC Connectors for Apache Flink®. For a fully detailed documentation, please see Documentation.

Supported (Tested) Databases

Connector Database Driver
  • MongoDB: 3.6, 4.x, 5.0, 6.0
  • MongoDB Driver: 4.3.4
  • MySQL: 5.6, 5.7, 8.0.x
  • RDS MySQL: 5.6, 5.7, 8.0.x
  • PolarDB MySQL: 5.6, 5.7, 8.0.x
  • Aurora MySQL: 5.6, 5.7, 8.0.x
  • MariaDB: 10.x
  • PolarDB X: 2.0.1
  • JDBC Driver: 8.0.28
  • OceanBase CE: 3.1.x, 4.x
  • OceanBase EE: 2.x, 3.x, 4.x
  • OceanBase Driver: 2.4.x
  • Oracle: 11, 12, 19, 21
  • Oracle Driver:
  • PostgreSQL: 9.6, 10, 11, 12, 13, 14
  • JDBC Driver: 42.5.1
  • Sqlserver: 2012, 2014, 2016, 2017, 2019
  • JDBC Driver: 9.4.1.jre8
  • TiDB: 5.1.x, 5.2.x, 5.3.x, 5.4.x, 6.0.0
  • JDBC Driver: 8.0.27
  • Db2: 11.5
  • Db2 Driver:
  • Vitess: 8.0.x, 9.0.x
  • MySql JDBC Driver: 8.0.26


    1. Supports reading database snapshot and continues to read transaction logs with exactly-once processing even failures happen.
    2. CDC connectors for DataStream API, users can consume changes on multiple databases and tables in a single job without Debezium and Kafka deployed.
    3. CDC connectors for Table/SQL API, users can use SQL DDL to create a CDC source to monitor changes on a single table.

    Usage for Table/SQL API

    We need several steps to setup a Flink cluster with the provided connector.

    1. Setup a Flink cluster with version 1.14+ and Java 8+ installed.
    2. Download the connector SQL jars from the Download page (or build yourself).
    3. Put the downloaded jars under FLINK_HOME/lib/.
    4. Restart the Flink cluster.

    The example shows how to create a MySQL CDC source in Flink SQL Client and execute queries on it.

    -- creates a mysql cdc table source
    CREATE TABLE mysql_binlog (
     id INT NOT NULL,
     name STRING,
     description STRING,
     weight DECIMAL(10,3)
    ) WITH (
     'connector' = 'mysql-cdc',
     'hostname' = 'localhost',
     'port' = '3306',
     'username' = 'flinkuser',
     'password' = 'flinkpw',
     'database-name' = 'inventory',
     'table-name' = 'products'
    -- read snapshot and binlog data from mysql, and do some transformation, and show on the client
    SELECT id, UPPER(name), description, weight FROM mysql_binlog;

    Usage for DataStream API

    Include following Maven dependency (available through Maven Central):

      <!-- add the dependency matching your database -->
      <!-- The dependency is available only for stable releases, SNAPSHOT dependencies need to be built based on master or release- branches by yourself. -->
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
    import com.ververica.cdc.connectors.mysql.source.MySqlSource;
    public class MySqlSourceExample {
      public static void main(String[] args) throws Exception {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .databaseList("yourDatabaseName") // set captured database
                .tableList("yourDatabaseName.yourTableName") // set captured table
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // enable checkpoint
          .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
          // set 4 parallel source tasks
          .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
        env.execute("Print MySQL Snapshot + Binlog");

    Building from source

    • Prerequisites:
      • git
      • Maven
      • At least Java 8
    git clone
    cd flink-cdc-connectors
    mvn clean install -DskipTests

    The dependencies are now available in your local .m2 repository.


    The code in this repository is licensed under the Apache Software License 2.


    CDC Connectors for Apache Flink® welcomes anyone that wants to help out in any way, whether that includes reporting problems, helping with documentation, or contributing code changes to fix bugs, add tests, or implement new features. You can report problems to request features in the GitHub Issues.

    Code Contribute

    1. Left comment under the issue that you want to take
    2. Fork Flink CDC project to your GitHub repositories
    3. Clone and compile your Flink CDC project
    git clone
    cd flink-cdc-connectors
    mvn clean install -DskipTests
    1. Check to a new branch and start your work
    git checkout -b my_feature
    -- develop and commit
    1. Push your branch to your github
    git push origin my_feature
    1. Open a PR to

    Code Style

    Code Formatting

    You need to install the google-java-format plugin. Spotless together with google-java-format is used to format the codes.

    It is recommended to automatically format your code by applying the following settings:

    1. Go to "Settings" → "Other Settings" → "google-java-format Settings".
    2. Tick the checkbox to enable the plugin.
    3. Change the code style to "Android Open Source Project (AOSP) style".
    4. Go to "Settings" → "Tools" → "Actions on Save".
    5. Under "Formatting Actions", select "Optimize imports" and "Reformat file".
    6. From the "All file types list" next to "Reformat code", select "Java".

    For earlier IntelliJ IDEA versions, the step 4 to 7 will be changed as follows.

    • 4.Go to "Settings" → "Other Settings" → "Save Actions".
    • 5.Under "General", enable your preferred settings for when to format the code, e.g. "Activate save actions on save".
    • 6.Under "Formatting Actions", select "Optimize imports" and "Reformat file".
    • 7.Under "File Path Inclusions", add an entry for .*\.java to avoid formatting other file types. Then the whole project could be formatted by command mvn spotless:apply.


    Checkstyle is used to enforce static coding guidelines.

    1. Go to "Settings" → "Tools" → "Checkstyle".
    2. Set "Scan Scope" to "Only Java sources (including tests)".
    3. For "Checkstyle Version" select "8.14".
    4. Under "Configuration File" click the "+" icon to add a new configuration.
    5. Set "Description" to "Flink cdc".
    6. Select "Use a local Checkstyle file" and link it to the file tools/maven/checkstyle.xml which is located within your cloned repository.
    7. Select "Store relative to project location" and click "Next".
    8. Configure the property checkstyle.suppressions.file with the value suppressions.xml and click "Next".
    9. Click "Finish".
    10. Select "Flink cdc" as the only active configuration file and click "Apply".

    You can now import the Checkstyle configuration for the Java code formatter.

    1. Go to "Settings" → "Editor" → "Code Style" → "Java".
    2. Click the gear icon next to "Scheme" and select "Import Scheme" → "Checkstyle Configuration".
    3. Navigate to and select tools/maven/checkstyle.xml located within your cloned repository.

    Then you could click "View" → "Tool Windows" → "Checkstyle" and find the "Check Module" button in the opened tool window to validate checkstyle. Or you can use the command mvn clean compile checkstyle:checkstyle to validate.

    Documentation Contribute

    Flink cdc documentations locates at docs/content.

    The contribution step is the same as the code contribution. We use markdown as the source code of the document.


    • DingTalk Chinese User Group

      You can search the group number [33121212] or scan the following QR code to join in the group.


    To get started, please see