[SPARK-53728][SDP] Print PipelineEvent Message with Error In Test

### What changes were proposed in this pull request?

Pulling some changes from #51644
For debugging purposes, print PipelineEvent with error to the console. Most tests expects pipeline to succeed, this makes it easier to see the error when it happens. Otherwise, developers have to manually add println to print the error message contained in the event, which is cumbersome.

### Why are the changes needed?

Easier to debug test failures, avoid the need to manually add println for test failures.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #52463 from JiaqiWang18/SPARK-53728-print-event-with-error-in-test.

Authored-by: Jacky Wang <[email protected]>
Signed-off-by: Sandy Ryza <[email protected]>

Author: jackywang-db

sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSender.scala

@@ -158,18 +158,6 @@ class PipelineEventSender(
   }
 
   private def constructProtoEvent(event: PipelineEvent): proto.PipelineEvent = {
-    val message = if (event.error.nonEmpty) {
-      // Returns the message associated with a Throwable and all its causes
-      def getExceptionMessages(throwable: Throwable): Seq[String] = {
-        throwable.getMessage +:
-          Option(throwable.getCause).map(getExceptionMessages).getOrElse(Nil)
-      }
-      val errorMessages = getExceptionMessages(event.error.get)
-      s"""${event.message}
-         |Error: ${errorMessages.mkString("\n")}""".stripMargin
-    } else {
-      event.message
-    }
     val protoEventBuilder = proto.PipelineEvent
       .newBuilder()
       .setTimestamp(
@@ -182,7 +170,7 @@ class PipelineEventSender(
           .setSeconds(event.timestamp.getTime / 1000)
           .setNanos(event.timestamp.getNanos)
           .build())
-      .setMessage(message)
+      .setMessage(event.messageWithError)
     protoEventBuilder.build()
   }
 }

sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/PipelineEvent.scala

@@ -39,7 +39,23 @@ case class PipelineEvent(
     message: String,
     details: EventDetails,
     error: Option[Throwable]
-)
+) {
+  /** Combines the message and error (if any) into a single string */
+  def messageWithError: String = {
+    if (error.nonEmpty) {
+      // Returns the message associated with a Throwable and all its causes
+      def getExceptionMessages(throwable: Throwable): Seq[String] = {
+        throwable.getMessage +:
+          Option(throwable.getCause).map(getExceptionMessages).getOrElse(Nil)
+      }
+      val errorMessages = getExceptionMessages(error.get)
+      s"""${message}
+         |Error: ${errorMessages.mkString("\n")}""".stripMargin
+    } else {
+      message
+    }
+  }
+}
 
 /**
  * Describes where the event originated from

sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/utils/ExecutionTest.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.pipelines.utils
 
+import org.scalatest.Assertions.fail
+
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.classic.SparkSession
 import org.apache.spark.sql.pipelines.common.{FlowStatus, RunState}
@@ -51,17 +53,34 @@ trait TestPipelineUpdateContextMixin {
    * @param fullRefreshTables Set of tables to be fully refreshed.
    * @param refreshTables  Set of tables to be refreshed.
    * @param resetCheckpointFlows Set of flows to be reset.
+   * @param failOnErrorEvent Whether to fail test when receiving event with error.
    */
   case class TestPipelineUpdateContext(
       spark: SparkSession,
       unresolvedGraph: DataflowGraph,
       fullRefreshTables: TableFilter = NoTables,
       refreshTables: TableFilter = AllTables,
-      resetCheckpointFlows: FlowFilter = AllFlows
+      resetCheckpointFlows: FlowFilter = AllFlows,
+      failOnErrorEvent: Boolean = false
   ) extends PipelineUpdateContext {
     val eventBuffer = new PipelineRunEventBuffer()
 
-    override val eventCallback: PipelineEvent => Unit = eventBuffer.addEvent
+    override val eventCallback: PipelineEvent => Unit = { event =>
+      eventBuffer.addEvent(event)
+      // For debugging purposes, print the event to the console.
+      // Most tests expects pipeline to succeed, this makes it easier to see
+      // the error when it happens.
+      if (event.error.nonEmpty) {
+        // scalastyle:off println
+        println("\n=== Received Pipeline Event with Error ===")
+        println(event.messageWithError)
+        println("=================================")
+        // scalastyle:on println
+        if (failOnErrorEvent) {
+          fail(s"Pipeline event with error received: ${event.messageWithError}")
+        }
+      }
+    }
 
     override def flowProgressEventLogger: FlowProgressEventLogger = {
       new FlowProgressEventLogger(eventCallback = eventCallback)