Scio JobTest端到端测试：如何控制处理时间并保留输入流顺序？

1. 保留CustomIO输入流的顺序 #

默认情况下，直接用.input(CustomIO[...], combinedSampleConfig)传入有序列表时，JobTest不一定能严格保证输入顺序——底层测试源可能会把数据当作无界集合处理。要确保顺序被严格保留，你需要改用**TestStream**来模拟CustomIO输入，它能精确控制元素的发射顺序。

修改后的测试代码示例：

import com.spotify.scio.testing.TestStream

// 构建严格遵循combinedSampleConfig顺序的TestStream
val configTestStream = TestStream(PubsubMessage)
  .addElements(combinedSampleConfig: _*)
  .advanceWatermarkToInfinity()

JobTest[MyApp.type]
  .args(commonArgs ++ Seq("--numWorkers=1", "--maxNumWorkers=1"): _*)
  .input(CustomIO[PubsubMessage](CONFIG_ID), configTestStream) // 用TestStream替代直接传列表
  .input(CustomIO[IndicatorEntry](INPUT_ID), sampleInput)
  .output(CustomIO[EnrichedIndicatorEntry](AGG_ID)) { _ should containInAnyOrder (expectedAggs) }
  .output(CustomIO[EnrichedIndicatorEntry](EVENT_ID)) { _ should containInAnyOrder (expectedEvents) }
  .run()

TestStream.addElements会完全按照你传入的元素顺序发射数据，再配合你已经设置的单worker参数，就能确保对顺序敏感的DoFn接收到有序的配置流。

2. 在JobTest中控制处理时间（含advanceProcessingTime的应用） #

advanceProcessingTime确实可以在JobTest中使用，但需要结合TestStream来实现——它是Scio测试框架中模拟时间推进的核心组件，能完美适配你“低频配置流”的场景。

场景1：模拟低频到达的配置流 #

如果你想模拟配置元素间隔一段时间到达，可以在TestStream中插入时间推进操作：

import java.time.Duration

val configTestStream = TestStream(PubsubMessage)
  // 发射第一个配置元素
  .addElements(combinedSampleConfig.head)
  // 模拟1小时后发射第二个元素
  .advanceProcessingTime(Duration.ofHours(1))
  .addElements(combinedSampleConfig(1))
  // 再模拟2小时后发射剩余所有配置元素
  .advanceProcessingTime(Duration.ofHours(2))
  .addElements(combinedSampleConfig.drop(2): _*)
  .advanceWatermarkToInfinity()

这样就能精确还原配置流的低频到达节奏，让测试更贴近真实场景。

场景2：全局推进处理时间触发窗口 #

如果你的管道包含窗口逻辑，需要推进处理时间来触发窗口计算，同样可以通过TestStream的advanceProcessingTime和advanceWatermark组合实现。比如在所有输入元素添加完成后，推进时间触发窗口关闭：

val inputTestStream = TestStream(IndicatorEntry)
  .addElements(sampleInput: _*)
  .advanceProcessingTime(Duration.ofMinutes(30)) // 推进30分钟处理时间
  .advanceWatermarkToInfinity()

额外注意事项 #

• 若你的输出结果也依赖顺序，要把断言从containInAnyOrder改成containTheSameElementsInOrder，确保输出顺序符合预期。
• 自定义IO需兼容TestStream：Scio内置IO（如PubsubIO）都支持TestStream，若你是自定义的CustomIO，要确保它在测试模式下能正确处理TestStream输入。

内容的提问来源于stack exchange，提问作者Ralph Gonzalez