读取MongoDB变更流的Splittable DoFn应选用何种合适的Restriction？

Restriction类型选择 #

你当前选用OffsetRange作为restriction类型完全适配场景，不需要额外替换。MongoDB集群时间戳BsonTimestamp的getValue()方法本身返回可比较的Long型值，刚好匹配OffsetRange的数值范围定义，对于仅携带时间戳、无精确偏移量的无限流场景来说，这个选择已经是最优解，自定义restriction类型不会带来额外收益。

CPU高与拆分优化方案 #

你遇到的CPU占用过高、迭代器频繁创建的问题，核心有两个原因：

1. GrowableOffsetRangeTracker默认允许Runner随时拆分restriction，尤其是DirectRunner为了验证拆分逻辑，会非常激进地触发拆分操作，导致tryClaim经常返回false，每次失败都会关闭当前迭代器，下次process调用又要重新创建Mongo游标拉取数据。
2. 正常读完当前可用的变更事件后，你返回的ProcessContinuation.resume()未设置延迟，会立刻触发下一次process调用，相当于空转循环反复创建游标。

可按以下方案优化：

1. 禁用restriction拆分：实现@SplitRestriction方法，返回空列表即可告知Beam当前restriction不可拆分：

@SplitRestriction
public void splitRestriction(@Element StreamConfig element, @Restriction OffsetRange restriction, OutputReceiver<OffsetRange> receiver) {
    // 无输出即代表没有可拆分的子restriction，禁用拆分逻辑
}

禁用拆分后tryClaim永远会返回true，不会再出现因拆分失败导致迭代器提前关闭的问题。
2. 给正常续跑添加延迟：读完当前所有可用变更事件后，不要立刻重试，增加几百毫秒到1秒的延迟，避免空转消耗CPU：

return ProcessContinuation.resume().withResumeDelay(Duration.millis(500L));

3. 可选优化：通过@State存储Mongo变更流的resume token，下次创建游标时直接用resume token续传，不需要每次都从restriction的起始时间戳重新拉取数据，避免重复处理已消费事件，同时降低MongoDB的访问压力。

额外说明：DirectRunner本身是为功能测试设计的，生产环境使用正式Runner（如Spark Runner、Flink Runner、Dataflow Runner）时，拆分策略本身会更温和，不会像测试环境这样激进触发拆分。

内容的提问来源于stack exchange，提问作者Patryk Koryzna