Towards Flink CDC 3.0

[leonardBang]

Motivation

Provide a user-friendly, end-to-end streaming ELT framework for Data integration users.

Solution

The design docs by Flink CDC maintainers as following:

• English design docs: Flink CDC 3.0 EN
• Chinese design docs: Flink CDC 3.0 ZH

Umbrella tasks for each module

[prepare and mis]

• Introduce basic modules for Flink CDC 3.0 #2610
• [Build] Bump Flink version to 1.18.0 #2670
• [build] Enable tests for CDC 3.0 modules #2671
• [3.0][prepare] Translate the design docs to English #2689

[module] flink-cdc-common

• [flink-cdc-common] Introduce Event related APIs #2624
• [flink-cdc-common] Introduce DataSource related APIs #2625
• [flink-cdc-common] Introduce DataSink related APIs #2626
• [flink-cdc-common] Introduce Connector Factory APIs #2640
• [flink-cdc-common] Introduce DataType for CDC 3.0 #2627
• [flink-cdc-common] Introduce Schema to represent schema of table or data collection #2628
• [flink-cdc-common] Introduce TableID to describe the identifier for data or collection #2629
• [flink-cdc-common] Add SchemaChangeEvent implementation #2642
• [flink-cdc-common] Add RecordData implementation #2644
• [flink-cdc-common] Introduce type structure for CDC types #2651
• [cdc-common] Introduce all ConfigOptions for CDC 3.0 #2668
• [cdc-common] Introduce CollectionFilter to filter collection from given TableId pattern #2669
• [cdc-common] Factory should provide ConText for Source/Sink initialization #2666
• [cdc-common] Introduce annotations for CDC 3.0 #2693
• [common] Support adding Column directly in schema builder #2739
• [common] Define pipeline options and schema change behaviors #2740
• [common] MetadataApplier should be serializable #2743
• [cdc-common] Provide binary record data and remove rowType from DataChangeEvent #2748

[module] flink-cdc-cli

• [flink-cdc-cli] Shell script bootstrapping Java CLI frontend #2606
• [flink-cdc-cli] Java implementation of the CLI frontend #2607

[module] flink-cdc-composer

• [flink-cdc-composer] Core interfaces exposed to CLI frontend #2608
• [flink-cdc-composer] Shared factory discovery utils for loading DataSource and DataSink by SPI #2616
• [flink-cdc-composer] Translate DataSource definition to Flink Source #2617
• [flink-cdc-composer] Translate DataSink definition to Flink sink #2618
• [flink-cdc-composer] Translate Route definitions to Flink function #2619
• [flink-cdc-composer] Connector JAR discovery utilities #2620
• [flink-cdc-composer] Flink job composer implementation #2609
• [3.0][cdc-composer] Introduce pipeline.name as flink job name #2769

[module] flink-cdc-runtime

• [flink-cdc-runtime] Provide schema evolution framework #2604
• [flink-cdc-runtime] Add SchemaOperator to send schema change event and block stream before schema changes finish #2633
• [flink-cdc-runtime] Add SchemaOperatorCoordinator to handle schema changes #2634
• [flink-cdc-runtime] Add schema evolution tests #2603
• [flink-cdc-runtime] Add DataSink API defined for Flink CDC 3.0 #2601
• [flink-cdc-runtime] Add new SinkWriterOperator to process FlushEvent #2605
• [3.0][cdc-common] Add tests for TypeSerializer s #2713
• [runtime] Introduce schema manager for tracking schema evolutions #2741
• [runtime] Copy SinkWriterOperator and its related classes from Flink #2742
• [runtime] Use CollectCoordinationResponse as a wrapper of CoordinationResponse #2744

[module] flink-cdc-dist

• [flink-cdc-dist] Build a uber JAR for all flink-cdc modules #2621
• [flink-cdc-dist] Collect all distro resources into a TAR #2622

[module] flink-cdc-connect/flink-cdc-pipeline-connectors

