Flink-JDBC SQL Connector报错: java.lang.Integer cannot be cast to java.lang.Long

一、基础信息

1.1 组件版本

• Flink:    1.13.0
• JDK：  1.8
• Mysql:  8.0.15
• com.ververica:flink-connector-mysql-cdc:2.0.0
• org.apache.flink:flink-connector-jdbc_2.11:1.13.0

1.2 建表语句

1.2.1 Mysql中的建表语句

-- 新建表demo
CREATE TABLE demo (
  sid int(6) ,
  name varchar(255)
);
 
-- 插入两条测试数据
insert into demo values(1,'hamawhite'),(2,'song.bs');

1.2.2 FlinkSQL中的建表语句

Mysql中字段sid定义的是int(6)类型，在FlinkSQL中定义的sid字段是BIGINT类型。

-- 新建JDBC SQL表, 字段sid定义为BIGINT类型
DROP TABLE IF EXISTS flink_demo_jdbc_bigint;
CREATE TABLE flink_demo_jdbc_bigint (
    sid BIGINT, 
    name STRING
)WITH
  (
    'connector'='jdbc',
    'url'='jdbc:mysql://192.168.90.150:3306/demo',
    'table-name'='demo',
    'username'='root',
    'password'='xxx'
  );

二、测试操作

运行SQL:

select * from flink_demo_jdbc_bigint limit 10

会遇到如下错误: 

java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Long
	at org.apache.flink.table.data.GenericRowData.getLong(GenericRowData.java:154)
	at org.apache.flink.table.data.RowData.lambda$createFieldGetter$245ca7d1$7(RowData.java:249)
	at org.apache.flink.table.data.RowData.lambda$createFieldGetter$25774257$1(RowData.java:296)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:170)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:131)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:48)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:69)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46)
	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26)
	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50)
	at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollect(StreamSourceContexts.java:317)
	at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collect(StreamSourceContexts.java:411)
	at org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:92)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)

flink_demo_jdbc_bigint表中只有sid字段是BIGINT类型，因此上面错误是针对sid字段的。

三、原因分析

3.1 分析GenericRowData类

 查看Flink源码GenericRowData类的getLong(int pos) 如下:

根据错误提示是因为Integer转Long出的错，但是看上述代码this.fields[pos]应该返回的是Integer类型，用基本数据类型long强转应该是可以的。

3.2 测试Java基本类型转换

3.2.1 测试代码1

例如下面的测试代码，能正常运行得出结果:

 3.2.2 测试代码2

但是flink中的(long) this.fields[pos]却出错，继续下面的测试代码:

至此，复现错误。

3.2.3 代码分析

对比上面两小节的代码，差异点如下:

1. 小节3.2.1的代码输入变量input在编译时，已经知道是Integer类型。

2. 小节3.2.2的代码输入变量input是根据key从JSONObject获取的value，在编译阶段其实不知道value的类型。

经过上述分析，开始怀疑编译后的源码有差异。

Test1.java编译后的代码如下:

 Test2.java编译后的代码如下:

可以看到Test2.java中把 long result = (long) input 经过编译后是long result = (Long)input，编译的时候把(long)改为(Long)。而Test1.class中还是跟源码一样，仍然是 long result = (long)input。

3.3 查看Flink编译后的代码

查看Flink中GenericRowData类的getLong(int pos) 编译后的代码。

 跟3.2节的测试结果一样，在编译阶段也是把(long)编译为(Long)，这样读出来Integer是没法转换为Long的，两个都继承自Number，但两者间是没有关系的。

四、修改方法

4.1 修改业务代码

由于在Mysql中sid字段是int(6)类型的，那么在定义Flink表的时候就用Integer，如下SQL:

-- 新建JDBC SQL表, 字段sid定义为INT类型
DROP TABLE IF EXISTS flink_demo_jdbc_int;
CREATE TABLE flink_demo_jdbc_int (
    sid INT, 
    name STRING
)WITH
  (
    'connector'='jdbc',
    'url'='jdbc:mysql://192.168.90.150:3306/demo',
    'table-name'='demo',
    'username'='root',
    'password'='xxx'
  );
 
 
-- 测试SQL
select * from flink_demo_jdbc_int limit 10

上面代码是能正确运行的。

4.2 修改Flink源码中的GenericRowData类

推荐修改Flink的GenericRowData类，把(long)改为(Long)避免给人造成迷惑。修改后的代码如下:

    @Override
    public long getLong(int pos) {
        return (Long) this.fields[pos];
    }

五、Mysql-CDC vs JDBC Connector

5.1 测试SQL 

上面测试的是jdbc connector，再测试下 mysql-cdc connector。测试SQL:

-- 新建Mysql CDC表, 字段sid定义为INT类型
DROP TABLE IF EXISTS flink_demo_cdc_int;
CREATE TABLE flink_demo_cdc_int (
    sid INT, 
    name STRING
)WITH
  (
    'connector'='mysql-cdc',
    'hostname'='192.168.90.150',
    'port'='3306',
    'username'='root',
    'password'='xxx',
    'database-name'='demo',
    'table-name'='demo',
    'scan.startup.mode'='initial',
    'scan.incremental.snapshot.enabled'='false'
  );
 
-- 新建Mysql CDC表, 字段sid定义为BIGINT类型
DROP TABLE IF EXISTS flink_demo_cdc_bigint;
CREATE TABLE flink_demo_cdc_bigint (
    sid BIGINT, 
    name STRING
)WITH
  (
    'connector'='mysql-cdc',
    'hostname'='192.168.90.150',
    'port'='3306',
    'username'='root',
    'password'='xxx',
    'database-name'='demo',
    'table-name'='demo',
    'scan.startup.mode'='initial',
    'scan.incremental.snapshot.enabled'='false'
  );
 
-- 测试SQL
select * from flink_demo_cdc_int limit 10;
 
select * from flink_demo_cdc_bigint limit 10;

上面不管sid是定义的INT，还是BIGINT类型，都是能正常读取数据的。

因此，jdbc connector和mysql-cdc connector是有差异的。

5.2 分析结果

在的Flink的GenericRowData.java的void setField(int pos, Object value) 方法中增加断点，开始调试。

5.2.1 JDBC Connector的结果

在org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter的RowData toInternal(ResultSet resultSet)的方法中只是把ResultSet中的类型setField进去，则sid塞进去的还是Integer类型。

那读取的时候执行的是(Long) this.fields[pos]，就会遇到Integer转Long的错误。

5.2.2 Mysql CDC Connector的结果

在com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema的RowData createRowConverter(RowType rowType)的方法中会把读出来的Integer类型根据FlinkSQL中定义的类型进行转换。

针对sid INT转换后还是Integer类型，sid BIGINT读取出来的则是BIGINT类型。

可以，由于mysql-cdc connector在读取数据塞进GenericRowData已经按照定义进行了类型转换，那么读取的时候则不会出现问题的。即针对mysql-cdc connector，mysql表中定义的是int(6)类型，在Flink SQL中不管定义的是INT还是BIGINT类型，则都能是正常读取数据的。