如何为MongoDB Sink Connector动态选择WriteStrategy？

针对你需要在单主题、按业务键分区保序的前提下，动态执行插入/更新/删除操作的需求，以下是三种落地方案，按推荐优先级排序：

1. 基于消息头/元数据动态切换内置策略（最推荐，无需自定义开发） #

MongoDB Sink Connector 1.10+版本支持通过EL表达式引用消息头/字段来动态指定write.strategy，完美匹配你的需求，且能保证分区内的顺序一致性。

配置步骤： #

• 第一步：标记操作类型：在发送消息时，通过Kafka消息头（比如x-operation）或消息体字段（比如operation）标识当前操作类型：
  • 插入：头值设为com.mongodb.kafka.connect.sink.writestrategy.InsertOneDefaultStrategy
  • 更新：头值设为com.mongodb.kafka.connect.sink.writestrategy.UpdateOneBusinessKeyTimestampStrategy
  • 删除：头值设为com.mongodb.kafka.connect.sink.writestrategy.DeleteOneBusinessKeyStrategy
• 第二步：配置Connector：在Connector的配置文件中，将write.strategy设为动态读取消息头的表达式：

  name=MongoDBSinkConnector
  connector.class=com.mongodb.kafka.connect.MongoSinkConnector
  topics=your-business-topic
  connection.uri=mongodb+srv://<user>:<pass>@cluster0.mongodb.net/
  database=your-db
  collection=your-collection
  # 配置业务键，确保同一业务键的消息在同一分区
  business.key.fields=your-business-key-field
  # 动态读取消息头指定写入策略
  write.strategy=${kafka.headers['x-operation']}
  # 其他必要配置...
  

核心优势： #

• 完全复用官方内置策略，无需开发
• 同一业务键的消息在同一分区，Connector按顺序处理，天然保证操作顺序一致性
• 逻辑清晰，后期维护成本低

2. 软删除方案（快速落地，适合对开发成本敏感的场景） #

如果动态策略配置存在环境限制（比如Connector版本过低），软删除是快速上线的折中方案：

实现逻辑： #

• 统一使用UpdateOneBusinessKeyTimestampStrategy作为写入策略
• 在文档中新增is_deleted（布尔型，默认false）和deleted_at（时间戳型）字段
• 插入/更新操作正常发送消息，删除操作则发送一条更新消息，将is_deleted设为true并填充deleted_at
• 业务查询时统一过滤is_deleted: false的文档

优缺点： #

• ✅ 优点：无需修改Connector配置，仅调整消息结构即可落地，开发成本极低
• ❌ 缺点：数据存在冗余，需定期清理软删除的历史数据；业务层需配合过滤逻辑

3. 自定义WriteStrategy（长期灵活方案，适合复杂场景） #

如果以上方案都无法满足你的特殊需求，再考虑自定义写入策略：

开发要点： #

• 实现官方的com.mongodb.kafka.connect.sink.writestrategy.WriteStrategy接口，在process方法中根据消息的操作标识（头或字段），调用对应的内置策略逻辑：

  public class DynamicOperationStrategy implements WriteStrategy {
      private InsertOneDefaultStrategy insertStrategy;
      private UpdateOneBusinessKeyTimestampStrategy updateStrategy;
      private DeleteOneBusinessKeyStrategy deleteStrategy;
  
      @Override
      public void init(SinkConnectorConfig config, MongoClient mongoClient) {
          // 初始化内置策略
          insertStrategy = new InsertOneDefaultStrategy();
          insertStrategy.init(config, mongoClient);
          updateStrategy = new UpdateOneBusinessKeyTimestampStrategy();
          updateStrategy.init(config, mongoClient);
          deleteStrategy = new DeleteOneBusinessKeyStrategy();
          deleteStrategy.init(config, mongoClient);
      }
  
      @Override
      public void process(SinkRecord record, MongoCollection<Document> collection) {
          // 从消息头或字段获取操作类型
          String operation = new String(record.headers().lastHeader("x-operation").value());
          switch(operation) {
              case "insert":
                  insertStrategy.process(record, collection);
                  break;
              case "update":
                  updateStrategy.process(record, collection);
                  break;
              case "delete":
                  deleteStrategy.process(record, collection);
                  break;
              default:
                  throw new IllegalArgumentException("Unsupported operation: " + operation);
          }
      }
  }
  

• 将自定义类打包成JAR，放到Kafka Connect的plugin.path目录下
• 在Connector配置中指定自定义策略：

  write.strategy=com.yourcompany.connect.DynamicOperationStrategy
  

优缺点： #

• ✅ 优点：完全自定义逻辑，适配各种复杂业务场景
• ❌ 缺点：需要开发维护成本，Connector版本升级时需注意兼容性

关键注意事项 #

• 分区顺序保障：确保主题的分区策略是按业务键哈希，且business.key.fields配置正确，保证同一业务键的消息落入同一个主题分区，Kafka Connect会按顺序处理同一分区内的消息，不会出现乱序。
• 幂等性处理：插入/更新操作建议开启upsert选项（通过update.one.upsert=true配置），避免重复消息导致的数据异常；删除操作可添加幂等校验，防止重复删除。

内容的提问来源于stack exchange，提问作者Nilay Sundarkar