[KYUUBI #3152] Introduce JDBC parameters to control connection timeout

### _Why are the changes needed?_

There are two concepts `connect timeout` and `so timeout` in the socket, currently, the Kyuubi Hive JDBC driver uses `DriverManager#loginTimeout` as both of them, it does not make sense, and will cause unexpected "read timeout" exceptions if `DriverManager#loginTimeout` was set a small value, e.g. Spring Boot set it to 30s in default.

This PR proposes to introduce two dedicated JDBC parameters to control them.

See also apache/hive#3379

### _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

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

Closes #3152 from pan3793/timeout.

Closes #3152

5dcae66 [Cheng Pan] nit
816a851 [Cheng Pan] nit
3b5e6ce [Cheng Pan] fix npe
4cb73df [Cheng Pan] Introduce JDBC parameters to control connection timeout

Authored-by: Cheng Pan <chengpan@apache.org>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: pan3793

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

@@ -37,7 +37,6 @@
 import java.sql.*;
 import java.util.*;
 import java.util.Map.Entry;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
@@ -51,6 +50,7 @@
 import org.apache.http.NoHttpResponseException;
 import org.apache.http.client.CookieStore;
 import org.apache.http.client.ServiceUnavailableRetryStrategy;
+import org.apache.http.client.config.RequestConfig;
 import org.apache.http.config.Registry;
 import org.apache.http.config.RegistryBuilder;
 import org.apache.http.conn.socket.ConnectionSocketFactory;
@@ -95,7 +95,8 @@ public class KyuubiConnection implements SQLConnection, KyuubiLoggable {
   private SQLWarning warningChain = null;
   private TSessionHandle sessHandle = null;
   private final List<TProtocolVersion> supportedProtocols = new LinkedList<>();
-  private int loginTimeout = 0;
+  private int connectTimeout = 0;
+  private int socketTimeout = 0;
   private TProtocolVersion protocol;
   private int fetchSize = KyuubiStatement.DEFAULT_FETCH_SIZE;
   private String initFile = null;
@@ -123,13 +124,12 @@ public static List<JdbcConnectionParams> getAllUrls(String zookeeperBasedHS2Url)
   }
 
   public KyuubiConnection(String uri, Properties info) throws SQLException {
-    setupLoginTimeout();
+    isBeeLineMode = Boolean.parseBoolean(info.getProperty(BEELINE_MODE_PROPERTY));
     try {
       connParams = Utils.parseURL(uri, info);
     } catch (ZooKeeperHiveClientException e) {
       throw new KyuubiSQLException(e);
     }
-    isBeeLineMode = Boolean.parseBoolean(info.getProperty(BEELINE_MODE_PROPERTY));
     jdbcUriString = connParams.getJdbcUriString();
     // JDBC URL: jdbc:hive2://<host>:<port>/dbName;sess_var_list?hive_conf_list#hive_var_list
     // each list: <key1>=<val1>;<key2>=<val2> and so on
@@ -140,6 +140,8 @@ public KyuubiConnection(String uri, Properties info) throws SQLException {
     port = connParams.getPort();
     sessConfMap = connParams.getSessionVars();
 
+    setupTimeout();
+
     if (sessConfMap.containsKey(FETCH_SIZE)) {
       fetchSize = Integer.parseInt(sessConfMap.get(FETCH_SIZE));
     }
@@ -153,17 +155,7 @@ public KyuubiConnection(String uri, Properties info) throws SQLException {
     }
 
     // add supported protocols
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V1);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V2);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V3);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V4);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V5);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V6);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V7);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V8);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V9);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V10);
-    supportedProtocols.add(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V11);
+    Collections.addAll(supportedProtocols, TProtocolVersion.values());
 
     int maxRetries = 1;
     try {
@@ -471,6 +463,15 @@ private CloseableHttpClient getHttpClient(Boolean useSsl) throws SQLException {
               customCookies);
     }
     HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
+
+    // Set timeout
+    RequestConfig config =
+        RequestConfig.custom()
+            .setConnectTimeout(connectTimeout)
+            .setSocketTimeout(socketTimeout)
+            .build();
+    httpClientBuilder.setDefaultRequestConfig(config);
+
     // Configure http client for cookie based authentication
     if (isCookieEnabled) {
       // Create a http client with a retry mechanism when the server returns a status code of 401.
@@ -576,15 +577,15 @@ private TTransport createUnderlyingTransport() throws TTransportException {
       String sslTrustStorePassword = sessConfMap.get(SSL_TRUST_STORE_PASSWORD);
 
       if (sslTrustStore == null || sslTrustStore.isEmpty()) {
-        transport = ThriftUtils.getSSLSocket(host, port, loginTimeout);
+        transport = ThriftUtils.getSSLSocket(host, port, connectTimeout, socketTimeout);
       } else {
         transport =
             ThriftUtils.getSSLSocket(
-                host, port, loginTimeout, sslTrustStore, sslTrustStorePassword);
+                host, port, connectTimeout, socketTimeout, sslTrustStore, sslTrustStorePassword);
       }
     } else {
       // get non-SSL socket transport
-      transport = ThriftUtils.getSocketTransport(host, port, loginTimeout);
+      transport = ThriftUtils.getSocketTransport(host, port, connectTimeout, socketTimeout);
     }
     return transport;
   }
