Flink-cdc 2.1.0 读取 MySQL 报错：com.ververica.cdc.connectors.mysql.table.StartupOptions; class invalid for deserialization

[WeiJinglun]

flink-cdc 2.1.0，读取 MySQL5.7.9
项目代码：

        Properties debeziumProperties = new Properties();
        debeziumProperties.put("snapshot.locking.mode", "none");// do not use lock

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(mysqlProperties.getProperty(MysqlConstants.MYSQL_HOST_KEY))
                .port(Integer.parseInt(mysqlProperties.getProperty(MysqlConstants.MYSQL_PORT_KEY)))
                // 监控设置的数据库
                .databaseList(mysqlProperties.getProperty(MysqlConstants.MYSQL_DATABASE_LIST_KEY))
                .tableList(mysqlProperties.getProperty(MysqlConstants.MYSQL_TABLE_LIST_KEY).split(","))
                .username(mysqlProperties.getProperty(MysqlConstants.MYSQL_USERNAME))
                .password(mysqlProperties.getProperty(MysqlConstants.MYSQL_PASSWORD))
                .deserializer(new JsonStringDeserializationSchema())
                .startupOptions(StartupOptions.latest())
                .debeziumProperties(debeziumProperties)
                .build();

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                .setParallelism(4)
                .print()
                .setParallelism(1);

报错信息：
Caused by: java.io.InvalidClassException: com.ververica.cdc.connectors.mysql.table.StartupOptions; class invalid for deserialization
at java.io.ObjectStreamClass$ExceptionInfo.newInvalidClassException(ObjectStreamClass.java:157) ~[?:1.8.0_172]
at java.io.ObjectStreamClass.checkDeserialize(ObjectStreamClass.java:862) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2041) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571) ~[?:1.8.0_172]
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2285) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2209) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2067) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571) ~[?:1.8.0_172]
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2285) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2209) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2067) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571) ~[?:1.8.0_172]
at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2285) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2209) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2067) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1571) ~[?:1.8.0_172]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:431) ~[?:1.8.0_172]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:600) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:587) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.util.SerializedValue.deserializeValue(SerializedValue.java:67) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder.create(OperatorCoordinatorHolder.java:431) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.(ExecutionJobVertex.java:211) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraph.attachJobGraph(DefaultExecutionGraph.java:792) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.executiongraph.DefaultExecutionGraphBuilder.buildGraph(DefaultExecutionGraphBuilder.java:196) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:107) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:342) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.SchedulerBase.(SchedulerBase.java:190) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.DefaultScheduler.(DefaultScheduler.java:122) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:132) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:110) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:340) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.JobMaster.(JobMaster.java:317) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:107) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) ~[flink-dist_2.11-1.13.2.jar:1.13.2]
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) ~[?:1.8.0_172]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_172]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_172]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_172]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[?:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_172]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_172]
at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_172]

[zengqinchris]

大佬，这个问题怎么解决的？

[wdssky]

是不是反序列化方法的问题，我flink1.16用的这个反序列化
.deserializer(new JsonDebeziumDeserializationSchema())

[zengqinchris]

我的用SQL方式，
SET execution.checkpointing.interval = 30s;
SET execution.checkpointing.mode = EXACTLY_ONCE;
SET state.checkpoints.num-retained = 3;
SET execution.checkpointing.externalized-checkpoint-retention = RETAIN_ON_CANCELLATION;
SET state.backend.incremental = true;
SET execution.savepoint.ignore-unclaimed-state = false;

DROP TABLE IF EXISTS datasource_info;
CREATE TABLE IF NOT EXISTS datasource_info (
id INT NOT NULL
,data_type STRING
,data_version STRING
,data_name STRING
,data_desc STRING
,link_json STRING
,data_json STRING
,status INT NOT NULL
,is_meta INT NOT NULL
,tenant_id INT NOT NULL
,data_type_code INT NOT NULL
,schema_name STRING
,is_deleted INT NOT NULL
,gmt_create TIMESTAMP
,gmt_modified TIMESTAMP
,create_user_id INT
,modify_user_id BIGINT
,ccname STRING
,PRIMARY KEY(id) NOT ENFORCED
)
WITH (
'connector' = 'mysql-cdc',
'hostname' = '172.21.129.219',
'port' = '3306',
'username' = '${219MysqlUser}',
'password' = '${219MysqlPass}',
'database-name' = 'data_called',
'table-name' = 'datasource_info',
'server-id' = '1',
'scan.startup.mode' = 'initial',
'server-time-zone' = 'Asia/Shanghai'
);

select * from datasource_info;