Flink临时视图中被过滤记录的删除事件生成问题

问题判定 #

该现象不是预期行为，核心原因是SQL语法缺失和源表配置错误，以下是具体根因和修复方案：

根因分析 #

1. 聚合查询缺失GROUP BY子句
   你当前的查询使用了COUNT聚合函数，但未声明GROUP BY user_id，Flink会将所有结果聚合为全局单一行，不会按用户维度生成独立的changelog事件，自然不会在单个用户删除时输出对应DELETE行。
2. CDC源表未声明主键
   Debezium CDC源表依赖主键标识行的唯一性，你未在users表的DDL中定义主键，Flink无法正确关联更新前后的同一用户行，导致撤回事件无法正确下发。
3. 过滤条件位置不规范
   你将is_deleted的过滤条件放在JOIN之后，部分低版本Flink处理JOIN后过滤的changelog时可能出现异常，导致撤回事件被丢弃。
4. 你原有DDL中properties.bootstrap.servers = '<brokers>'行末尾遗漏逗号，会导致DDL解析报错。

修复方案 #

1 修正源表DDL，添加主键 #

CREATE TABLE users (
 user_id INT,
 is_deleted BOOLEAN,
 PRIMARY KEY (user_id) NOT ENFORCED -- 新增主键声明，NOT ENFORCED表示Flink不校验数据唯一性，仅用做changelog关联
) WITH (
 'connector' = 'kafka',
 'topic' = 'cdc.users',
 'properties.bootstrap.servers' = '<brokers>', -- 补全遗漏的逗号
 'properties.group.id' = '1',
 'format' = 'debezium-json',
 'scan.startup.mode' = 'earliest-offset'
)

如果purchases表也是CDC源，也建议添加对应主键声明。

2 修正SQL查询逻辑 #

补全GROUP BY子句，同时将过滤条件前置到users表的子查询中，语义更清晰，也能减少JOIN的数据量：

SELECT 
 u.user_id,
 COUNT(p.product_id) as purchases_count
FROM (SELECT * FROM users WHERE is_deleted = FALSE) u
LEFT JOIN purchases p ON u.user_id = p.user_id
GROUP BY u.user_id -- 补充分组逻辑，按用户维度聚合

3 （可选）调整ChangelogStream配置 #

如果修复上述配置后仍未输出DELETE事件，可以关闭changelog规范化，强制输出所有类型的changelog行：

import org.apache.flink.types.RowKind;
import org.apache.flink.table.api.ChangelogMode;

Table table = tableEnv.sqlQuery("<修正后的查询语句>");
DataStream<Row> ds = tableEnv.toChangelogStream(
    table,
    ChangelogMode.newBuilder()
        .addKind(RowKind.INSERT)
        .addKind(RowKind.DELETE)
        .addKind(RowKind.UPDATE_BEFORE)
        .addKind(RowKind.UPDATE_AFTER)
        .build()
);

完成上述修改后，当user2的is_deleted更新为true时，视图会正常输出对应user2的DELETE行，同步到Elasticsearch后会自动删除对应文档。

内容的提问来源于stack exchange，提问作者D P