错误的flink-cdc语句sql

CREATE TABLE t_wx_source_1 (id String,name String,age String
) WITH ('connector' = 'oracle-cdc','hostname' = '192.168.1.135','port' = '1521','username' = 'flink','password' = 'XXXXX','database-name' = 'dbc','schema-name' = 'FLINK','table-name' = 't_wx_source_2'
);
CREATE TABLE t_wx_target (id String Not Null,name String,age String,PRIMARY KEY(id) NOT ENFORCED
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://192.168.1.91:3306/test','table-name' = 't_wx_target_1','username' = 'root','password' = 'XXXXX','driver' =  'com.mysql.cj.jdbc.Driver'
);
insert into t_wx_target select id,name,age from t_wx_source_1;

我们看一下oracle的数据库字段

再看一下错误sql里面的内容

flink报错内容

Column ‘id’ is NOT NULL, however, a null value is being written into it. You can set job configuration ‘table.exec.sink.not-null-enforcer’=‘DROP’ to suppress this exception and drop such records silently
大致意思就是不能插入为空的数值。
为什么会报这个错误，我们来看DML的执行语句：
insert into t_wx_target select id,name,age from t_wx_source_1;

每次数据都是null,然后我们的sink（t_wx_target ）表里面的字段id，是非空字段，所以就报错了

正确的sql

CREATE TABLE t_wx_source_1 (id String NOT NULL,name String,age String
) WITH ('connector' = 'oracle-cdc','hostname' = '192.168.1.135','port' = '1521','username' = 'flink','password' = 'XXXX','database-name' = 'dbc','schema-name' = 'FLINK','table-name' = 't_wx_source_2'
);
CREATE TABLE t_wx_target (id String,name String,age String,PRIMARY KEY(id) NOT ENFORCED
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://192.168.1.91:3306/test','table-name' = 't_wx_target_1','username' = 'root','password' = 'XXXXXX','driver' =  'com.mysql.cj.jdbc.Driver'
);
insert into t_wx_target select id,name,age from t_wx_source_1;

三级目录