• [flink-cdc-pipeline-connectors] Add test implementation for DataSource #2643
• [CDC 3.0] Add test implementation for new DataSink API #2602
• [flink-cdc-pipeline-connectors] Add Implementation of DataSink In StarRocks #2645
• [flink-cdc-pipeline-connectors] Add Implementation of DataSink In Doris #2646
• [flink-cdc-pipeline-connectors] Add Implementation of DataSource in MySQL #2647

• Kafka Sink
• Paimon datalake:（optional in 3.0 @lvyanquan ）

[module] Docs for cdc 3.0 pipeline

• [docs] Re-fact current website organization for CDC 3.0 #2652
• [docs] Add Quick start for CDC 3.0 pipeline #2653
• [docs] Add tutorial for CDC 3.0 pipeline #2654
• [docs] Introduce core concepts for CDC 3.0 pipeline #2655
• [3.0][docs] Add documents for MySql pipeline connector #2827
• [3.0][docs] Add documents for StarRocks pipeline connector #2840
• [docs] introduces of Doris Pipeline Connector #2788

Shared Utilities

• Introduce JUnit 5 and AssertJ dependencies #2657

[GOODBOY008]

Hi @PatrickRen , I can collaborate with you on the subtasks of flink-cdc-composer.

[maver1ck]

Hi All,
Are we planning to bump Debezium as part of this task ?

[leonardBang]

Hi All, Are we planning to bump Debezium as part of this task ?

Hi, maverck, we don't have a plan to dump debezium version yet, but this should happen soon, currently we‘re busy on the 3.0 version whose position is the end-to-end steaming ELT framework.

[maver1ck]

Hi @leonardBang,
As Debezium 2.x is not fully compatible with 1.9.7 I think upgrade also needs major version bump in this repo.
So migration to 3.0 is great opportunity for this to happen.

[leonardBang]

Hi @leonardBang, As Debezium 2.x is not fully compatible with 1.9.7 I think upgrade also needs major version bump in this repo. So migration to 3.0 is great opportunity for this to happen.

Hey, maver1ck
As we plan to keep compatibility between 2.4.2 cdc connector and 3.0 cdc connector, the bump version work will be scheduled later, Bumping DBZ 2.x is a big but should to do work from my side.

[maver1ck]

Hi @leonardBang
Thank you for the answer. I went through documentation and I'm excited. I was waiting years for Flink based Kafka Connect alternative. You may have one of the first users here :)

[leonardBang]

Close this issue as all subtasks finished, will kickoff 3.1 version soon

[sunyaf]

sink type mysql error

Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "mysql" in the classpath.

Available factory classes are:

com.ververica.cdc.connectors.mysql.factory.MySqlDataSourceFactory
at com.ververica.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:60)
at com.ververica.cdc.composer.flink.FlinkPipelineComposer.createDataSink(FlinkPipelineComposer.java:144)
at com.ververica.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:108)
at com.ververica.cdc.cli.CliExecutor.run(CliExecutor.java:65)
at com.ververica.cdc.cli.CliFrontend.main(CliFrontend.java:62)

[Jiang-HT]

sink mysql demo error : Cannot find factory with identifier "mysql" in the classpath

[KiwiGeorge]

sink type mysql error

Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "mysql" in the classpath.

Available factory classes are:

com.ververica.cdc.connectors.mysql.factory.MySqlDataSourceFactory at com.ververica.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:60) at com.ververica.cdc.composer.flink.FlinkPipelineComposer.createDataSink(FlinkPipelineComposer.java:144) at com.ververica.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:108) at com.ververica.cdc.cli.CliExecutor.run(CliExecutor.java:65) at com.ververica.cdc.cli.CliFrontend.main(CliFrontend.java:62)

Have you finished this issue? I encountered this too

[lvyanquan]

sink mysql demo error : Cannot find factory with identifier "mysql" in the classpath

