我正在开发一个自定义streamingquerylistener，并希望在测试中触发它的onqueryterminated方法。
这就是我尝试实现的：

import org.apache.spark.sql.{ SQLContext, SparkSession }
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.functions.{ col, to_date }
import org.apache.spark.sql.streaming.StreamingQueryListener
import org.scalatest.flatspec.AnyFlatSpec

class MyListener extends StreamingQueryListener {
  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit       = {}
  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit     = {}
  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = println(event.exception)
}

class ListenerSpec extends AnyFlatSpec {

  it should "trigger onQueryTerminated" in {
    val spark = SparkSession.builder().master("local[*]").getOrCreate()
    spark.streams.addListener(new MyListener())
    implicit val sqlContext: SQLContext = spark.sqlContext

    import spark.implicits._

    val stream = MemoryStream[Int]
    stream.addData(Seq(1, 3, 4))

    val query = stream
      .toDF()
      .withColumn("columnDoesntExist", to_date(col("names")))
      .writeStream
      .format("console")
      .start()

    query.awaitTermination()
  }
}

但是，这不起作用，因为它会引发analysisexception，但onqueryterminated方法不会由流查询的终止触发。
在什么情况下该方法被触发，而event.exception是一些（异常）？

更新

以下代码成功触发 onQueryTerminated :

val exceptionUdf = udf(() => throw new Exception())

val query = stream
      .toDF()
      .withColumn("exception", exceptionUdf())
      .writeStream
      .format("console")
      .start()

请参阅公认的答案来解释原因。