人为生成retract记录

[lec ssmi]

Hi:
    假设我现在将上游retract后的结果写入到kafka，然后下游程序消费kafka去做聚合操作。
    因为需要利用聚合算子能够自动处理retract的特性，所以需要将kafka的结果封装成带有不同header的row，即组装为INSERT
DELETE UPDATE类型的数据。
    有什么办法可以解决吗？
    如果将上下游程序合在一起 ，是没问题的，现在的难点就是拆分。
   谢谢。