[KYUUBI #2032][Subtask] Hive Backend Engine - new APIs with hive-service-rpc 3.1.2 - SetClientInfo

### _Why are the changes needed?_

Hive Backend Engine - new APIs with hive-service-rpc 3.1.2 - SetClientInfo

Implement SetClientInfo based on [HIVE-18240](https://issues.apache.org/jira/browse/HIVE-18240?page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel&focusedCommentId=16959838#comment-16959838).

### _How was this patch tested?_
- [x] 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 #2384 from yangrong688/2032.

Closes #2032

ce63ce07 [yangrong688] fix
2264986b [yangrong688] rebase from apache/master

Authored-by: yangrong688 <yangrong.jxufe@gmail.com>
Signed-off-by: Kent Yao <yao@apache.org>

kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala

@@ -537,7 +537,22 @@ abstract class TFrontendService(name: String)
   override def SetClientInfo(req: TSetClientInfoReq): TSetClientInfoResp = {
     debug(req.toString)
     val resp = new TSetClientInfoResp
-    resp.setStatus(KyuubiSQLException.featureNotSupported().toTStatus)
+    if (req.isSetConfiguration) {
+      val sessionHandle = SessionHandle(req.getSessionHandle)
+      val stringBuilder = new StringBuilder("Client information for ")
+        .append(sessionHandle)
+        .append(": ")
+      val entries = req.getConfiguration.entrySet.asScala.toSeq
+      entries.headOption.foreach(e => {
+        stringBuilder.append(e.getKey).append(" = ").append(e.getValue)
+      })
+      entries.tail.foreach { e =>
+        stringBuilder.append(", ")
+        stringBuilder.append(e.getKey).append(" = ").append(e.getValue)
+      }
+      info(stringBuilder.toString())
+    }
+    resp.setStatus(OK_STATUS)
     resp
   }
 

kyuubi-common/src/test/scala/org/apache/kyuubi/HiveEngineTests.scala

@@ -427,4 +427,19 @@ trait HiveEngineTests extends HiveJDBCTestHelper {
       assert(typeInfo.getInt(DATA_TYPE) === java.sql.Types.OTHER)
     }
   }
+
+  test("test setClientInfo") {
+    assume(SystemUtils.isJavaVersionAtMost(JavaVersion.JAVA_1_8))
+    withJdbcStatement() { statement =>
+      val res = statement.getConnection.getMetaData.getClientInfoProperties
+      assert(res.next())
+      assert(res.getString(1) === "ApplicationName")
+      assert(res.getInt("MAX_LEN") === 1000);
+      assert(!res.next());
+
+      val connection = statement.getConnection
+      connection.setClientInfo("ApplicationName", "test kyuubi hive jdbc")
+      assert(connection.getClientInfo("ApplicationName") == "test kyuubi hive jdbc")
+    }
+  }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/operation/SparkMetadataTests.scala

@@ -396,7 +396,6 @@ trait SparkMetadataTests extends HiveJDBCTestHelper {
         () => metaData.getRowIdLifetime,
         () => metaData.supportsStoredFunctionsUsingCallSyntax,
         () => metaData.autoCommitFailureClosesAllResultSets,
-        () => metaData.getClientInfoProperties,
         () => metaData.getFunctionColumns("", "%", "%", "%"),
         () => metaData.getPseudoColumns("", "%", "%", "%"),
         () => metaData.generatedKeyAlwaysReturned).foreach { func =>
@@ -405,6 +404,7 @@ trait SparkMetadataTests extends HiveJDBCTestHelper {
       }
 
       assert(metaData.allTablesAreSelectable)
+      assert(metaData.getClientInfoProperties.next)
       assert(metaData.getDatabaseProductName === "Apache Kyuubi (Incubating)")
       assert(metaData.getDatabaseProductVersion === KYUUBI_VERSION)
       assert(metaData.getDriverName === "Kyuubi Project Hive JDBC Shaded Client")

kyuubi-hive-jdbc-shaded/pom.xml

@@ -148,6 +148,11 @@
             <artifactId>commons-lang</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-framework</artifactId>
@@ -291,6 +296,10 @@
                             <pattern>org.apache.commons.lang</pattern>
                             <shadedPattern>${kyuubi.shade.packageName}.org.apache.commons.lang</shadedPattern>
                         </relocation>
+                        <relocation>
+                            <pattern>org.apache.commons.lang3</pattern>
+                            <shadedPattern>${kyuubi.shade.packageName}.org.apache.commons.lang3</shadedPattern>
+                        </relocation>
                         <relocation>
                             <pattern>org.apache.curator</pattern>
                             <shadedPattern>${kyuubi.shade.packageName}.org.apache.curator</shadedPattern>

kyuubi-hive-jdbc-shaded/src/main/resources/META-INF/LICENSE

@@ -219,6 +219,7 @@ com.google.guava:failureaccess
 com.google.guava:guava
 commons-codec:commons-codec
 commons-lang:commons-lang
+org.apache.commons:commons-lang3
 org.apache.curator:curator-framework
 org.apache.curator:curator-client
 org.apache.httpcomponents:httpclient

kyuubi-hive-jdbc/pom.xml

@@ -147,6 +147,11 @@
             <artifactId>commons-lang</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

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

@@ -105,6 +105,7 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
   private volatile boolean launchEngineOpCompleted = false;
 
   private boolean isBeeLineMode;
+  private Properties clientInfo;
 
   public KyuubiConnection(String uri, Properties info) throws SQLException {
     setupLoginTimeout();
@@ -1119,8 +1120,7 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
 
   @Override
   public Properties getClientInfo() throws SQLException {
-    // TODO Auto-generated method stub
-    throw new SQLFeatureNotSupportedException("Method not supported");
+    return clientInfo == null ? new Properties() : clientInfo;
   }
 
   /*
@@ -1131,8 +1131,8 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
 
   @Override
   public String getClientInfo(String name) throws SQLException {
-    // TODO Auto-generated method stub
-    throw new SQLFeatureNotSupportedException("Method not supported");
+    if (clientInfo == null) return null;
+    return clientInfo.getProperty(name);
   }
 
   /*
@@ -1463,8 +1463,8 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
 
   @Override
   public void setClientInfo(Properties properties) throws SQLClientInfoException {
-    // TODO Auto-generated method stub
-    throw new SQLClientInfoException("Method not supported", null);
+    clientInfo = properties;
+    setClientInfo();
   }
 
   /*
@@ -1475,8 +1475,30 @@ public class KyuubiConnection implements java.sql.Connection, KyuubiLoggable {
 
   @Override
   public void setClientInfo(String name, String value) throws SQLClientInfoException {
-    // TODO Auto-generated method stub
-    throw new SQLClientInfoException("Method not supported", null);
+    if (clientInfo == null) {
+      clientInfo = new Properties();
+    }
+    clientInfo.put(name, value);
+    setClientInfo();
+  }
+
+  private void setClientInfo() throws SQLClientInfoException {
+    TSetClientInfoReq req = new TSetClientInfoReq(sessHandle);
+    Map<String, String> map = new HashMap<>();
+    if (clientInfo != null) {
+      for (Entry<Object, Object> e : clientInfo.entrySet()) {
+        if (e.getKey() == null || e.getValue() == null) continue;
+        map.put(e.getKey().toString(), e.getValue().toString());
+      }
+    }
+    req.setConfiguration(map);
+    try {
+      TSetClientInfoResp openResp = client.SetClientInfo(req);
+      Utils.verifySuccess(openResp.getStatus());
+    } catch (TException | SQLException e) {
+      LOG.error("Error setting client info", e);
+      throw new SQLClientInfoException("Error setting client info", null, e);
+    }
   }
 
   /*

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

@@ -23,12 +23,16 @@ import java.sql.ResultSet;
 import java.sql.RowIdLifetime;
 import java.sql.SQLException;
 import java.sql.SQLFeatureNotSupportedException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.List;
 import java.util.jar.Attributes;
 import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hive.service.cli.GetInfoType;
 import org.apache.hive.service.cli.HiveSQLException;
+import org.apache.hive.service.cli.TableSchema;
 import org.apache.hive.service.rpc.thrift.*;
 import org.apache.kyuubi.jdbc.KyuubiHiveDriver;
 import org.apache.thrift.TException;
@@ -122,8 +126,46 @@ public class KyuubiDatabaseMetaData implements DatabaseMetaData {
         .build();
   }
 
+  private static final class ClientInfoPropertiesResultSet extends KyuubiMetaDataResultSet<Object> {
+    private static final String[] COLUMNS = {"NAME", "MAX_LEN", "DEFAULT_VALUE", "DESCRIPTION"};
+    private static final String[] COLUMN_TYPES = {"STRING", "INT", "STRING", "STRING"};
+
+    private static final Object[][] DATA = {
+      {"ApplicationName", 1000, null, null},
+    };
+    private int index = -1;
+
+    public ClientInfoPropertiesResultSet() throws SQLException {
+      super(Arrays.asList(COLUMNS), Arrays.asList(COLUMN_TYPES), null);
+      List<FieldSchema> fieldSchemas = new ArrayList<>(COLUMNS.length);
+      for (int i = 0; i < COLUMNS.length; ++i) {
+        fieldSchemas.add(new FieldSchema(COLUMNS[i], COLUMN_TYPES[i], null));
+      }
+      setSchema(new TableSchema(fieldSchemas));
+    }
+
+    @Override
+    public boolean next() throws SQLException {
+      if ((++index) >= DATA.length) return false;
+      row = Arrays.copyOf(DATA[index], DATA[index].length);
+      return true;
+    }
+
+    public <T> T getObject(String columnLabel, Class<T> type) throws SQLException {
+      for (int i = 0; i < COLUMNS.length; ++i) {
+        if (COLUMNS[i].equalsIgnoreCase(columnLabel)) return getObject(i, type);
+      }
+      throw new SQLException("No column " + columnLabel);
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> T getObject(int columnIndex, Class<T> type) throws SQLException {
+      return (T) super.getObject(columnIndex);
+    }
+  }
+
   public ResultSet getClientInfoProperties() throws SQLException {
-    throw new SQLFeatureNotSupportedException("Method not supported");
+    return new ClientInfoPropertiesResultSet();
   }
 
   public ResultSet getColumnPrivileges(