You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by mu...@apache.org on 2023/03/31 13:13:05 UTC
[incubator-streampark] branch dev updated: [Improve] flink shims regexp pattern imporvement (#2540)
This is an automated email from the ASF dual-hosted git repository.
muchunjin pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new 74a216809 [Improve] flink shims regexp pattern imporvement (#2540)
74a216809 is described below
commit 74a216809a26549b112937e1d805ac6e38ab09bb
Author: benjobs <be...@apache.org>
AuthorDate: Fri Mar 31 21:12:59 2023 +0800
[Improve] flink shims regexp pattern imporvement (#2540)
Co-authored-by: benjobs <be...@gmail.com>
---
.../streampark/common/conf/ConfigConst.scala | 2 --
.../console/core/runner/EnvInitializer.java | 2 +-
.../flink/client/impl/YarnApplicationClient.scala | 7 +----
.../flink/packer/pipeline/BuildRequest.scala | 8 +-----
.../streampark/flink/proxy/FlinkShimsProxy.scala | 30 +++++++++++++++++-----
5 files changed, 26 insertions(+), 23 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
index 288769bb6..029b258f6 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
@@ -20,8 +20,6 @@ import java.time.LocalDateTime
object ConfigConst {
- val STREAMPARK_FLINK_SHIMS_JAR_REGEX = "streampark-flink-shims_flink-(1.12|1.13|1.14|1.15|1.16|1.17)_(2.11|2.12)-(.*).jar"
-
/**
* common const
*/
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
index bd9929bdf..2df38ff43 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
@@ -69,7 +69,7 @@ public class EnvInitializer implements ApplicationRunner {
private static final Pattern PATTERN_FLINK_SHIMS_JAR =
Pattern.compile(
- String.format("^%s$", ConfigConst.STREAMPARK_FLINK_SHIMS_JAR_REGEX()),
+ "^streampark-flink-shims_flink-1.1[2-7]_(2.11|2.12)-(.*).jar$",
Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
@Override
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
index d410ec83a..bd61719c0 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
@@ -69,12 +69,7 @@ object YarnApplicationClient extends YarnClientTrait {
case DevelopmentMode.FLINK_SQL =>
val version = submitRequest.flinkVersion.version.split("\\.").map(_.trim.toInt)
version match {
- case Array(1, 12, _) => array += s"${workspace.APP_SHIMS}/flink-1.12"
- case Array(1, 13, _) => array += s"${workspace.APP_SHIMS}/flink-1.13"
- case Array(1, 14, _) => array += s"${workspace.APP_SHIMS}/flink-1.14"
- case Array(1, 15, _) => array += s"${workspace.APP_SHIMS}/flink-1.15"
- case Array(1, 16, _) => array += s"${workspace.APP_SHIMS}/flink-1.16"
- case Array(1, 17, _) => array += s"${workspace.APP_SHIMS}/flink-1.17"
+ case Array(1, v, _) if v >= 12 && v <= 17 => array += s"${workspace.APP_SHIMS}/flink-1.$v"
case _ => throw new UnsupportedOperationException(s"Unsupported flink version: ${submitRequest.flinkVersion}")
}
val jobLib = s"${workspace.APP_WORKSPACE}/${submitRequest.id}/lib"
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
index 8c93acbd3..6d3373d38 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
@@ -18,7 +18,6 @@
package org.apache.streampark.flink.packer.pipeline
import scala.collection.mutable.ArrayBuffer
-
import org.apache.streampark.common.conf.{FlinkVersion, Workspace}
import org.apache.streampark.common.enums.{DevelopmentMode, ExecutionMode}
import org.apache.streampark.flink.kubernetes.model.K8sPodTemplates
@@ -59,12 +58,7 @@ sealed trait FlinkBuildParam extends BuildParam {
providedLibs += {
val version = flinkVersion.version.split("\\.").map(_.trim.toInt)
version match {
- case Array(1, 12, _) => s"${localWorkspace.APP_SHIMS}/flink-1.12"
- case Array(1, 13, _) => s"${localWorkspace.APP_SHIMS}/flink-1.13"
- case Array(1, 14, _) => s"${localWorkspace.APP_SHIMS}/flink-1.14"
- case Array(1, 15, _) => s"${localWorkspace.APP_SHIMS}/flink-1.15"
- case Array(1, 16, _) => s"${localWorkspace.APP_SHIMS}/flink-1.16"
- case Array(1, 17, _) => s"${localWorkspace.APP_SHIMS}/flink-1.17"
+ case Array(1, v, _) if v >= 12 && v <= 17 => s"${localWorkspace.APP_SHIMS}/flink-1.$v"
case _ => throw new UnsupportedOperationException(s"Unsupported flink version: $flinkVersion")
}
}
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 a25d13c98..ff9bd25bb 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
@@ -33,12 +33,10 @@ object FlinkShimsProxy extends Logger {
"(json4s|jackson)(.*).jar",
Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
- private[this] val SHIMS_PATTERN = Pattern.compile(
- ConfigConst.STREAMPARK_FLINK_SHIMS_JAR_REGEX,
- Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
-
private[this] val SHIMS_CLASS_LOADER_CACHE = MutableMap[String, ClassLoader]()
+ private[this] val SHIMS_PATTERN_CACHE = MutableMap[String, Pattern]()
+
private[this] val VERIFY_SQL_CLASS_LOADER_CACHE = MutableMap[String, ClassLoader]()
private[this] def getFlinkShimsResourcePattern(flinkLargeVersion: String) =
@@ -50,11 +48,29 @@ object FlinkShimsProxy extends Logger {
s"streampark-(.*)_$scalaVersion-(.*).jar",
Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
+ private[this] def getShimsLibPattern(flinkVersion: FlinkVersion): Pattern = {
+ flinkVersion.version.split("\\.").map(_.trim.toInt) match {
+ case Array(1, v, _) =>
+ SHIMS_PATTERN_CACHE.getOrElseUpdate(s"1.$v", {
+ val regexp = if (v >= 12 && v <= 14) {
+ s"streampark-flink-shims_flink-1.${v}_(2.11|2.12)-(.*).jar"
+ } else if (v >= 15) {
+ s"streampark-flink-shims_flink-1.${v}_2.12-(.*).jar"
+ } else {
+ throw new UnsupportedOperationException(s"Unsupported flink version: $flinkVersion")
+ }
+ Pattern.compile(regexp, Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
+ })
+ case _ =>
+ throw new UnsupportedOperationException(s"Unsupported flink version: $flinkVersion")
+ }
+ }
+
/**
* Get shimsClassLoader to execute for scala API
*
* @param flinkVersion flinkVersion
- * @param func execute function
+ * @param func execute function
* @tparam T
* @return
*/
@@ -67,7 +83,7 @@ object FlinkShimsProxy extends Logger {
* Get shimsClassLoader to execute for java API
*
* @param flinkVersion flinkVersion
- * @param func execute function
+ * @param func execute function
* @tparam T
* @return
*/
@@ -117,7 +133,7 @@ object FlinkShimsProxy extends Logger {
libPath.listFiles().foreach((jar: File) => {
try {
- val shimsMatcher = SHIMS_PATTERN.matcher(jar.getName)
+ val shimsMatcher = getShimsLibPattern(flinkVersion).matcher(jar.getName)
if (shimsMatcher.matches()) {
if (majorVersion == shimsMatcher.group(1) && scalaVersion == shimsMatcher.group(2)) {
addShimUrl(jar)