使用Apache Beam的PAssert能否断言预期抛出的异常？

核心结论 #

PAssert本身仅支持校验PCollection的输出内容，没有直接提供预期管道执行阶段异常的内置能力，可以通过以下两种常用方案实现需求。

方案1：使用测试框架的异常断言（推荐适用于管道会直接终止的场景） #

管道执行时抛出的异常会被包装为PipelineExecutionException，直接在测试框架的异常断言中包裹testPipeline.run()调用即可实现校验，不需要修改原有业务逻辑。

import org.junit.jupiter.api.Assertions.assertThrows
import org.apache.beam.sdk.Pipeline.PipelineExecutionException

// 原有业务逻辑不变
val invalidParDo = ParDo.of(new SomeClass("This is an invalid parameter"))
inputPCollection.apply("", invalidParDo)

// 断言管道运行时会抛出预期异常
val executionEx = assertThrows(classOf[PipelineExecutionException], () => testPipeline.run())
// 可选：进一步校验根异常的类型和信息
assert(executionEx.getCause.isInstanceOf[YourExpectedExceptionType])
assert(executionEx.getCause.getMessage.contains("预期的错误提示"))

方案2：侧输出流捕获元素级异常（适用于异常不会终止管道的场景） #

如果异常是处理特定元素时抛出、不会终止整个管道运行，可以通过额外的DoFn封装原有逻辑，把捕获到的异常输出到侧输出流，再用PAssert校验侧输出流的内容。

import org.apache.beam.sdk.values.{OutputTag, TupleTagList}

// 定义侧输出标签，用于收集处理过程中的异常
val exceptionTag = OutputTag[Throwable]("par-do-exceptions")
val mainOutputTag = OutputTag[YourOutputType]("main-output")

// 封装原有业务逻辑，捕获处理异常
val exceptionCachingParDo = ParDo.of(new DoFn[YourInputType, YourOutputType] {
  @ProcessElement
  def process(c: ProcessContext): Unit = {
    try {
      // 调用原有SomeClass的处理逻辑
      val processor = new SomeClass("This is an invalid parameter")
      c.output(processor.process(c.element()))
    } catch {
      case e: Throwable => c.output(exceptionTag, e)
    }
  }
}).withOutputTags(mainOutputTag, TupleTagList.of(exceptionTag))

// 执行管道并获取侧输出的异常流
val outputTuple = inputPCollection.apply(exceptionCachingParDo)
val exceptionStream = outputTuple.get(exceptionTag)

// 用PAssert校验异常是否符合预期
PAssert.that(exceptionStream).satisfies(exceptions => {
  val ex = exceptions.iterator.next()
  assert(ex.isInstanceOf[YourExpectedExceptionType])
  assert(ex.getMessage.contains("预期的错误提示"))
  null
})

testPipeline.run()

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