Routine Load处理Flink CDC debezium格式的删除的情况

[liuxuzxx]

Flink CDC 3在做同步的时候，能够获取到数据库的:insert,update,delete操作。目前发现debezium对应的三种格式如下

1. insert格式

{
    "before": null,
    "after": {
        "user_id": 5859346,
        "enterprise_id": 192661,
        "timeinterval": 30,
         ......还有很多字段，省略
    },
    "op": "c",
    "source": {
        "db": null,
        "table": "kafka_gsms_user_info"
    }
}

2. update格式

{
    "before": {
        "user_id":5859346,
        "enterprise_id": 192661,
        "timeinterval": 90,
        ......省略
    },
    "after": {
        "user_id": 5859346,
        "enterprise_id": 192661,
        "timeinterval": 30,
         ......还有很多字段，省略
    },
    "op": "u",
    "source": {
        "db": null,
        "table": "kafka_gsms_user_info"
    }
}

3. delete格式

{
    "before": {
        "user_id":5859346,
        "enterprise_id": 192661,
        "timeinterval": 90,
        ......省略
    },
    "after": null,
    "op": "u",
    "source": {
        "db": null,
        "table": "kafka_gsms_user_info"
    }
}

对应给出的Routine Load为:

create routine load sedp_test.gsms_user_info_rl on gsms_user_info properties (
    "format" = "json",
    "json_root" = "$.after"
)
FROM KAFKA (
        "kafka_broker_list" = "172.16.1.118:9092,172.16.1.135:9092,172.16.1.149:9092", 
        "kafka_topic" = "kafka_gsms_user_info", 
         "kafka_offsets" = "266565,0,0", 
        "kafka_partitions" = "0,1,2", 
        "property.group.id" = "doris_gsms_user_info_cg"
    );

问题是:

这种Routine Load的写法，能够处理: insert和update操作，因为能够获取到$.after .但是无法处理delete的操作,因为delete的时候 $.after是null,会报错

但是insert/update/delete的CDC都会投递到同一个Topic，这种怎么处理？我翻了下Doris的官方文档，没有看到能够兼容处理insert/update和delete的事件

[liuxuzxx]

目前想到了一个方案：
就是首先使用Flink CDC 3执行Mysql的数据抽取到同一个Kafka‘的Topic中，然后，提供一个工具，消费这个Topic，根据source.db和source.table,根据映射关系，转发到另外的Topic中，在转发的过程中，丢弃掉:op:"d"的CDC Event