Flink mysql-cdc 2.2.1报bug

[fuyoucanghai]

bug是在使用cdc采集mysql数据表时产生的

一、环境：
Flink1.13.1
mysql5.7
flink的lib包如下：

二、bug是在sql-client里执行sql语句时报的错，所用Sql语句为：

Flink SQL> CREATE TABLE test_data (

id INT ,
name varchar,
PRIMARY KEY (id) NOT ENFORCED

) WITH (
'connector' = 'mysql-cdc',
'hostname' = '192.168.159.131',
'port' = '3306',
'username' = 'root',
'password' = '*******',
'database-name' = 'test_db',
'table-name' = 'test_1'
);
[INFO] Execute statement succeed.

Flink SQL>

CREATE TABLE kfk_sink_test (
id INT
,name string
,PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka'
,'topic' = 'kfk_sink_test7'
,'properties.bootstrap.servers' = 'gp102:9092'
,'key.format' = 'csv'
,'value.format' = 'csv'
);
[INFO] Execute statement succeed.

Flink SQL>

insert into kfk_sink_test select * from test_data;

报错信息如下：
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:201)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:161)
Caused by: java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/ObjectMapper
at com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory.createStringConverter(MySqlDeserializationConverterFactory.java:69)
at com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory.access$000(MySqlDeserializationConverterFactory.java:44)
at com.ververica.cdc.connectors.mysql.table.MySqlDeserializationConverterFactory$1.createUserDefinedConverter(MySqlDeserializationConverterFactory.java:57)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.createNotNullConverter(RowDataDebeziumDeserializeSchema.java:253)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.createConverter(RowDataDebeziumDeserializeSchema.java:237)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.lambda$createRowConverter$0(RowDataDebeziumDeserializeSchema.java:586)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546)
at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.createRowConverter(RowDataDebeziumDeserializeSchema.java:590)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.createNotNullConverter(RowDataDebeziumDeserializeSchema.java:319)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.createConverter(RowDataDebeziumDeserializeSchema.java:237)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema.(RowDataDebeziumDeserializeSchema.java:107)
at com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema$Builder.build(RowDataDebeziumDeserializeSchema.java:217)
at com.ververica.cdc.connectors.mysql.table.MySqlTableSource.getScanRuntimeProvider(MySqlTableSource.java:174)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.validateScanSource(DynamicSourceUtils.java:453)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.prepareDynamicSource(DynamicSourceUtils.java:161)
at org.apache.flink.table.planner.connectors.DynamicSourceUtils.convertSourceToRel(DynamicSourceUtils.java:119)
at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:85)
at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3585)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2507)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2144)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2093)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2050)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:663)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:644)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3438)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:570)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:170)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:162)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:967)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:936)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:275)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:595)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:268)
at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$parseStatement$1(LocalExecutor.java:176)
at org.apache.flink.table.client.gateway.context.ExecutionContext.wrapClassLoader(ExecutionContext.java:90)
at org.apache.flink.table.client.gateway.local.LocalExecutor.parseStatement(LocalExecutor.java:176)
at org.apache.flink.table.client.cli.CliClient.parseCommand(CliClient.java:385)
at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:326)
at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:297)
at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:221)
at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:151)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:95)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:187)
... 1 more
Caused by: java.lang.ClassNotFoundException: com.fasterxml.jackson.databind.ObjectMapper
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355)
at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
... 52 more

Shutting down the session...
done.

有遇到过这个bug的大佬吗。。。请问怎么解决的。感谢！