ProcessAllWindowFunction 的 Scala 单元测试

Question

阅读官方flink测试文档后（https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/testing.html） 我能够使用测试工具为 ProcessFunction 开发测试，如下所示：

pendingPartitionBuilder = new PendingPartitionBuilder(":::some_name", "")

testHarness =
  new OneInputStreamOperatorTestHarness[StaticAdequacyTilePublishedData, PendingPartition](
    new ProcessOperator[StaticAdequacyTilePublishedData, PendingPartition](pendingPartitionBuilder)
  )

testHarness.open()

现在，我正在尝试对 ProcessAllWindowFunction 执行相同的操作，如下所示：

class MapVersionValidationDistributor(batchSize: Int) extends
  ProcessAllWindowFunction[MapVersionValidation, Seq[StaticAdequacyTilePublishedData],TimeWindow] {

lazy val state: ValueState[Long] = getRuntimeContext .getState(new ValueStateDescriptor[Long]("latestMapVersion", classOf[Long]))
(...)

首先我意识到我不能对 ProcessAllWindowFunction 使用 TestHarness，因为它没有 processElement 方法。在这种情况下，我应该遵循什么单元测试策略？

编辑：目前我的测试代码如下所示：

val collector = mock[Collector[Seq[StaticAdequacyTilePublishedData]]]
    val mvv = new MapVersionValidationDistributor(1)
    val input3 = Iterable(new MapVersionValidation("123",Seq(TileValidation(1,true,Seq(1,3,4)))))

    val ctx = mock[mvv.Context]
    val streamContext =  mock[RuntimeContext]
    mvv.setRuntimeContext(streamContext)

    mvv.open(mock[Configuration])
    mvv.process(ctx,input3,collector)

我收到了这个错误：

 Unexpected call: <mock-3> RuntimeContext.getState[T](ValueStateDescriptor{name=latestMapVersion, defaultValue=null, serializer=null}) Expected: inAnyOrder { }

Answer 1

您真的不需要测试工具来对ProcessAllWindowFunction 的process 方法进行单元测试。 process 函数接受 3 个参数：Context、Iterable[IN]、Collector[OUT]。您可以使用一些库，具体取决于用于模拟 Context 的语言。您还可以根据您在此处的偏好轻松实现或模拟Collector。而 Iterable[IN] 只是一个 Iterable 包含您的窗口的元素，它会在窗口被触发后传递给函数。