[KYUUBI #1946] Close the kyuubi connection on launch engine operation failure

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->
For kyuubi connection, if the launch engine operation fails, we need close the kyuubi connection.

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1946 from turboFei/close_conn_on_engine_failure.

Closes #1946

3a4a465 [Fei Wang] remove final:
8111726 [Fei Wang] add ut
74d3b28 [Fei Wang] address commments
0c41ac5 [Fei Wang] refactor
8dcc7ec [Fei Wang] Close the kyuubi connection on launch engine operation failure

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: turboFei

kyuubi-hive-beeline/src/main/java/org/apache/hive/beeline/KyuubiDatabaseConnection.java

@@ -140,6 +140,7 @@ public Connection getConnectionFromDefaultDriver(String url, Properties properti
         new Thread(beeLine.commands.createLogRunnable(kyuubiConnection, eventNotifier));
     logThread.setDaemon(true);
     logThread.start();
+    kyuubiConnection.setEngineLogThread(logThread);
 
     kyuubiConnection.waitLaunchEngineToComplete();
     logThread.interrupt();

kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiConnection.java

@@ -78,6 +78,7 @@
 public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
   public static final Logger LOG = LoggerFactory.getLogger(KyuubiConnection.class.getName());
   public static final String BEELINE_MODE_PROPERTY = "BEELINE_MODE";
+  public static int DEFAULT_ENGINE_LOG_THREAD_TIMEOUT = 10 * 1000;
 
   private String jdbcUriString;
   private String host;
@@ -100,6 +101,7 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
   private boolean initFileCompleted = false;
 
   private TOperationHandle launchEngineOpHandle = null;
+  private Thread engineLogThread;
   private boolean engineLogInflight = true;
   private volatile boolean launchEngineOpCompleted = false;
 
@@ -263,7 +265,7 @@ public List<String> getExecLog() throws SQLException, ClosedOrCancelledException
   private void showLaunchEngineLog() {
     if (launchEngineOpHandle != null) {
       LOG.info("Starting to get launch engine log.");
-      Thread logThread =
+      engineLogThread =
           new Thread("engine-launch-log") {
 
             @Override
@@ -282,10 +284,14 @@ public void run() {
               LOG.info("Finished to get launch engine log.");
             }
           };
-      logThread.start();
+      engineLogThread.start();
     }
   }
 
+  public void setEngineLogThread(Thread logThread) {
+    this.engineLogThread = logThread;
+  }
+
   public void executeInitSql() throws SQLException {
     if (initFileCompleted) return;
     if (initFile != null) {
@@ -937,6 +943,18 @@ public void close() throws SQLException {
     }
   }
 
+  private void closeOnLaunchEngineFailure() throws SQLException {
+    if (engineLogThread != null && engineLogThread.isAlive()) {
+      engineLogThread.interrupt();
+      try {
+        engineLogThread.join(DEFAULT_ENGINE_LOG_THREAD_TIMEOUT);
+      } catch (Exception e) {
+      }
+    }
+    engineLogThread = null;
+    close();
+  }
+
   /*
    * (non-Javadoc)
    *
@@ -1669,12 +1687,14 @@ public void waitLaunchEngineToComplete() throws SQLException {
               break;
           }
         }
-      } catch (SQLException e) {
-        engineLogInflight = false;
-        throw e;
       } catch (Exception e) {
         engineLogInflight = false;
-        throw new SQLException(e.toString(), "08S01", e);
+        closeOnLaunchEngineFailure();
+        if (e instanceof SQLException) {
+          throw e;
+        } else {
+          throw new SQLException(e.getMessage(), "08S01", e);
+        }
       }
     }
   }

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala

@@ -30,6 +30,7 @@ import org.apache.kyuubi.{Utils, WithKyuubiServer}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.SESSION_CONF_ADVISOR
 import org.apache.kyuubi.jdbc.KyuubiHiveDriver
+import org.apache.kyuubi.jdbc.hive.KyuubiConnection
 import org.apache.kyuubi.plugin.SessionConfAdvisor
 
 /**
@@ -203,6 +204,18 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
       }
     }
   }
+
+  test("close kyuubi connection on launch engine operation failure") {
+    withSessionConf(Map.empty)(Map.empty)(Map(
+      KyuubiConf.SESSION_ENGINE_LAUNCH_ASYNC.key -> "true",
+      "spark.master" -> "invalid")) {
+      val prop = new Properties()
+      prop.setProperty(KyuubiConnection.BEELINE_MODE_PROPERTY, "true")
+      val kyuubiConnection = new KyuubiConnection(jdbcUrlWithConf, prop)
+      intercept[SQLException](kyuubiConnection.waitLaunchEngineToComplete())
+      assert(kyuubiConnection.isClosed)
+    }
+  }
 }
 
 class TestSessionConfAdvisor extends SessionConfAdvisor {