[Backport-0.7]: Parse error message in client in EndStreamAction (#717)

Author: JialeTomTian

server/src/main/scala/io/delta/sharing/server/model.scala

@@ -132,8 +132,8 @@ case class RemoveFile(
 }
 
 case class EndStreamAction(
-    refreshToken: String
-) extends Action {
+    refreshToken: String,
+    errorMessage: String = null) extends Action {
   override def wrap: SingleAction = SingleAction(endStreamAction = this)
 }
 

spark/src/main/scala/io/delta/sharing/spark/DeltaSharingClient.scala

@@ -729,6 +729,10 @@ private[spark] object DeltaSharingRestClient extends Logging {
         logInfo(
           s"Successfully verified endStreamAction in the response" + queryIdForLogging
         )
+        if(lastEndStreamAction.errorMessage != null) {
+          throw new DeltaSharingServerException("Request failed during streaming response " +
+            s"with error message ${lastEndStreamAction.errorMessage}")
+        }
       case Some(false) =>
         logWarning(s"Client sets ${DELTA_SHARING_INCLUDE_END_STREAM_ACTION}=true in the " +
           s"header, but the server responded with the header set to false(" +

spark/src/main/scala/io/delta/sharing/spark/DeltaSharingErrors.scala

@@ -20,6 +20,8 @@ import org.apache.spark.sql.types.StructType
 
 class MissingEndStreamActionException(message: String) extends IllegalStateException(message)
 
+class DeltaSharingServerException(message: String) extends RuntimeException(message)
+
 
 object DeltaSharingErrors {
   def nonExistentDeltaSharingTable(tableId: String): Throwable = {

spark/src/main/scala/io/delta/sharing/spark/model.scala

@@ -115,7 +115,8 @@ private[sharing] case class Protocol(minReaderVersion: Int) extends Action {
 }
 
 private[sharing] case class EndStreamAction(
-    refreshToken: String)
+    refreshToken: String,
+    errorMessage: String = null)
   extends Action {
   override def wrap: SingleAction = SingleAction(endStreamAction = this)
 }

spark/src/test/scala/io/delta/sharing/spark/DeltaSharingRestClientSuite.scala

@@ -886,6 +886,10 @@ class DeltaSharingRestClientSuite extends DeltaSharingIntegrationTest {
   val fakeEndStreamActionStr = JsonUtils.toJson(EndStreamAction(
     refreshToken = "random-refresh"
   ).wrap)
+  val fakeEndStreamActionErrorStr = JsonUtils.toJson(EndStreamAction(
+    refreshToken = null,
+    errorMessage = "BAD REQUEST: Error Occurred During Streaming"
+  ).wrap)
 
   test("checkEndStreamAction succeeded") {
     // DELTA_SHARING_INCLUDE_END_STREAM_ACTION=true
@@ -973,4 +977,36 @@ class DeltaSharingRestClientSuite extends DeltaSharingIntegrationTest {
     }
     checkErrorMessage(e, s"and 0 lines, and last line as [Empty_Seq_in_checkEndStreamAction].")
   }
+
+  test("checkEndStreamAction with error message throws streaming error") {
+    def checkErrorMessage(
+      e: DeltaSharingServerException,
+      additionalErrorMsg: String): Unit = {
+      val commonErrorMsg = "Request failed during streaming response with error message"
+      assert(e.getMessage.contains(commonErrorMsg))
+      assert(e.getMessage.contains(additionalErrorMsg))
+    }
+
+    // checkEndStreamAction throws error if the last line is EndStreamAction with error message
+    var e = intercept[DeltaSharingServerException] {
+      checkEndStreamAction(
+        Some(s"$DELTA_SHARING_INCLUDE_END_STREAM_ACTION=true"),
+        Map(DELTA_SHARING_INCLUDE_END_STREAM_ACTION -> "true"),
+        Seq(fakeAddFileStr, fakeEndStreamActionErrorStr),
+        "random-query-id"
+      )
+    }
+    checkErrorMessage(e, "BAD REQUEST: Error Occurred During Streaming")
+
+    // checkEndStreamAction throws error if the only line is EndStreamAction with error message
+    e = intercept[DeltaSharingServerException] {
+      checkEndStreamAction(
+        Some(s"$DELTA_SHARING_INCLUDE_END_STREAM_ACTION=true"),
+        Map(DELTA_SHARING_INCLUDE_END_STREAM_ACTION -> "true"),
+        Seq(fakeEndStreamActionErrorStr),
+        "random-query-id"
+      )
+    }
+    checkErrorMessage(e, "BAD REQUEST: Error Occurred During Streaming")
+  }
 }