@@ -856,13 +857,24 @@ private String getSessionValue(String varName, String varDefault) {
     return varValue;
   }
 
-  // copy loginTimeout from driver manager. Thrift timeout needs to be in millis
-  private void setupLoginTimeout() {
-    long timeOut = TimeUnit.SECONDS.toMillis(DriverManager.getLoginTimeout());
-    if (timeOut > Integer.MAX_VALUE) {
-      loginTimeout = Integer.MAX_VALUE;
-    } else {
-      loginTimeout = (int) timeOut;
+  private void setupTimeout() {
+    if (sessConfMap.containsKey(CONNECT_TIMEOUT)) {
+      String connectTimeoutStr = sessConfMap.get(CONNECT_TIMEOUT);
+      try {
+        long connectTimeoutMs = Long.parseLong(connectTimeoutStr);
+        connectTimeout = (int) Math.max(0, Math.min(connectTimeoutMs, Integer.MAX_VALUE));
+      } catch (NumberFormatException e) {
+        LOG.info("Failed to parse connectTimeout of value " + connectTimeoutStr);
+      }
+    }
+    if (sessConfMap.containsKey(SOCKET_TIMEOUT)) {
+      String socketTimeoutStr = sessConfMap.get(SOCKET_TIMEOUT);
+      try {
+        long socketTimeoutMs = Long.parseLong(socketTimeoutStr);
+        socketTimeout = (int) Math.max(0, Math.min(socketTimeoutMs, Integer.MAX_VALUE));
+      } catch (NumberFormatException e) {
+        LOG.info("Failed to parse socketTimeout of value " + socketTimeoutStr);
+      }
     }
   }
 

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

@@ -126,6 +126,9 @@ public static class JdbcConnectionParams {
     // Cookie prefix
     static final String HTTP_COOKIE_PREFIX = "http.cookie.";
 
+    static final String CONNECT_TIMEOUT = "connectTimeout";
+    static final String SOCKET_TIMEOUT = "socketTimeout";
+
     // We support ways to specify application name modeled after some existing DBs, since
     // there's no standard approach.
     // MSSQL: applicationName

kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/auth/ThriftUtils.java

@@ -29,27 +29,35 @@
  * given configuration as well as helps with authenticating requests.
  */
 public class ThriftUtils {
-  public static TTransport getSocketTransport(String host, int port, int loginTimeout) {
-    return new TSocket(host, port, loginTimeout);
+  public static TTransport getSocketTransport(
+      String host, int port, int connectTimeout, int socketTimeout) {
+    return new TSocket(host, port, socketTimeout, connectTimeout);
   }
 
-  public static TTransport getSSLSocket(String host, int port, int loginTimeout)
-      throws TTransportException {
+  public static TTransport getSSLSocket(
+      String host, int port, int connectTimeout, int socketTimeout) throws TTransportException {
     // The underlying SSLSocket object is bound to host:port with the given SO_TIMEOUT
-    TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port, loginTimeout);
+    TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port, socketTimeout);
+    tSSLSocket.setConnectTimeout(connectTimeout);
     return getSSLSocketWithHttps(tSSLSocket);
   }
 
   public static TTransport getSSLSocket(
-      String host, int port, int loginTimeout, String trustStorePath, String trustStorePassWord)
+      String host,
+      int port,
+      int connectTimeout,
+      int socketTimeout,
+      String trustStorePath,
+      String trustStorePassWord)
       throws TTransportException {
     TSSLTransportFactory.TSSLTransportParameters params =
         new TSSLTransportFactory.TSSLTransportParameters();
     params.setTrustStore(trustStorePath, trustStorePassWord);
     params.requireClientAuth(true);
     // The underlying SSLSocket object is bound to host:port with the given SO_TIMEOUT and
     // SSLContext created with the given params
-    TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port, loginTimeout, params);
+    TSocket tSSLSocket = TSSLTransportFactory.getClientSocket(host, port, socketTimeout, params);
+    tSSLSocket.setConnectTimeout(connectTimeout);
     return getSSLSocketWithHttps(tSSLSocket);
   }