Debezium连接器过滤与Kafka处理器过滤方案选型咨询

方案一：Debezium连接器端Groovy过滤 #

通过配置Debezium的JSR223 Groovy过滤器，在数据进入Kafka前完成过滤：

"transforms.filter.language": "jsr223.groovy",
"transforms.filter.condition": "value.before.status != value.after.status && value.after.status == 'canceled'"

优缺点分析 #

• 优势：
  • 过滤逻辑前置，直接拦截不符合条件的记录，减少Kafka的存储、带宽消耗，降低下游处理器的负载
  • 纯配置实现，无需修改业务代码，调整过滤规则时只需更新连接器配置，无需重新部署业务服务
  • 业务处理器只需处理目标事件，代码更简洁，专注核心业务逻辑
• 劣势：
  • 不符合条件的记录会被直接丢弃，无法为其他需要全量变更数据的业务场景提供支持
  • Groovy表达式的调试、排查难度比Java代码高，缺乏成熟的IDE调试工具支持

方案二：Kafka处理器端过滤 #

将全量变更记录发送到Kafka，在业务处理器中通过代码完成过滤：

public void onMessage(EntityChangedMessage entityChangedMessage) {
  if(entityChangedMessage.getAfter().getStatus() == entityChangedMessage.getBefore().getStatus() ||
     entityChangedMessage.getAfter().getStatus() != "canceled") {
   return;
  }
  // 满足条件后的业务逻辑...
}

优缺点分析 #

• 优势：
  • 保留全量数据库变更记录，其他开发者可以基于这些数据扩展新业务（比如统计状态变更趋势、同步至其他系统等）
  • Java代码实现过滤逻辑，调试、维护更方便，有完整的代码追踪和调试能力
• 劣势：
  • 全量数据进入Kafka，会增加集群的存储和带宽成本，数据量较大时开销明显
  • 过滤逻辑混入业务方法，代码显得冗余，增加了处理器的额外处理负担

选型建议 #

• 如果没有全量变更数据的复用需求，或者短期内不会有相关业务扩展，优先选择方案一：既节省资源，又能让业务代码保持简洁，规则调整也更高效
• 如果存在其他业务方需要使用全量变更数据，或者需要更灵活的调试能力，选择方案二。若觉得代码杂乱，可以把过滤逻辑抽成独立的工具方法（比如boolean isStatusChangedToCanceled(EntityChangedMessage msg)），让业务方法更聚焦核心逻辑

内容的提问来源于stack exchange，提问作者Tom Carmi