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/04/04 14:11:45 UTC
[incubator-streampark] branch dev updated: [Improve] Flink version check improvement (#2566)
This is an automated email from the ASF dual-hosted git repository.
benjobs 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 5a4d07c70 [Improve] Flink version check improvement (#2566)
5a4d07c70 is described below
commit 5a4d07c705ab99eef4344559afa6bbf9647fb01f
Author: benjobs <be...@apache.org>
AuthorDate: Tue Apr 4 22:11:37 2023 +0800
[Improve] Flink version check improvement (#2566)
* [Improve] Flink version check improvement
* [Bug] flink shims lib matche bug fixed.
---------
Co-authored-by: benjobs <be...@gmail.com>
---
.../streampark/common/conf/FlinkVersion.scala | 12 +++
.../ApplicationBuildPipelineController.java | 22 +++++-
.../streampark/flink/client/FlinkClient.scala | 1 +
.../flink/client/impl/YarnApplicationClient.scala | 6 +-
.../flink/packer/pipeline/BuildRequest.scala | 8 +-
.../flink/proxy/ChildFirstClassLoader.scala | 9 +--
.../streampark/flink/proxy/FlinkShimsProxy.scala | 86 ++++++++--------------
7 files changed, 67 insertions(+), 77 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala
index 0f247fed6..1cb86ee1d 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/conf/FlinkVersion.scala
@@ -109,6 +109,18 @@ class FlinkVersion(val flinkHome: String) extends java.io.Serializable with Logg
distJar.head
}
+ def checkVersion(throwException: Boolean = true): Boolean = {
+ version.split("\\.").map(_.trim.toInt) match {
+ case Array(1, v, _) if v >= 12 && v <= 17 => true
+ case _ =>
+ if (throwException) {
+ throw new UnsupportedOperationException(s"Unsupported flink version: $version")
+ } else {
+ false
+ }
+ }
+ }
+
// StreamPark flink shims version, like "streampark-flink-shims_flink-1.13"
lazy val shimsVersion: String = s"streampark-flink-shims_flink-$majorVersion"
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
index ca9136885..d2b560dac 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
@@ -25,10 +25,12 @@ import org.apache.streampark.console.core.bean.AppBuildDockerResolvedDetail;
import org.apache.streampark.console.core.entity.AppBuildPipeline;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.ApplicationLog;
+import org.apache.streampark.console.core.entity.FlinkEnv;
import org.apache.streampark.console.core.enums.Operation;
import org.apache.streampark.console.core.service.AppBuildPipeService;
import org.apache.streampark.console.core.service.ApplicationLogService;
import org.apache.streampark.console.core.service.ApplicationService;
+import org.apache.streampark.console.core.service.FlinkEnvService;
import org.apache.streampark.console.core.service.FlinkSqlService;
import org.apache.streampark.flink.packer.pipeline.DockerResolvedSnapshot;
import org.apache.streampark.flink.packer.pipeline.PipelineType;
@@ -66,6 +68,8 @@ public class ApplicationBuildPipelineController {
@Autowired private ApplicationLogService applicationLogService;
+ @Autowired private FlinkEnvService flinkEnvService;
+
/**
* Release application building pipeline.
*
@@ -100,11 +104,15 @@ public class ApplicationBuildPipelineController {
try {
Application app = applicationService.getById(appId);
- ApplicationLog applicationLog = new ApplicationLog();
- applicationLog.setOptionName(Operation.RELEASE.getValue());
- applicationLog.setAppId(app.getId());
- applicationLog.setOptionTime(new Date());
+ // 1) check flink version
+ FlinkEnv env = flinkEnvService.getById(app.getVersionId());
+ boolean checkVersion = env.getFlinkVersion().checkVersion(false);
+ if (!checkVersion) {
+ throw new ApiAlertException(
+ "Unsupported flink version: " + env.getFlinkVersion().version());
+ }
+ // 2) check env
boolean envOk = applicationService.checkEnv(app);
if (!envOk) {
throw new ApiAlertException(
@@ -118,6 +126,12 @@ public class ApplicationBuildPipelineController {
// check if you need to go through the build process (if the jar and pom have changed,
// you need to go through the build process, if other common parameters are modified,
// you don't need to go through the build process)
+
+ ApplicationLog applicationLog = new ApplicationLog();
+ applicationLog.setOptionName(Operation.RELEASE.getValue());
+ applicationLog.setAppId(app.getId());
+ applicationLog.setOptionTime(new Date());
+
boolean needBuild = applicationService.checkBuildAndUpdate(app);
if (!needBuild) {
applicationLog.setSuccess(true);
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 14c5f4b8a..328995036 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
@@ -60,6 +60,7 @@ object FlinkClient extends Logger {
}
private[this] def proxy[T: ClassTag](request: Object, flinkVersion: FlinkVersion, requestBody: (String, String)): T = {
+ flinkVersion.checkVersion()
FlinkShimsProxy.proxy(flinkVersion, (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/YarnApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
index bd61719c0..a4d6015b1 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
@@ -67,11 +67,7 @@ object YarnApplicationClient extends YarnClientTrait {
submitRequest.hdfsWorkspace.appPlugins)
submitRequest.developmentMode match {
case DevelopmentMode.FLINK_SQL =>
- val version = submitRequest.flinkVersion.version.split("\\.").map(_.trim.toInt)
- version match {
- 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}")
- }
+ array += s"${workspace.APP_SHIMS}/flink-${submitRequest.flinkVersion.majorVersion}"
val jobLib = s"${workspace.APP_WORKSPACE}/${submitRequest.id}/lib"
if (HdfsUtils.exists(jobLib)) {
array += jobLib
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 6d3373d38..dd4389e90 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
@@ -55,13 +55,7 @@ sealed trait FlinkBuildParam extends BuildParam {
localWorkspace.APP_PLUGINS,
customFlinkUserJar)
if (developmentMode == DevelopmentMode.FLINK_SQL) {
- providedLibs += {
- val version = flinkVersion.version.split("\\.").map(_.trim.toInt)
- version match {
- 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")
- }
- }
+ providedLibs += s"${localWorkspace.APP_SHIMS}/flink-${flinkVersion.majorVersion}"
}
dependencyInfo.merge(providedLibs.toSet)
}
diff --git a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/ChildFirstClassLoader.scala b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/ChildFirstClassLoader.scala
index 7123691f6..eec159cb8 100644
--- a/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/ChildFirstClassLoader.scala
+++ b/streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/ChildFirstClassLoader.scala
@@ -34,7 +34,7 @@ import scala.util.Try
*/
class ChildFirstClassLoader(urls: Array[URL], parent: ClassLoader, flinkResourcePattern: Pattern, classLoadingExceptionHandler: Consumer[Throwable])
- extends URLClassLoader(urls, parent) {
+ extends URLClassLoader(urls, parent) {
ClassLoader.registerAsParallelCapable()
@@ -66,10 +66,9 @@ class ChildFirstClassLoader(urls: Array[URL], parent: ClassLoader, flinkResource
@throws[ClassNotFoundException]
override def loadClass(name: String, resolve: Boolean): Class[_] = {
- try this.synchronized {
- return this.loadClassWithoutExceptionHandling(name, resolve)
- }
- catch {
+ try {
+ this.synchronized(this.loadClassWithoutExceptionHandling(name, resolve))
+ } catch {
case e: Throwable =>
classLoadingExceptionHandler.accept(e)
throw e
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 ff9bd25bb..d7a420ec8 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
@@ -29,42 +29,21 @@ import org.apache.streampark.common.util.{ClassLoaderUtils, Logger, Utils}
object FlinkShimsProxy extends Logger {
- private[this] val INCLUDE_PATTERN: Pattern = Pattern.compile(
- "(json4s|jackson)(.*).jar",
- 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] val INCLUDE_PATTERN: Pattern = Pattern.compile(
+ "(json4s|jackson)(.*).jar",
+ Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
+
private[this] def getFlinkShimsResourcePattern(flinkLargeVersion: String) =
Pattern.compile(
s"flink-(.*)-$flinkLargeVersion(.*).jar",
Pattern.CASE_INSENSITIVE | Pattern.DOTALL)
- private[this] def getStreamParkLibPattern(scalaVersion: String): Pattern = Pattern.compile(
- 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")
- }
- }
+ private[this] lazy val FLINK_SHIMS_PREFIX = "streampark-flink-shims_flink"
/**
* Get shimsClassLoader to execute for scala API
@@ -92,7 +71,6 @@ object FlinkShimsProxy extends Logger {
ClassLoaderUtils.runAsClassLoader[T](shimsClassLoader, () => func(shimsClassLoader))
}
- // flink 1.12 1.13~1.14 1.15 1.16 parseSql class exist in different dependencies,
// need to load all flink-table dependencies compatible with different versions
def getVerifySqlLibClassLoader(flinkVersion: FlinkVersion): ClassLoader = {
logInfo(s"add verify sql lib,flink version: $flinkVersion")
@@ -107,13 +85,12 @@ object FlinkShimsProxy extends Logger {
val shimsUrls = ListBuffer[URL](libTableURL ++ optTableURL: _*)
// 3) add only streampark shims jar
- addShimsUrls(
- flinkVersion,
- file => {
- if (file != null && file.getName.startsWith("streampark-flink-shims")) {
- shimsUrls += file.toURI.toURL
- }
- })
+ addShimsUrls(flinkVersion, file => {
+ if (file.getName.startsWith("streampark-flink-shims")) {
+ shimsUrls += file.toURI.toURL
+ }
+ })
+
new ChildFirstClassLoader(
shimsUrls.toArray,
Thread.currentThread().getContextClassLoader,
@@ -127,29 +104,28 @@ object FlinkShimsProxy extends Logger {
val libPath = new File(s"$appHome/lib")
require(libPath.exists())
+
val majorVersion = flinkVersion.majorVersion
val scalaVersion = flinkVersion.scalaVersion
- val streamParkMatcher = getStreamParkLibPattern(scalaVersion)
libPath.listFiles().foreach((jar: File) => {
- try {
- val shimsMatcher = getShimsLibPattern(flinkVersion).matcher(jar.getName)
- if (shimsMatcher.matches()) {
- if (majorVersion == shimsMatcher.group(1) && scalaVersion == shimsMatcher.group(2)) {
+ val jarName = jar.getName
+ if (jarName.endsWith(".jar")) {
+ if (jarName.startsWith(FLINK_SHIMS_PREFIX)) {
+ val prefixVer = s"$FLINK_SHIMS_PREFIX-${majorVersion}_$scalaVersion"
+ if (jarName.startsWith(prefixVer)) {
addShimUrl(jar)
+ logInfo(s"include flink shims jar lib: $jarName")
}
} else {
- if (INCLUDE_PATTERN.matcher(jar.getName).matches()) {
+ if (INCLUDE_PATTERN.matcher(jarName).matches()) {
addShimUrl(jar)
- logInfo(s"include jar lib: ${jar.getName}")
- }
- if (streamParkMatcher.matcher(jar.getName).matches()) {
+ logInfo(s"include jar lib: $jarName")
+ } else if (jarName.matches(s"^streampark-.*_$scalaVersion.*$$")) {
addShimUrl(jar)
- logInfo(s"include streampark lib: ${jar.getName}")
+ logInfo(s"include streampark lib: $jarName")
}
}
- } catch {
- case e: Exception => e.printStackTrace()
}
})
}
@@ -177,18 +153,17 @@ object FlinkShimsProxy extends Logger {
val shimsUrls = ListBuffer[URL](libURL: _*)
// 2) add all shims jar
- addShimsUrls(
- flinkVersion,
- file => {
- if (file != null) {
- shimsUrls += file.toURI.toURL
- }
- })
+ addShimsUrls(flinkVersion, file => {
+ if (file != null) {
+ shimsUrls += file.toURI.toURL
+ }
+ })
new ChildFirstClassLoader(
shimsUrls.toArray,
Thread.currentThread().getContextClassLoader,
- getFlinkShimsResourcePattern(flinkVersion.majorVersion))
+ getFlinkShimsResourcePattern(flinkVersion.majorVersion)
+ )
})
}
@@ -198,8 +173,7 @@ object FlinkShimsProxy extends Logger {
file.listFiles.filter(filterFun).map(_.toURI.toURL).toList
}
- @throws[Exception]
- def getObject[T](loader: ClassLoader, obj: Object): T = {
+ @throws[Exception] def getObject[T](loader: ClassLoader, obj: Object): T = {
val arrayOutputStream = new ByteArrayOutputStream
val result = Utils.tryWithResource(new ObjectOutputStream(arrayOutputStream))(objectOutputStream => {
objectOutputStream.writeObject(obj)