|
28 | 28 | import java.io.ByteArrayInputStream; |
29 | 29 | import java.io.ByteArrayOutputStream; |
30 | 30 | import java.io.File; |
| 31 | +import java.io.IOException; |
31 | 32 | import java.io.ObjectOutputStream; |
32 | 33 | import java.net.URL; |
33 | 34 | import java.util.ArrayList; |
@@ -108,7 +109,7 @@ public static <T> T proxyVerifySql(FlinkVersion flinkVersion, Function<ClassLoad |
108 | 109 | } |
109 | 110 |
|
110 | 111 | @SuppressWarnings("unchecked") |
111 | | - public static <T> T getObject(ClassLoader loader, Object obj) throws Exception { |
| 112 | + public static <T> T getObject(ClassLoader loader, Object obj) throws IOException, ClassNotFoundException { |
112 | 113 | try ( |
113 | 114 | ByteArrayOutputStream arrayOutputStream = new ByteArrayOutputStream(); |
114 | 115 | ObjectOutputStream out = new ObjectOutputStream(arrayOutputStream)) { |
@@ -185,24 +186,29 @@ private static void addShimsUrls(FlinkVersion flinkVersion, Consumer<File> addSh |
185 | 186 | if (!jarName.endsWith(Constants.JAR_SUFFIX)) { |
186 | 187 | continue; |
187 | 188 | } |
188 | | - if (jarName.startsWith(FLINK_SHIMS_PREFIX)) { |
189 | | - String prefixVer = FLINK_SHIMS_PREFIX + "-" + majorVersion + "_" + scalaVersion; |
190 | | - if (jarName.startsWith(prefixVer)) { |
191 | | - addShimUrl.accept(jar); |
192 | | - LOG.logInfo("Include flink shims jar lib: " + jarName); |
193 | | - } |
194 | | - } else { |
195 | | - if (INCLUDE_PATTERN.matcher(jarName).matches()) { |
196 | | - addShimUrl.accept(jar); |
197 | | - LOG.logInfo("Include jar lib: " + jarName); |
198 | | - } else if (jarName.matches("^streampark-.*_" + scalaVersion + ".*$")) { |
199 | | - addShimUrl.accept(jar); |
200 | | - LOG.logInfo("Include streampark lib: " + jarName); |
201 | | - } |
| 189 | + String includeReason = matchShimIncludeReason(jarName, majorVersion, scalaVersion); |
| 190 | + if (includeReason != null) { |
| 191 | + addShimUrl.accept(jar); |
| 192 | + LOG.logInfo(includeReason + jarName); |
202 | 193 | } |
203 | 194 | } |
204 | 195 | } |
205 | 196 |
|
| 197 | + private static String matchShimIncludeReason( |
| 198 | + String jarName, String majorVersion, String scalaVersion) { |
| 199 | + if (jarName.startsWith(FLINK_SHIMS_PREFIX)) { |
| 200 | + String prefixVer = FLINK_SHIMS_PREFIX + "-" + majorVersion + "_" + scalaVersion; |
| 201 | + return jarName.startsWith(prefixVer) ? "Include flink shims jar lib: " : null; |
| 202 | + } |
| 203 | + if (INCLUDE_PATTERN.matcher(jarName).matches()) { |
| 204 | + return "Include jar lib: "; |
| 205 | + } |
| 206 | + if (jarName.matches("^streampark-.*_" + scalaVersion + ".*$")) { |
| 207 | + return "Include streampark lib: "; |
| 208 | + } |
| 209 | + return null; |
| 210 | + } |
| 211 | + |
206 | 212 | private static ClassLoader getFlinkShimsClassLoader(FlinkVersion flinkVersion) { |
207 | 213 | LOG.logInfo("add flink shims urls classloader,flink version: " + flinkVersion); |
208 | 214 | return SHIMS_CLASS_LOADER_CACHE.computeIfAbsent( |
|
0 commit comments