[KYUUBI #1685] Support jar and lib start option

<!--
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.
-->
Support jar and lib start option.

### _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 #1818 from SteNicholas/KYUUBI-1685.

Closes #1685

dd5a704 [SteNicholas] [KYUUBI #1685] Support jar and lib start option

Authored-by: SteNicholas <programgeek@163.com>
Signed-off-by: Kent Yao <yao@apache.org>

Author: SteNicholas

externals/kyuubi-flink-sql-engine/pom.xml

@@ -198,11 +198,12 @@
               </includes>
             </relocation>
             <relocation>
-              <pattern>org.apache.commons</pattern>
-              <shadedPattern>${kyuubi.shade.packageName}.org.apache.commons</shadedPattern>
-              <includes>
-                <include>org.apache.commons.**</include>
-              </includes>
+              <pattern>org.apache.commons.codec</pattern>
+              <shadedPattern>${kyuubi.shade.packageName}.org.apache.commons.codec</shadedPattern>
+            </relocation>
+            <relocation>
+              <pattern>org.apache.commons.lang</pattern>
+              <shadedPattern>${kyuubi.shade.packageName}.org.apache.commons.lang</shadedPattern>
             </relocation>
             <relocation>
               <pattern>io.netty</pattern>

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkEngineUtils.scala

@@ -17,9 +17,17 @@
 
 package org.apache.kyuubi.engine.flink
 
+import java.io.File
+import java.net.URL
+
+import scala.collection.JavaConverters._
+
+import org.apache.commons.cli.{CommandLine, DefaultParser, Option, Options, ParseException}
+import org.apache.flink.core.fs.Path
 import org.apache.flink.runtime.util.EnvironmentInformation
 import org.apache.flink.table.client.SqlClientException
-import org.apache.flink.table.client.cli.{CliOptions, CliOptionsParser}
+import org.apache.flink.table.client.cli.CliOptions
+import org.apache.flink.table.client.cli.CliOptionsParser._
 import org.apache.flink.table.client.gateway.context.SessionContext
 import org.apache.flink.table.client.gateway.local.LocalExecutor
 
@@ -28,6 +36,7 @@ import org.apache.kyuubi.Logging
 object FlinkEngineUtils extends Logging {
 
   val MODE_EMBEDDED = "embedded"
+  val EMBEDDED_MODE_CLIENT_OPTIONS: Options = getEmbeddedModeClientOptions(new Options);
 
   def checkFlinkVersion(): Unit = {
     val flinkVersion = EnvironmentInformation.getVersion
@@ -40,19 +49,69 @@ object FlinkEngineUtils extends Logging {
     val (mode, modeArgs) =
       if (args.isEmpty || args(0).startsWith("-")) (MODE_EMBEDDED, args)
       else (args(0), args.drop(1))
-    // TODO remove requirement of flink-python
-    val options = CliOptionsParser.parseEmbeddedModeClient(modeArgs)
-    mode match {
-      case MODE_EMBEDDED if options.isPrintHelp => CliOptionsParser.printHelpEmbeddedModeClient()
-      case MODE_EMBEDDED =>
-      case _ => throw new SqlClientException("Other mode is not supported yet.")
+    val options = parseEmbeddedModeClient(modeArgs)
+    if (mode == MODE_EMBEDDED) {
+      if (options.isPrintHelp) {
+        printHelpEmbeddedModeClient()
+      }
+      options
+    } else {
+      throw new SqlClientException("Other mode is not supported yet.")
     }
-    options
   }
 
   def getSessionContext(localExecutor: LocalExecutor, sessionId: String): SessionContext = {
     val method = classOf[LocalExecutor].getDeclaredMethod("getSessionContext", classOf[String])
     method.setAccessible(true)
     method.invoke(localExecutor, sessionId).asInstanceOf[SessionContext]
   }
+
+  def parseEmbeddedModeClient(args: Array[String]): CliOptions =
+    try {
+      val parser = new DefaultParser
+      val line = parser.parse(EMBEDDED_MODE_CLIENT_OPTIONS, args, true)
+      val jarUrls = checkUrls(line, OPTION_JAR)
+      val libraryUrls = checkUrls(line, OPTION_LIBRARY)
+      new CliOptions(
+        line.hasOption(OPTION_HELP.getOpt),
+        checkSessionId(line),
+        checkUrl(line, OPTION_INIT_FILE),
+        checkUrl(line, OPTION_FILE),
+        if (jarUrls != null && jarUrls.nonEmpty) jarUrls.asJava else null,
+        if (libraryUrls != null && libraryUrls.nonEmpty) libraryUrls.asJava else null,
+        line.getOptionValue(OPTION_UPDATE.getOpt),
+        line.getOptionValue(OPTION_HISTORY.getOpt),
+        null)
+    } catch {
+      case e: ParseException =>
+        throw new SqlClientException(e.getMessage)
+    }
+
+  def checkSessionId(line: CommandLine): String = {
+    val sessionId = line.getOptionValue(OPTION_SESSION.getOpt)
+    if (sessionId != null && !sessionId.matches("[a-zA-Z0-9_\\-.]+")) {
+      throw new SqlClientException("Session identifier must only consists of 'a-zA-Z0-9_-.'.")
+    } else sessionId
+  }
+
+  def checkUrl(line: CommandLine, option: Option): URL = {
+    val urls: List[URL] = checkUrls(line, option)
+    if (urls != null && urls.nonEmpty) urls.head
+    else null
+  }
+
+  def checkUrls(line: CommandLine, option: Option): List[URL] = {
+    if (line.hasOption(option.getOpt)) {
+      line.getOptionValues(option.getOpt).distinct.map((url: String) => {
+        checkFilePath(url)
+        try Path.fromLocalFile(new File(url).getAbsoluteFile).toUri.toURL
+        catch {
+          case e: Exception =>
+            throw new SqlClientException(
+              "Invalid path for option '" + option.getLongOpt + "': " + url,
+              e)
+        }
+      }).toList
+    } else null
+  }
 }

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/FlinkSQLEngine.scala

@@ -17,17 +17,24 @@
 
 package org.apache.kyuubi.engine.flink
 
+import java.io.File
+import java.net.URL
 import java.time.Instant
 import java.util.concurrent.CountDownLatch
 
 import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
 
 import org.apache.flink.client.cli.{CliFrontend, CustomCommandLine, DefaultCLI}
-import org.apache.flink.configuration.{DeploymentOptions, GlobalConfiguration}
+import org.apache.flink.configuration.DeploymentOptions
+import org.apache.flink.configuration.GlobalConfiguration
+import org.apache.flink.table.client.SqlClientException
 import org.apache.flink.table.client.gateway.context.DefaultContext
+import org.apache.flink.util.JarUtils
 
-import org.apache.kyuubi.Logging
-import org.apache.kyuubi.Utils.{addShutdownHook, currentUser, FLINK_ENGINE_SHUTDOWN_PRIORITY}
+import org.apache.kyuubi.{KyuubiSQLException, Logging}
+import org.apache.kyuubi.Utils.{addShutdownHook, FLINK_ENGINE_SHUTDOWN_PRIORITY}
+import org.apache.kyuubi.Utils.currentUser
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.engine.flink.FlinkSQLEngine.{countDownLatch, currentEngine}
 import org.apache.kyuubi.service.Serverable
@@ -93,8 +100,13 @@ object FlinkSQLEngine extends Logging {
           debug(s"Skip generating app name for execution target $other")
       }
 
+      val cliOptions = FlinkEngineUtils.parseCliOptions(args)
+      val jars = if (cliOptions.getJars != null) cliOptions.getJars.asScala else List.empty
+      val libDirs =
+        if (cliOptions.getLibraryDirs != null) cliOptions.getLibraryDirs.asScala else List.empty
+      val dependencies = discoverDependencies(jars, libDirs)
       val engineContext = new DefaultContext(
-        List.empty.asJava,
+        dependencies.asJava,
         flinkConf,
         List[CustomCommandLine](new DefaultCLI).asJava)
 
@@ -124,4 +136,37 @@ object FlinkSQLEngine extends Logging {
       addShutdownHook(() => engine.stop(), FLINK_ENGINE_SHUTDOWN_PRIORITY + 1)
     }
   }
+
+  private def discoverDependencies(
+      jars: Seq[URL],
+      libraries: Seq[URL]): List[URL] = {
+    try {
+      var dependencies: ListBuffer[URL] = ListBuffer()
+      // find jar files
+      jars.foreach { url =>
+        JarUtils.checkJarFile(url)
+        dependencies = dependencies += url
+      }
+      // find jar files in library directories
+      libraries.foreach { libUrl =>
+        val dir: File = new File(libUrl.toURI)
+        if (!dir.isDirectory) throw new SqlClientException("Directory expected: " + dir)
+        else if (!dir.canRead) throw new SqlClientException("Directory cannot be read: " + dir)
+        val files: Array[File] = dir.listFiles
+        if (files == null) throw new SqlClientException("Directory cannot be read: " + dir)
+        files.foreach { f =>
+          // only consider jars
+          if (f.isFile && f.getAbsolutePath.toLowerCase.endsWith(".jar")) {
+            val url: URL = f.toURI.toURL
+            JarUtils.checkJarFile(url)
+            dependencies = dependencies += url
+          }
+        }
+      }
+      dependencies.toList
+    } catch {
+      case e: Exception =>
+        throw KyuubiSQLException(s"Could not load all required JAR files.", e)
+    }
+  }
 }

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/WithFlinkSQLEngine.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.engine.flink
 
+import java.nio.file.Files
+
 import scala.collection.JavaConverters._
 
 import org.apache.flink.client.cli.{CustomCommandLine, DefaultCLI}
@@ -26,6 +28,7 @@ import org.apache.flink.table.client.gateway.context.DefaultContext
 
 import org.apache.kyuubi.KyuubiFunSuite
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.engine.flink.util.TestUserClassLoaderJar
 
 trait WithFlinkSQLEngine extends KyuubiFunSuite {
 
@@ -38,6 +41,17 @@ trait WithFlinkSQLEngine extends KyuubiFunSuite {
 
   protected var connectionUrl: String = _
 
+  protected val GENERATED_UDF_CLASS: String = "LowerUDF"
+
+  protected val GENERATED_UDF_CODE: String =
+    s"""
+      public class $GENERATED_UDF_CLASS extends org.apache.flink.table.functions.ScalarFunction {
+        public String eval(String str) {
+          return str.toLowerCase();
+        }
+      }
+     """
+
   override def beforeAll(): Unit = {
     startMiniCluster()
     startFlinkEngine()
@@ -55,8 +69,13 @@ trait WithFlinkSQLEngine extends KyuubiFunSuite {
       System.setProperty(k, v)
       kyuubiConf.set(k, v)
     }
+    val udfJar = TestUserClassLoaderJar.createJarFile(
+      Files.createTempDirectory("test-jar").toFile,
+      "test-classloader-udf.jar",
+      GENERATED_UDF_CLASS,
+      GENERATED_UDF_CODE)
     val engineContext = new DefaultContext(
-      List.empty.asJava,
+      List(udfJar.toURI.toURL).asJava,
       flinkConfig,
       List[CustomCommandLine](new DefaultCLI).asJava)
     FlinkSQLEngine.startEngine(engineContext)

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala

@@ -717,11 +717,21 @@ class FlinkOperationSuite extends WithFlinkSQLEngine with HiveJDBCTestHelper {
       // thus read all rows to find the desired one
       var success = false
       while (resultSet.next()) {
-        if (resultSet.getString(1) == "pipeline.jars" && resultSet.getString(2) == "") {
+        if (resultSet.getString(1) == "pipeline.jars" &&
+          !resultSet.getString(2).contains("my.jar")) {
           success = true
         }
       }
       assert(success)
     })
   }
+
+  test("execute statement - select udf") {
+    withJdbcStatement() { statement =>
+      statement.execute(s"create function $GENERATED_UDF_CLASS AS '$GENERATED_UDF_CLASS'")
+      val resultSet = statement.executeQuery(s"select $GENERATED_UDF_CLASS('A')")
+      assert(resultSet.next())
+      assert(resultSet.getString(1) === "a")
+    }
+  }
 }

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/util/TestUserClassLoaderJar.scala

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.engine.flink.util
+
+import java.io.{File, FileOutputStream}
+import java.nio.file.Paths
+import java.util.Collections
+import java.util.jar.{JarEntry, JarOutputStream}
+import javax.tools.{DiagnosticCollector, JavaFileObject, ToolProvider}
+
+import scala.collection.JavaConverters._
+
+import org.apache.flink.util.FileUtils
+
+object TestUserClassLoaderJar {
+
+  /**
+   * Pack the generated UDF class into a JAR and return the path of the JAR.
+   */
+  def createJarFile(tmpDir: File, jarName: String, className: String, javaCode: String): File = {
+    // write class source code to file
+    val javaFile = Paths.get(tmpDir.toString, className + ".java").toFile
+    // no inspection ResultOfMethodCallIgnored
+    javaFile.createNewFile
+    FileUtils.writeFileUtf8(javaFile, javaCode)
+    // compile class source code
+    val diagnostics = new DiagnosticCollector[JavaFileObject]
+    val compiler = ToolProvider.getSystemJavaCompiler
+    val fileManager = compiler.getStandardFileManager(diagnostics, null, null)
+    val compilationUnit =
+      fileManager.getJavaFileObjectsFromFiles(Collections.singletonList(javaFile))
+    val task =
+      compiler.getTask(null, fileManager, diagnostics, List.empty.asJava, null, compilationUnit)
+    task.call
+    // pack class file to jar
+    val classFile = Paths.get(tmpDir.toString, className + ".class").toFile
+    val jarFile = Paths.get(tmpDir.toString, jarName).toFile
+    val jos = new JarOutputStream(new FileOutputStream(jarFile))
+    val jarEntry = new JarEntry(className + ".class")
+    jos.putNextEntry(jarEntry)
+    val classBytes = FileUtils.readAllBytes(classFile.toPath)
+    jos.write(classBytes)
+    jos.closeEntry()
+    jos.close()
+    jarFile
+  }
+}