[KYUUBI #2500][FOLLOWUP] Resolve flink conf at engine side

### _Why are the changes needed?_

removes hacky method calling from `flink-sql-client`

### _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 #2572 from yaooqinn/2500.

Closes #2500

5531145 [Kent Yao] yarn test
cb7e08f [Kent Yao] wip
e696bdf [Kent Yao] wip
318b762 [Kent Yao] wip

Authored-by: Kent Yao <yao@apache.org>
Signed-off-by: Kent Yao <yao@apache.org>

Author: yaooqinn

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

@@ -19,13 +19,14 @@ package org.apache.kyuubi.engine.flink
 
 import java.io.File
 import java.net.URL
+import java.nio.file.Paths
 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, GenericCLI}
+import org.apache.flink.client.cli.{DefaultCLI, GenericCLI}
 import org.apache.flink.configuration.{Configuration, DeploymentOptions, GlobalConfiguration}
 import org.apache.flink.table.client.SqlClientException
 import org.apache.flink.table.client.gateway.context.DefaultContext
@@ -74,7 +75,17 @@ object FlinkSQLEngine extends Logging {
 
     try {
       Utils.fromCommandLineArgs(args, kyuubiConf)
-      val flinkConfDir = CliFrontend.getConfigurationDirectoryFromEnv
+      val flinkConfDir = sys.env.getOrElse(
+        "FLINK_CONF_DIR", {
+          val flinkHome = sys.env.getOrElse(
+            "FLINK_HOME", {
+              // detect the FLINK_HOME by flink-core*.jar location if unset
+              val jarLoc =
+                classOf[GlobalConfiguration].getProtectionDomain.getCodeSource.getLocation
+              new File(jarLoc.toURI).getParentFile.getParent
+            })
+          Paths.get(flinkHome, "conf").toString
+        })
       val flinkConf = GlobalConfiguration.loadConfiguration(flinkConfDir)
       val flinkConfFromArgs =
         kyuubiConf.getAll.filterKeys(_.startsWith("flink."))
@@ -106,7 +117,7 @@ object FlinkSQLEngine extends Logging {
       val engineContext = new DefaultContext(
         dependencies.asJava,
         flinkConf,
-        List[CustomCommandLine](new GenericCLI(flinkConf, flinkConfDir), new DefaultCLI).asJava)
+        Seq(new GenericCLI(flinkConf, flinkConfDir), new DefaultCLI).asJava)
 
       kyuubiConf.setIfMissing(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
 

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala

@@ -103,8 +103,6 @@ trait ProcBuilder {
 
   protected def env: Map[String, String] = conf.getEnvs
 
-  protected def childProcEnv: Map[String, String] = env
-
   protected val extraEngineLog: Option[OperationLog]
 
   protected val workingDir: Path = {
@@ -137,7 +135,7 @@ trait ProcBuilder {
     val pb = new ProcessBuilder(commands: _*)
 
     val envs = pb.environment()
-    envs.putAll(childProcEnv.asJava)
+    envs.putAll(env.asJava)
     pb.directory(workingDir.toFile)
     pb.redirectError(engineLog)
     pb.redirectOutput(engineLog)

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/flink/FlinkProcessBuilder.scala

@@ -37,17 +37,14 @@ class FlinkProcessBuilder(
     override val proxyUser: String,
     override val conf: KyuubiConf,
     val extraEngineLog: Option[OperationLog] = None)
-  extends ProcBuilder with Logging {
+  extends ProcBuilder {
 
   private val flinkHome: String = getEngineHome(shortName)
 
   override protected def module: String = "kyuubi-flink-sql-engine"
 
   override protected def mainClass: String = "org.apache.kyuubi.engine.flink.FlinkSQLEngine"
 
-  override protected def childProcEnv: Map[String, String] = conf.getEnvs +
-    ("FLINK_CONF_DIR" -> env.getOrElse("FLINK_CONF_DIR", s"$flinkHome${File.separator}conf"))
-
   override protected def commands: Array[String] = {
     val buffer = new ArrayBuffer[String]()
     buffer += executable