Skip to content

Commit 60768b8

Browse files
shangeyaocursoragent
shangeyao
and
cursoragent
authored
[Migrate] Migrate streampark-flink-proxy from Scala to Java (#4464)
* [Migrate] #4450 Migrate streampark-flink-proxy from Scala to Java Convert FlinkShimsProxy to Java preserving classloader cache keys and ChildFirstClassLoader behavior; add JUnit tests for getObject round-trip. Co-authored-by: Cursor <cursoragent@cursor.com> * [CI] Fix Sonar reliability issues for #4464 Narrow FlinkShimsProxy.getObject throws clause, extract matchShimIncludeReason to reduce cognitive complexity, and add explicit JUnit test dependency. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent 8d30256 commit 60768b8

4 files changed

Lines changed: 343 additions & 229 deletions

File tree

‎streampark-flink/streampark-flink-proxy/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,12 @@
3535
<scope>provided</scope>
3636
</dependency>
3737

38+
<dependency>
39+
<groupId>org.junit.jupiter</groupId>
40+
<artifactId>junit-jupiter-engine</artifactId>
41+
<scope>test</scope>
42+
</dependency>
43+
3844
</dependencies>
3945

4046
<profiles>

‎streampark-flink/streampark-flink-proxy/src/main/java/org/apache/streampark/flink/proxy/FlinkShimsProxy.java‎

Lines changed: 266 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,266 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.proxy;
19+
20+
import org.apache.streampark.common.conf.ConfigKeys;
21+
import org.apache.streampark.common.conf.FlinkVersion;
22+
import org.apache.streampark.common.constants.Constants;
23+
import org.apache.streampark.common.util.ChildFirstClassLoader;
24+
import org.apache.streampark.common.util.ClassLoaderObjectInputStream;
25+
import org.apache.streampark.common.util.ClassLoaderUtils;
26+
import org.apache.streampark.common.util.LoggerSupport;
27+
28+
import java.io.ByteArrayInputStream;
29+
import java.io.ByteArrayOutputStream;
30+
import java.io.File;
31+
import java.io.IOException;
32+
import java.io.ObjectOutputStream;
33+
import java.net.URL;
34+
import java.util.ArrayList;
35+
import java.util.Arrays;
36+
import java.util.Collections;
37+
import java.util.List;
38+
import java.util.Map;
39+
import java.util.concurrent.ConcurrentHashMap;
40+
import java.util.function.Consumer;
41+
import java.util.function.Function;
42+
import java.util.function.Predicate;
43+
import java.util.regex.Pattern;
44+
45+
/** Proxy for loading Flink version-specific shims behind an isolated classloader. */
46+
public final class FlinkShimsProxy extends LoggerSupport {
47+
48+
private static final FlinkShimsProxy LOG = new FlinkShimsProxy();
49+
50+
private static final Map<String, ClassLoader> SHIMS_CLASS_LOADER_CACHE = new ConcurrentHashMap<>();
51+
52+
private static final Map<String, ClassLoader> VERIFY_SQL_CLASS_LOADER_CACHE = new ConcurrentHashMap<>();
53+
54+
private static final Pattern FLINK_JAR_PATTERN =
55+
Pattern.compile("flink-(.*).jar", Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
56+
57+
private static final Pattern INCLUDE_PATTERN =
58+
Pattern.compile("(streampark-shaded-jackson-)(.*).jar", Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
59+
60+
private static final String FLINK_SHIMS_PREFIX = "streampark-flink-shims_flink";
61+
62+
private static final List<String> PARENT_FIRST_PATTERNS = Collections.unmodifiableList(
63+
Arrays.asList(
64+
"java.",
65+
"javax.xml",
66+
"org.slf4j",
67+
"org.apache.log4j",
68+
"org.apache.logging",
69+
"org.apache.commons.logging",
70+
"org.apache.commons.cli",
71+
"ch.qos.logback",
72+
"org.xml",
73+
"org.w3c",
74+
"org.apache.hadoop"));
75+
76+
private FlinkShimsProxy() {
77+
}
78+
79+
private static Pattern getFlinkShimsResourcePattern(String majorVersion) {
80+
return Pattern.compile(
81+
"flink-(.*)-" + majorVersion + "(.*).jar",
82+
Pattern.CASE_INSENSITIVE | Pattern.DOTALL);
83+
}
84+
85+
/**
86+
* Get shimsClassLoader to execute for java/scala API (SAM {@link Function}).
87+
*
88+
* @param flinkVersion flinkVersion
89+
* @param func execute function
90+
* @param <T> return type
91+
* @return result of func
92+
*/
93+
public static <T> T proxy(FlinkVersion flinkVersion, Function<ClassLoader, T> func) {
94+
ClassLoader shimsClassLoader = getFlinkShimsClassLoader(flinkVersion);
95+
return ClassLoaderUtils.runAsClassLoader(shimsClassLoader, () -> func.apply(shimsClassLoader));
96+
}
97+
98+
/**
99+
* Get ClassLoader to verify sql.
100+
*
101+
* @param flinkVersion flinkVersion
102+
* @param func execute function
103+
* @param <T> return type
104+
* @return result of func
105+
*/
106+
public static <T> T proxyVerifySql(FlinkVersion flinkVersion, Function<ClassLoader, T> func) {
107+
ClassLoader shimsClassLoader = getVerifySqlLibClassLoader(flinkVersion);
108+
return ClassLoaderUtils.runAsClassLoader(shimsClassLoader, () -> func.apply(shimsClassLoader));
109+
}
110+
111+
@SuppressWarnings("unchecked")
112+
public static <T> T getObject(ClassLoader loader, Object obj) throws IOException, ClassNotFoundException {
113+
try (
114+
ByteArrayOutputStream arrayOutputStream = new ByteArrayOutputStream();
115+
ObjectOutputStream out = new ObjectOutputStream(arrayOutputStream)) {
116+
out.writeObject(obj);
117+
try (
118+
ByteArrayInputStream byteArrayInputStream =
119+
new ByteArrayInputStream(arrayOutputStream.toByteArray());
120+
ClassLoaderObjectInputStream in =
121+
new ClassLoaderObjectInputStream(loader, byteArrayInputStream)) {
122+
return (T) in.readObject();
123+
}
124+
}
125+
}
126+
127+
// need to load all flink-table dependencies compatible with different versions
128+
private static ClassLoader getVerifySqlLibClassLoader(FlinkVersion flinkVersion) {
129+
LOG.logInfo("Add verify sql lib,flink version: " + flinkVersion);
130+
return VERIFY_SQL_CLASS_LOADER_CACHE.computeIfAbsent(
131+
flinkVersion.fullVersion(),
132+
key -> {
133+
Predicate<File> getFlinkTable = f -> f.getName().startsWith("flink-table");
134+
List<URL> libTableURL = getFlinkHomeLib(flinkVersion.flinkHome, "lib", getFlinkTable);
135+
List<URL> optTableURL = getFlinkHomeLib(flinkVersion.flinkHome, "opt", getFlinkTable);
136+
List<URL> shimsUrls = new ArrayList<>(libTableURL);
137+
shimsUrls.addAll(optTableURL);
138+
139+
addShimsUrls(
140+
flinkVersion,
141+
file -> {
142+
if (file.getName().startsWith("streampark-flink-shims")) {
143+
try {
144+
shimsUrls.add(file.toURI().toURL());
145+
} catch (Exception e) {
146+
throw new RuntimeException(e);
147+
}
148+
}
149+
});
150+
151+
return new ChildFirstClassLoader(
152+
shimsUrls.toArray(new URL[0]),
153+
Thread.currentThread().getContextClassLoader(),
154+
PARENT_FIRST_PATTERNS,
155+
jarName -> loadJarFilter(jarName, flinkVersion));
156+
});
157+
}
158+
159+
private static boolean loadJarFilter(String jarName, FlinkVersion flinkVersion) {
160+
Pattern childFirstPattern = getFlinkShimsResourcePattern(flinkVersion.majorVersion());
161+
return FLINK_JAR_PATTERN.matcher(jarName).matches()
162+
&& !childFirstPattern.matcher(jarName).matches();
163+
}
164+
165+
private static void addShimsUrls(FlinkVersion flinkVersion, Consumer<File> addShimUrl) {
166+
String appHome = System.getProperty(ConfigKeys.KEY_APP_HOME());
167+
if (appHome == null) {
168+
throw new IllegalArgumentException(
169+
String.format("%s is not found on System env.", ConfigKeys.KEY_APP_HOME()));
170+
}
171+
172+
File libPath = new File(appHome + "/lib");
173+
if (!libPath.exists()) {
174+
throw new IllegalArgumentException("lib path does not exist: " + libPath);
175+
}
176+
177+
String majorVersion = flinkVersion.majorVersion();
178+
String scalaVersion = flinkVersion.scalaVersion();
179+
File[] jars = libPath.listFiles();
180+
if (jars == null) {
181+
return;
182+
}
183+
184+
for (File jar : jars) {
185+
String jarName = jar.getName();
186+
if (!jarName.endsWith(Constants.JAR_SUFFIX)) {
187+
continue;
188+
}
189+
String includeReason = matchShimIncludeReason(jarName, majorVersion, scalaVersion);
190+
if (includeReason != null) {
191+
addShimUrl.accept(jar);
192+
LOG.logInfo(includeReason + jarName);
193+
}
194+
}
195+
}
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+
212+
private static ClassLoader getFlinkShimsClassLoader(FlinkVersion flinkVersion) {
213+
LOG.logInfo("add flink shims urls classloader,flink version: " + flinkVersion);
214+
return SHIMS_CLASS_LOADER_CACHE.computeIfAbsent(
215+
flinkVersion.fullVersion(),
216+
key -> {
217+
Predicate<File> filter =
218+
file -> !file.getName().startsWith("log4j") && file.getName().endsWith(".jar");
219+
List<URL> libURL = getFlinkHomeLib(flinkVersion.flinkHome, "lib", filter);
220+
List<URL> shimsUrls = new ArrayList<>(libURL);
221+
222+
addShimsUrls(
223+
flinkVersion,
224+
file -> {
225+
if (file != null) {
226+
try {
227+
shimsUrls.add(file.toURI().toURL());
228+
} catch (Exception e) {
229+
throw new RuntimeException(e);
230+
}
231+
}
232+
});
233+
234+
return new ChildFirstClassLoader(
235+
shimsUrls.toArray(new URL[0]),
236+
Thread.currentThread().getContextClassLoader(),
237+
PARENT_FIRST_PATTERNS,
238+
jarName -> loadJarFilter(jarName, flinkVersion));
239+
});
240+
}
241+
242+
private static List<URL> getFlinkHomeLib(
243+
String flinkHome,
244+
String childDir,
245+
Predicate<File> filterFun) {
246+
File file = new File(flinkHome, childDir);
247+
if (!file.isDirectory()) {
248+
throw new IllegalArgumentException("FLINK_HOME " + file + " does not exist");
249+
}
250+
File[] files = file.listFiles();
251+
if (files == null) {
252+
return Collections.emptyList();
253+
}
254+
List<URL> urls = new ArrayList<>();
255+
for (File f : files) {
256+
if (filterFun.test(f)) {
257+
try {
258+
urls.add(f.toURI().toURL());
259+
} catch (Exception e) {
260+
throw new RuntimeException(e);
261+
}
262+
}
263+
}
264+
return urls;
265+
}
266+
}

0 commit comments

Comments
 (0)