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

【问题标题】：Is it possible to control input processing time with scio JobTest?是否可以使用 scio JobTest 控制输入处理时间？
【发布时间】：2019-04-19 14:56:31
【问题描述】：

我们正在使用 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()

【问题讨论】：

目前没有办法，但通过允许 Beam TestStream（支持事件计时）作为输入可能是可行的。我提交了github.com/spotify/scio/issues/1891。

标签： spotify-scio

【解决方案1】：

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

此 PR 最近被合并，应该允许这样的用例。可以试试吗？

【讨论】：