Flink CDC读取kafka中的订单和订单明细数据，同步到sink表后，这一般是什么问题啊？

Flink CDC读取kafka中的订单和订单明细数据，同步到sink表后，sink表的数据数量一直跳来跳去，一会几百条一会变成几十条，这一般是什么问题啊？
感觉总是最后一次数据【新产生的一小部分流数据】，会把sink表中所有历史数据都清空掉用的mysql: CREATE TABLE sink_sale_order_detail_rt
(
order_item_id bigint,
order_id bigint,
delivery_order_id bigint,
product_id bigint,
order_create_time bigint,
item_price decimal(10, 2),
user_id bigint,
user_action_type int,
order_sell_way int,
store_id bigint,
sale_id bigint,
order_creation_type int,
product_offline_sign int,
gmv_date date,
gmv_channel_name string,
gmv_action_name string,
PRIMARY KEY (order_item_id,order_id) NOT ENFORCED
) WITH (
'connector.type' = 'jdbc',
'connector.url' = 'jdbc:mysql://10.254.33.177:3306/beta2_data_screen?useUnicode=true&characterEncoding=utf-8',
'connector.table' = 'sale_order_detail_rt',
'connector.driver' = 'com.mysql.jdbc.Driver',
'connector.username' = 'test',
'connector.password' = 'test123',
'connector.write.flush.interval' = '1s'
);