Hi @Jiang-HT @KiwiGeorge We don't support MySQL as data sink now, and current support data sinks are Doris and StarRocks.

[KiwiGeorge]

sink mysql demo error : Cannot find factory with identifier "mysql" in the classpath

Hi @Jiang-HT @KiwiGeorge We don't support MySQL as data sink now, and current support data sinks are Doris and StarRocks.

I see there is source code in the project, so... when will you support mysql as data sink?

[shreelatak]

Do you support Kafka as a data sink?
https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/kafka/#pipeline-connector-options per this, I see Kafka sink, but I seem to face the error:
Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "kafka" in the classpath.

Available factory classes are:

com.ververica.cdc.connectors.mysql.factory.MySqlDataSourceFactory
at com.ververica.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:60)
at com.ververica.cdc.composer.flink.FlinkPipelineComposer.createDataSink(FlinkPipelineComposer.java:144)
at com.ververica.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:108)
at com.ververica.cdc.cli.CliExecutor.run(CliExecutor.java:65)
at com.ververica.cdc.cli.CliFrontend.main(CliFrontend.java:62)

[yuxiqian]

Hi @shreelatak, Kafka pipeline connector is supported in Flink CDC 3.1. However, it's a breaking update and package names like com.ververica.cdc have been renamed to org.apache.flink.cdc, so it won't work with any previous versions. You may download latest version of Flink CDC here and Kafka connectors here.

[shreelatak]

Hello @yuxiqian , thank you so much for the revert! Do you know which supporting mysql-cdc connector I can find for Flink CDC version 3.1, and where?

My yaml file has the below contents:

source:
   type: mysql
   name: MySQL Source
   hostname: 127.0.0.1
   port: 3306
   username: ****
   password: ****
   tables: patient.\.*

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: PLAINTEXT://localhost:9092
  topic: quickstart_events
  

pipeline:
  name: MySQL to Kafka Pipeline
  parallelism: 2

With the previous Flink CDC version (3.0.1) and connector [flink-cdc-pipeline-connector-mysql-3.0.0.jar], I encountered no issues. Now since having upgraded the Flink CDC (to 3.1) I am facing :

Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "mysql" in the classpath.
Available factory classes are:

at org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:62)
at org.apache.flink.cdc.composer.flink.translator.DataSourceTranslator.translate(DataSourceTranslator.java:47)
at org.apache.flink.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:101)
at org.apache.flink.cdc.cli.CliExecutor.run(CliExecutor.java:71)
at org.apache.flink.cdc.cli.CliFrontend.main(CliFrontend.java:71)

[yuxiqian]

Connectors are now released under Group ID org.apache.flink. Here are all available connector jars: https://central.sonatype.com/search?q=g:org.apache.flink%20cdc&smo=true

P.S. It is preferred to share Flink CDC related issues on GitHub Discussions or Flink JIRA Kanban.

[wangrenjun-vs]

Do you support Kafka as a data sink? https://nightlies.apache.org/flink/flink-cdc-docs-master/docs/connectors/kafka/#pipeline-connector-options per this, I see Kafka sink, but I seem to face the error: Exception in thread "main" java.lang.RuntimeException: Cannot find factory with identifier "kafka" in the classpath.

Available factory classes are:

com.ververica.cdc.connectors.mysql.factory.MySqlDataSourceFactory at com.ververica.cdc.composer.utils.FactoryDiscoveryUtils.getFactoryByIdentifier(FactoryDiscoveryUtils.java:60) at com.ververica.cdc.composer.flink.FlinkPipelineComposer.createDataSink(FlinkPipelineComposer.java:144) at com.ververica.cdc.composer.flink.FlinkPipelineComposer.compose(FlinkPipelineComposer.java:108) at com.ververica.cdc.cli.CliExecutor.run(CliExecutor.java:65) at com.ververica.cdc.cli.CliFrontend.main(CliFrontend.java:62)

do you has solve this problem？ I encountered this too