[CARMEL-6006] Log stack trace to track the source of query cancelation (#962)

Author: xingchaozh

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkDownloadDataOperation.scala

@@ -577,7 +577,9 @@ private[hive] class SparkDownloadDataOperation(
 
   override def cancel(): Unit = {
     if (statementId != null) {
-      HiveThriftServer2.eventManager.onStatementCanceled(statementId)
+      HiveThriftServer2.eventManager.onStatementCanceled(statementId,
+        s"Canceling operation, stack trace:\n"
+          + Thread.currentThread().getStackTrace.mkString("\n"))
     }
     HiveThriftServer2.eventManager.onQueryExist(
       statementId,

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala

@@ -588,7 +588,9 @@ private[hive] class SparkExecuteStatementOperation(
         // 'InterruptedException', which could be misleading, the query final state
         // should be canceled instead of failed. Hence we need guarantee that the
         // cancellation event is posted before the query execution is actually canceled.
-        HiveThriftServer2.eventManager.onStatementCanceled(statementId)
+        HiveThriftServer2.eventManager.onStatementCanceled(statementId,
+          s"Canceling operation, stack trace:\n"
+            + Thread.currentThread().getStackTrace.mkString("\n"))
         cleanup()
       }
     }

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2EventManager.scala

@@ -61,8 +61,10 @@ private[thriftserver] class HiveThriftServer2EventManager(sc: SparkContext) {
     postLiveListenerBus(SparkListenerThriftServerOperationParsed(id, executionPlan))
   }
 
-  def onStatementCanceled(id: String): Unit = {
-    postLiveListenerBus(SparkListenerThriftServerOperationCanceled(id, System.currentTimeMillis()))
+  def onStatementCanceled(id: String, reason: String = ""): Unit = {
+    postLiveListenerBus(SparkListenerThriftServerOperationCanceled(id,
+      reason,
+      System.currentTimeMillis()))
   }
 
   def onStatementError(id: String, errorMsg: String, errorTrace: String): Unit = {
@@ -115,7 +117,7 @@ private[thriftserver] case class SparkListenerThriftServerOperationParsed(
     executionPlan: String) extends SparkListenerEvent
 
 private[thriftserver] case class SparkListenerThriftServerOperationCanceled(
-    id: String, finishTime: Long) extends SparkListenerEvent
+    id: String, reason: String, finishTime: Long) extends SparkListenerEvent
 
 private[thriftserver] case class SparkListenerThriftServerOperationError(
     id: String,

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala

@@ -278,6 +278,7 @@ private[thriftserver] class HiveThriftServer2Listener(
   private def onOperationCanceled(e: SparkListenerThriftServerOperationCanceled): Unit = {
     executionList.get(e.id).finishTimestamp = e.finishTime
     executionList.get(e.id).state = ExecutionState.CANCELED
+    executionList.get(e.id).detail = e.reason
     updateLiveStore(executionList.get(e.id))
   }