spotify-scio - 是否可以使用 scio JobTest 控制输入处理时间？

Question

我们正在使用 com.spotify.scio.testing.JobTest 对我们的 scio 管道进行端到端测试。该管道包括一个对数据排序敏感的 DoFn，位于不常到达的配置数据流上。

我们将配置值的有序列表combinedSampleConfig作为输入传递给 JobTest Builder。当我们运行端到端测试时，有没有办法让 JobTest 保留这个 CustomIO 输入流的顺序？

我看到测试框架advanceProcessingTime在测试单个组件时可以很好地控制源到达时间（使用 ），但看不到如何使用 JobTest 将其应用于端到端测试。

    JobTest[MyApp.type]
      .args(commonArgs ++ Seq(
        "--numWorkers=1",
        "--maxNumWorkers=1",
      ): _*
      )
      .input(CustomIO[PubsubMessage](CONFIG_ID), combinedSampleConfig)
      .input(CustomIO[IndicatorEntry](INPUT_ID), sampleInput)
      .output(CustomIO[EnrichedIndicatorEntry](AGG_ID)) {
        _ should containInAnyOrder (expectedAggs)
      }
      .output(CustomIO[EnrichedIndicatorEntry](EVENT_ID)) {
        _ should containInAnyOrder (expectedEvents)
      }
      .run()

Answer 1

https://github.com/spotify/scio/pull/1905

这个 PR 最近被合并了，应该允许这样的用例。你能试一试吗？