monrg commented on code in PR #1738:
URL:
https://github.com/apache/incubator-streampark/pull/1738#discussion_r992067739
##########
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 = {
Review Comment:
hi @1996fanrui i refactor it !
--
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]
