monrg commented on code in PR #1738:
URL:
https://github.com/apache/incubator-streampark/pull/1738#discussion_r991277153
##########
streampark-flink/streampark-flink-proxy/src/main/scala/org/apache/streampark/flink/proxy/FlinkShimsProxy.scala:
##########
@@ -81,59 +83,106 @@ object FlinkShimsProxy extends Logger {
})
}
- private[this] def getFlinkShimsClassLoader(flinkVersion: FlinkVersion):
ClassLoader = {
+ // flink 1.12 1.13~1.14 1.15 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")
+
VERIFY_SQL_CLASS_LOADER_CACHE.getOrElseUpdate(s"${flinkVersion.fullVersion}", {
+ val getFlinkTable: File => Boolean = _.getName.startsWith("flink-table")
+ // 1) flink/lib/flink-table*
+ val libTableURL = getFlinkHomeLib(flinkVersion.flinkHome, "lib",
getFlinkTable)
+
+ // 2) After version 1.15 need add flink/opt/flink-table*
+ val optTableURL = getFlinkHomeLib(flinkVersion.flinkHome, "opt",
getFlinkTable)
+ val shimsUrls = ListBuffer[URL](libTableURL ++ optTableURL: _*)
+
+ // 3) add only streampark shims jar
+ addShimsUrls(flinkVersion, shimsUrls,
_.getName.startsWith("streampark-flink-shims"))
+ new ChildFirstClassLoader(
+ shimsUrls.toArray,
+ Thread.currentThread().getContextClassLoader,
+ getFlinkShimsResourcePattern(flinkVersion.majorVersion)
+ )
+ })
+ }
+
+ def addShimsUrls(flinkVersion: FlinkVersion, shimsUrls: ListBuffer[URL],
filterJar: File => Boolean): Unit = {
+ val appHome = System.getProperty(ConfigConst.KEY_APP_HOME)
+ require(appHome != null, String.format("%s is not found on System env.",
ConfigConst.KEY_APP_HOME))
+
+ val libPath = new File(s"$appHome/lib")
+ require(libPath.exists())
val majorVersion = flinkVersion.majorVersion
val scalaVersion = flinkVersion.scalaVersion
- logInfo(flinkVersion.toString)
+ val streamParkMatcher = getStreamParkLibPattern(scalaVersion)
+
+ def addShimUrl(jar: File): Unit = {
+ if (filterJar.apply(jar)) {
+ shimsUrls += jar.toURI.toURL
+ }
+ }
+
+ libPath.listFiles().foreach((jar: File) => {
+ try {
+ val shimsMatcher = SHIMS_PATTERN.matcher(jar.getName)
+ if (shimsMatcher.matches()) {
+ if (majorVersion == shimsMatcher.group(1) && scalaVersion ==
shimsMatcher.group(2)) {
+ addShimUrl(jar)
+ }
+ } else {
+ if (INCLUDE_PATTERN.matcher(jar.getName).matches()) {
+ addShimUrl(jar)
+ logInfo(s"include jar lib: ${jar.getName}")
+ }
+ if (streamParkMatcher.matcher(jar.getName).matches()) {
+ addShimUrl(jar)
+ logInfo(s"include streampark lib: ${jar.getName}")
+ }
+ }
+ } catch {
+ case e: Exception => e.printStackTrace()
+ }
+ })
+ }
+
+ /**
+ * Get ClassLoader to verify sql
+ *
+ * @param flinkVersion flinkVersion
+ * @param func execute function
+ * @tparam T
+ * @return
+ */
+ def proxyVerifySql[T](flinkVersion: FlinkVersion, func:
JavaFunc[ClassLoader, T]): T = {
Review Comment:
hi, @wolfboys Thank you for your feedback，Because after flink 1.15
flink-table-planner migrated to opt directory, streampark classloader failed to
load the CalciteParser class, error when verify SQL script . nedd to add
flink-table-planner to verify sql classloader classpath . but ExecutorFactory
has two implementation: DefaultExecutorFactory(flink-table-planer) and
DelegateExecutorFactory(flink-table-planer-loader), spi load 2 servi will
conflict,
so i add a new method to separate the verify sql from the classloader of the
submitted job, compatible with multiple versions to get the check sql
dependency to create a new classloader
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
