diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
index aeabd6a15881d..4ad12bf9fb0ac 100644
--- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
+++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala
@@ -241,6 +241,8 @@ private[hive] class SparkExecuteStatementOperation(
       dataTypes = result.queryExecution.analyzed.output.map(_.dataType).toArray
     } catch {
       case e: HiveSQLException =>
+        HiveThriftServer2.listener.onStatementError(
+          statementId, e.getMessage, SparkUtils.exceptionString(e))
         if (getStatus().getState() == OperationState.CANCELED) {
           return
         } else {
@@ -252,13 +254,35 @@ private[hive] class SparkExecuteStatementOperation(
       case e: Throwable =>
         val currentState = getStatus().getState()
         logError(s"Error executing query, currentState $currentState, ", e)
-        setState(OperationState.ERROR)
         HiveThriftServer2.listener.onStatementError(
           statementId, e.getMessage, SparkUtils.exceptionString(e))
+        if (statementId != null) {
+          sqlContext.sparkContext.cancelJobGroup(statementId)
+        }
+        getStatus().getState() match {
+          case OperationState.INITIALIZED =>
+            setState(OperationState.CLOSED)
+          case OperationState.CANCELED =>
+          case OperationState.FINISHED =>
+          case OperationState.CLOSED =>
+          case OperationState.ERROR =>
+          case _ =>
+            setState(OperationState.ERROR)
+        }
+
         throw new HiveSQLException(e.toString)
     }
-    setState(OperationState.FINISHED)
     HiveThriftServer2.listener.onStatementFinish(statementId)
+    getStatus().getState() match {
+      case OperationState.INITIALIZED =>
+        setState(OperationState.CLOSED)
+      case OperationState.CANCELED =>
+      case OperationState.FINISHED =>
+      case OperationState.CLOSED =>
+      case OperationState.ERROR =>
+      case _ =>
+        setState(OperationState.FINISHED)
+    }
   }
 
   override def cancel(): Unit = {