[Bug] [mysqlcdc - kafka] cannot be cast to io.debezium.relational.DefaultValueConverter

[yanxingtong]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

dinky v1.1.0     flink 1.17.2
mysqlcdc kafka任务正常提交但是查看flink ui 发现任务状态异常，任务异常log如下：
----------------------------------------------------------------------------------------------
2025-02-19 09:37:54
org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'Source: MySQL CDC Source -> PartitionByPrimarykey -> Process -> Flat Map' (operator cbc357ccb763df2852fee8c4fc7d55f2).
	at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:600)
	at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$QuiesceableContext.failJob(RecreateOnResetOperatorCoordinator.java:242)
	at org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext.failJob(SourceCoordinatorContext.java:394)
	at org.apache.flink.runtime.source.coordinator.SourceCoordinator.start(SourceCoordinator.java:221)
	at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator$DeferrableCoordinator.resetAndStart(RecreateOnResetOperatorCoordinator.java:410)
	at org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator.lambda$resetToCheckpoint$7(RecreateOnResetOperatorCoordinator.java:155)
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
	at org.apache.flink.runtime.operators.coordination.ComponentClosingUtils.lambda$closeAsyncWithTimeout$0(ComponentClosingUtils.java:77)
	at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to discover captured tables for enumerator
	at org.apache.flink.cdc.connectors.mysql.source.MySqlSource.createEnumerator(MySqlSource.java:213)
	at org.apache.flink.runtime.source.coordinator.SourceCoordinator.start(SourceCoordinator.java:217)
	... 8 more
Caused by: java.lang.ClassCastException: io.debezium.connector.mysql.MySqlDefaultValueConverter cannot be cast to io.debezium.relational.DefaultValueConverter
	at io.debezium.relational.TableSchemaBuilder.<init>(TableSchemaBuilder.java:100)
	at io.debezium.connector.mysql.MySqlDatabaseSchema.<init>(MySqlDatabaseSchema.java:100)
	at org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils.createMySqlDatabaseSchema(DebeziumUtils.java:111)
	at org.apache.flink.cdc.connectors.mysql.schema.MySqlSchema.<init>(MySqlSchema.java:54)
	at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.createChunkSplitter(MySqlSnapshotSplitAssigner.java:598)
	at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.<init>(MySqlSnapshotSplitAssigner.java:168)
	at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlSnapshotSplitAssigner.<init>(MySqlSnapshotSplitAssigner.java:99)
	at org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlHybridSplitAssigner.<init>(MySqlHybridSplitAssigner.java:63)
	at org.apache.flink.cdc.connectors.mysql.source.MySqlSource.createEnumerator(MySqlSource.java:209)
	... 9 more
----------------------------------------------------------------------------------------------

What you expected to happen

貌似版本兼容问题，但是切换其他flink-sql-connector-mysql-cdc版本后无效，在配置参数已添加classloader.resolve-order: parent-first，但也没有作用

How to reproduce

MySQLCDC 整库到 Kafka
mysql v8.0 dinky v1.1.0 flink 1.17.2

Anything else

No response

Version

1.1.0(Archived)

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[github-actions]

Hello @yanxingtong, this issue is about CDC/CDCSOURCE, so I assign it to @aiwenmo. If you have any questions, you can comment and reply.

你好 @yanxingtong, 这个 issue 是关于 CDC/CDCSOURCE 的，所以我把它分配给了 @aiwenmo。如有任何问题，可以评论回复。

[yanxingtong]

我的customJar如下

[aiwenmo]

Hi. What is your sql?

[yanxingtong]

这是我的flinksql

EXECUTE CDCSOURCE cdc_upsert_kafka WITH (
 'connector' = 'mysql-cdc',
 'hostname' = '192.168.xx.xxx',
 'port' = '9008',
 'username' = 'root',
 'password' = 'xxx',
 'checkpoint' = '3000',
 'scan.startup.mode' = 'initial',
 'parallelism' = '1',
 'table-name' = 'test\.t_id_name',
 'sink.connector' = 'upsert-kafka',
 'sink.topic' = 'dinky_#{tableName}',
 'sink.timezone' = 'UTC',
 'sink.properties.bootstrap.servers' = '192.168.xx.xxx:xxxx',
 'debezium.skipped.operations'='d',
 'sink.key.format' = 'json',
 'sink.value.format' = 'json'
)

这是我的mysql ddl

CREATE TABLE `t_id_name` (
  `id` int NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

[yanxingtong]

@aiwenmo @aiwenmo

[github-actions]

Hello @, this issue has not been active for more than 30 days. This issue will be closed in 7 days if there is no response. If you have any questions, you can comment and reply.

你好 @, 这个 issue 30 天内没有活跃，7 天后将关闭，如需回复，可以评论回复。