错误的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;

三级目录