diff --git a/kyuubi-main/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala b/kyuubi-main/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
index 0f002d5d4c1..7292a1353ad 100644
--- a/kyuubi-main/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
+++ b/kyuubi-main/src/main/scala/org/apache/kyuubi/operation/ExecuteStatement.scala
@@ -79,7 +79,7 @@ class ExecuteStatement(
     } catch onError()
   }
 
-  private def waitStatementComplete(): Unit = {
+  private def waitStatementComplete(): Unit = try {
     setState(OperationState.RUNNING)
     var statusResp = client.GetOperationStatus(statusReq)
     var isComplete = false
@@ -119,7 +119,7 @@ class ExecuteStatement(
     }
     // see if anymore log could be fetched
     getQueryLog()
-  }
+  } catch onError()
 
   private def getQueryLog(): Unit = {
     getOperationLog.foreach { logger =>
diff --git a/kyuubi-main/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala b/kyuubi-main/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
index ecbc16710f4..75da6246f6b 100644
--- a/kyuubi-main/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
+++ b/kyuubi-main/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.operation
 
+import org.apache.hive.service.rpc.thrift.{TExecuteStatementReq, TGetOperationStatusReq, TOperationState, TStatusCode}
+
 import org.apache.kyuubi.WithKyuubiServer
 import org.apache.kyuubi.config.KyuubiConf
 
@@ -30,4 +32,19 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with BasicJDBCT
   override protected val conf: KyuubiConf = {
     KyuubiConf().set(KyuubiConf.ENGINE_SHARE_LEVEL, "connection")
   }
+
+  test("KYUUBI #647 - engine crash") {
+    withSessionHandle { (client, handle) =>
+      val executeStmtReq = new TExecuteStatementReq()
+      executeStmtReq.setStatement("select java_method('java.lang.System', 'exit', 1)")
+      executeStmtReq.setSessionHandle(handle)
+      executeStmtReq.setRunAsync(true)
+      val executeStmtResp = client.ExecuteStatement(executeStmtReq)
+
+      val getOpStatusReq = new TGetOperationStatusReq(executeStmtResp.getOperationHandle)
+      val getOpStatusResp = client.GetOperationStatus(getOpStatusReq)
+      assert(getOpStatusResp.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
+      assert(getOpStatusResp.getOperationState === TOperationState.ERROR_STATE)
+    }
+  }
 }