You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/10/28 07:09:43 UTC
(incubator-streampark) branch dev-2.1.2 updated: [Improve] remote submit mode improvement
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev-2.1.2
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev-2.1.2 by this push:
new e1148d8a5 [Improve] remote submit mode improvement
e1148d8a5 is described below
commit e1148d8a5e9e62b6842debdaa3483f5eba7cceb3
Author: benjobs <be...@gmail.com>
AuthorDate: Sat Oct 28 15:09:31 2023 +0800
[Improve] remote submit mode improvement
---
.../apache/streampark/flink/client/FlinkClient.scala | 14 ++++++++++++++
.../streampark/flink/client/impl/RemoteClient.scala | 3 ++-
.../streampark/flink/proxy/FlinkShimsProxy.scala | 19 ++++++++++++-------
3 files changed, 28 insertions(+), 8 deletions(-)
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
index c2ea9ba48..14d063e88 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
@@ -18,10 +18,13 @@
package org.apache.streampark.flink.client
import org.apache.streampark.common.conf.FlinkVersion
+import org.apache.streampark.common.enums.ExecutionMode
import org.apache.streampark.common.util.Logger
import org.apache.streampark.flink.client.bean._
import org.apache.streampark.flink.proxy.FlinkShimsProxy
+import java.net.URL
+
import scala.language.{implicitConversions, reflectiveCalls}
import scala.reflect.ClassTag
@@ -69,9 +72,20 @@ object FlinkClient extends Logger {
request: Object,
flinkVersion: FlinkVersion,
requestBody: (String, String)): T = {
+
+ val classpath = request match {
+ case request: SubmitRequest =>
+ request.executionMode match {
+ case ExecutionMode.REMOTE => List(request.userJarFile.toURI.toURL)
+ case _ => List.empty[URL]
+ }
+ case _ => List.empty[URL]
+ }
+
flinkVersion.checkVersion()
FlinkShimsProxy.proxy(
flinkVersion,
+ classpath,
(classLoader: ClassLoader) => {
val submitClass = classLoader.loadClass(FLINK_CLIENT_HANDLER_CLASS_NAME)
val requestClass = classLoader.loadClass(requestBody._1)
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteClient.scala
index c986bc8d0..5569662a0 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteClient.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.client.impl
-import org.apache.streampark.common.util.Utils
+import org.apache.streampark.common.util.{ClassLoaderUtils, Utils}
import org.apache.streampark.flink.client.`trait`.FlinkClientTrait
import org.apache.streampark.flink.client.bean.{CancelRequest, CancelResponse, SavepointRequestTrait, SavepointResponse, SubmitRequest, SubmitResponse, TriggerSavepointRequest}
import org.apache.streampark.flink.client.tool.FlinkSessionSubmitHelper
@@ -46,6 +46,7 @@ object RemoteClient extends FlinkClientTrait {
override def doSubmit(
submitRequest: SubmitRequest,
flinkConfig: Configuration): SubmitResponse = {
+
// 2) submit job
super.trySubmit(submitRequest, flinkConfig, submitRequest.userJarFile)(restApiSubmit)(
jobGraphSubmit)
diff --git a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
index 56dede3e6..016c51873 100644
--- a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
+++ b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala
@@ -22,6 +22,7 @@ import org.apache.streampark.common.util.{ClassLoaderUtils, Logger, Utils}
import java.io.{ByteArrayInputStream, ByteArrayOutputStream, File, ObjectOutputStream}
import java.net.URL
+import java.sql.DriverManager
import java.util.function.{Function => JavaFunc}
import java.util.regex.Pattern
@@ -55,8 +56,8 @@ object FlinkShimsProxy extends Logger {
* @tparam T
* @return
*/
- def proxy[T](flinkVersion: FlinkVersion, func: ClassLoader => T): T = {
- val shimsClassLoader = getFlinkShimsClassLoader(flinkVersion)
+ def proxy[T](flinkVersion: FlinkVersion, classPaths: List[URL], func: ClassLoader => T): T = {
+ val shimsClassLoader = getFlinkShimsClassLoader(flinkVersion, classPaths)
ClassLoaderUtils.runAsClassLoader[T](shimsClassLoader, () => func(shimsClassLoader))
}
@@ -70,8 +71,11 @@ object FlinkShimsProxy extends Logger {
* @tparam T
* @return
*/
- def proxy[T](flinkVersion: FlinkVersion, func: JavaFunc[ClassLoader, T]): T = {
- val shimsClassLoader = getFlinkShimsClassLoader(flinkVersion)
+ def proxy[T](
+ flinkVersion: FlinkVersion,
+ classPaths: List[URL],
+ func: JavaFunc[ClassLoader, T]): T = {
+ val shimsClassLoader = getFlinkShimsClassLoader(flinkVersion, classPaths)
ClassLoaderUtils.runAsClassLoader[T](shimsClassLoader, () => func(shimsClassLoader))
}
@@ -157,14 +161,15 @@ object FlinkShimsProxy extends Logger {
ClassLoaderUtils.runAsClassLoader[T](shimsClassLoader, () => func(shimsClassLoader))
}
- private[this] def getFlinkShimsClassLoader(flinkVersion: FlinkVersion): ClassLoader = {
+ private[this] def getFlinkShimsClassLoader(
+ flinkVersion: FlinkVersion,
+ classPaths: List[URL]): ClassLoader = {
logInfo(s"add flink shims urls classloader,flink version: $flinkVersion")
-
SHIMS_CLASS_LOADER_CACHE.getOrElseUpdate(
s"${flinkVersion.fullVersion}", {
// 1) flink/lib
val libURL = getFlinkHomeLib(flinkVersion.flinkHome, "lib", !_.getName.startsWith("log4j"))
- val shimsUrls = ListBuffer[URL](libURL: _*)
+ val shimsUrls = ListBuffer[URL](libURL: _*) ++ classPaths
// 2) add all shims jar
addShimsUrls(