Skip to content

Commit 9a0921c

Browse files
shangeyaocursoragent
shangeyao
and
cursoragent
committed
[Common][Flink] Sonar-oriented fixes and expand migration regression tests
Replace printStackTrace and generic RuntimeException wraps with specific exceptions, null-safe SubmitRequest accessors, FlinkShimsProxy URL helper, and TrackId validation cleanup. Add ParameterCliTest, TrackIdTest, and more SqlClient/SubmitRequest coverage. Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent 4da35ea commit 9a0921c

12 files changed

Lines changed: 240 additions & 53 deletions

File tree

‎streampark-common/src/main/java/org/apache/streampark/common/conf/FlinkVersion.java‎

Lines changed: 14 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,11 @@
2323

2424
import java.io.File;
2525
import java.io.Serializable;
26+
import java.net.MalformedURLException;
2627
import java.net.URL;
2728
import java.util.Arrays;
2829
import java.util.Collections;
2930
import java.util.List;
30-
import java.util.function.Consumer;
3131
import java.util.regex.Matcher;
3232
import java.util.regex.Pattern;
3333
import java.util.stream.Collectors;
@@ -122,8 +122,8 @@ public List<URL> getFlinkLibs() throws Exception {
122122
return Arrays.stream(files).map(f -> {
123123
try {
124124
return f.toURI().toURL();
125-
} catch (Exception e) {
126-
throw new RuntimeException(e);
125+
} catch (MalformedURLException e) {
126+
throw new IllegalArgumentException("Invalid Flink lib URL: " + f, e);
127127
}
128128
}).collect(Collectors.toList());
129129
}
@@ -141,22 +141,18 @@ public String getVersion() {
141141
CommandUtils.execute(
142142
getFlinkLib().getAbsolutePath(),
143143
cmd,
144-
new Consumer<String>() {
145-
146-
@Override
147-
public void accept(String out) {
148-
buffer.append(out).append("\n");
149-
Matcher matcher = FLINK_VERSION_PATTERN.matcher(out);
150-
if (matcher.find()) {
151-
String ver = matcher.group(1);
152-
Matcher m1 = APACHE_FLINK_VERSION_PATTERN.matcher(ver);
153-
if (m1.find()) {
144+
out -> {
145+
buffer.append(out).append("\n");
146+
Matcher matcher = FLINK_VERSION_PATTERN.matcher(out);
147+
if (matcher.find()) {
148+
String ver = matcher.group(1);
149+
Matcher m1 = APACHE_FLINK_VERSION_PATTERN.matcher(ver);
150+
if (m1.find()) {
151+
flinkVersion[0] = ver;
152+
} else {
153+
Matcher m2 = OTHER_FLINK_VERSION_PATTERN.matcher(ver);
154+
if (m2.find()) {
154155
flinkVersion[0] = ver;
155-
} else {
156-
Matcher m2 = OTHER_FLINK_VERSION_PATTERN.matcher(ver);
157-
if (m2.find()) {
158-
flinkVersion[0] = ver;
159-
}
160156
}
161157
}
162158
}

‎streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/FlinkClient.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
3131
import org.apache.streampark.flink.proxy.FlinkShimsProxy;
3232

33+
import java.io.IOException;
3334
import java.util.function.Function;
3435

3536
public final class FlinkClient {
@@ -107,8 +108,9 @@ private static <T> T proxy(
107108
return FlinkShimsProxy.getObject(FlinkClient.class.getClassLoader(), obj);
108109
} catch (RuntimeException e) {
109110
throw e;
110-
} catch (Exception e) {
111-
throw new RuntimeException(e);
111+
} catch (ReflectiveOperationException | IOException e) {
112+
throw new IllegalStateException(
113+
"Failed to invoke Flink client via shims proxy: " + methodName, e);
112114
}
113115
});
114116
}

‎streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/ClientBeanUtils.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,9 @@
2626
import javax.annotation.Nullable;
2727

2828
import java.io.File;
29+
import java.io.IOException;
2930
import java.io.Serializable;
31+
import java.io.UncheckedIOException;
3032
import java.util.HashMap;
3133
import java.util.Map;
3234

@@ -45,8 +47,9 @@ static HdfsWorkspace createHdfsWorkspace(FlinkVersion flinkVersion) {
4547
FileUtils.isSymlink(flinkHomeDir)
4648
? flinkHomeDir.getCanonicalFile().getName()
4749
: flinkHomeDir.getName();
48-
} catch (Exception e) {
49-
throw new RuntimeException(e);
50+
} catch (IOException e) {
51+
throw new UncheckedIOException(
52+
"Failed to resolve Flink home directory name: " + flinkHome, e);
5053
}
5154
String flinkHdfsHome = workspace.APP_FLINK() + "/" + flinkName;
5255
return new HdfsWorkspace(

‎streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/bean/SubmitRequest.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444

4545
import java.io.File;
4646
import java.io.Serializable;
47+
import java.net.MalformedURLException;
4748
import java.net.URL;
4849
import java.util.ArrayList;
4950
import java.util.Collections;
@@ -225,8 +226,8 @@ public List<URL> libs() {
225226
for (File file : files) {
226227
try {
227228
urls.add(file.toURI().toURL());
228-
} catch (Exception e) {
229-
throw new RuntimeException(e);
229+
} catch (MalformedURLException e) {
230+
throw new IllegalArgumentException("Invalid lib file URL: " + file, e);
230231
}
231232
}
232233
libs = urls;
@@ -250,7 +251,11 @@ public List<URL> classPaths() {
250251

251252
public String flinkSQL() {
252253
if (flinkSQL == null) {
253-
flinkSQL = extraParameter.get(ConfigKeys.KEY_FLINK_SQL()).toString();
254+
if (extraParameter == null) {
255+
return null;
256+
}
257+
Object sql = extraParameter.get(ConfigKeys.KEY_FLINK_SQL());
258+
flinkSQL = sql != null ? sql.toString() : null;
254259
}
255260
return flinkSQL;
256261
}
@@ -336,7 +341,7 @@ public boolean hasExtra(String key) {
336341
}
337342

338343
public Object getExtra(String key) {
339-
return extraParameter.get(key);
344+
return extraParameter != null ? extraParameter.get(key) : null;
340345
}
341346

342347
public HdfsWorkspace hdfsWorkspace() {

‎streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/java/org/apache/streampark/flink/client/test/SubmitRequestTest.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,41 @@ void propertiesMapCopiesSerializableIntegerValues() {
9999
assertThat(request.getProp("key")).isEqualTo(42);
100100
}
101101

102+
@Test
103+
void flinkSqlShouldReturnNullWhenExtraParameterMissing() {
104+
SubmitRequest request = createRequest(FlinkJobType.FLINK_SQL, null);
105+
assertThat(request.flinkSQL()).isNull();
106+
}
107+
108+
@Test
109+
void flinkSqlShouldReadFromExtraParameter() {
110+
Map<String, Object> extra = new HashMap<>();
111+
extra.put(ConfigKeys.KEY_FLINK_SQL(), "select 1");
112+
SubmitRequest request =
113+
new SubmitRequest(
114+
FLINK_VERSION,
115+
FlinkDeployMode.YARN_APPLICATION,
116+
Collections.emptyMap(),
117+
SubmitApplicationSpec.builder().jobType(FlinkJobType.FLINK_SQL).build(),
118+
null,
119+
null,
120+
extra);
121+
assertThat(request.flinkSQL()).isEqualTo("select 1");
122+
}
123+
124+
@Test
125+
void getExtraShouldBeNullSafe() {
126+
SubmitRequest request = createRequest(FlinkJobType.FLINK_JAR, null);
127+
assertThat(request.getExtra("missing")).isNull();
128+
assertThat(request.hasExtra("missing")).isFalse();
129+
}
130+
131+
@Test
132+
void allowNonRestoredStateShouldDefaultToFalse() {
133+
SubmitRequest request = createRequest(FlinkJobType.FLINK_JAR, null);
134+
assertThat(request.allowNonRestoredState()).isFalse();
135+
}
136+
102137
private static SubmitRequest createRequest(FlinkJobType jobType, String appConf) {
103138
SubmitApplicationSpec application =
104139
SubmitApplicationSpec.builder()

‎streampark-flink/streampark-flink-kubernetes/src/main/java/org/apache/streampark/flink/kubernetes/model/TrackId.java‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -128,11 +128,7 @@ public static TrackId onApplication(
128128
}
129129

130130
private static boolean isNotEmpty(String value) {
131-
try {
132-
return value != null && !value.isEmpty();
133-
} catch (Exception e) {
134-
return false;
135-
}
131+
return value != null && !value.isEmpty();
136132
}
137133

138134
@Override

‎streampark-flink/streampark-flink-kubernetes/src/test/java/org/apache/streampark/flink/kubernetes/model/TrackIdTest.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
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.kubernetes.model;
19+
20+
import org.junit.jupiter.api.Test;
21+
22+
import static org.assertj.core.api.Assertions.assertThat;
23+
24+
class TrackIdTest {
25+
26+
@Test
27+
void applicationTrackShouldBeLegalWithNamespaceAndClusterId() {
28+
TrackId trackId =
29+
TrackId.onApplication("ns1", "cluster-a", 1L, null, null, null);
30+
assertThat(trackId.isLegal()).isTrue();
31+
assertThat(trackId.isActive()).isFalse();
32+
}
33+
34+
@Test
35+
void sessionTrackShouldRequireJobIdToBeActive() {
36+
TrackId trackId = TrackId.onSession("ns1", "cluster-a", 1L, "job-1", null, null);
37+
assertThat(trackId.isLegal()).isTrue();
38+
assertThat(trackId.isActive()).isTrue();
39+
}
40+
41+
@Test
42+
void incompleteSessionTrackShouldBeIllegal() {
43+
TrackId trackId = TrackId.onSession("ns1", "cluster-a", 1L, null, null, null);
44+
assertThat(trackId.isLegal()).isFalse();
45+
}
46+
47+
@Test
48+
void equalsShouldCompareIdentityFields() {
49+
TrackId left = TrackId.onApplication("ns1", "cluster-a", 1L, "job-1", "g1", null);
50+
TrackId right = TrackId.onApplication("ns1", "cluster-a", 1L, "job-1", "g1", null);
51+
assertThat(left).isEqualTo(right);
52+
assertThat(left.hashCode()).isEqualTo(right.hashCode());
53+
}
54+
}

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

Lines changed: 12 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.io.File;
3131
import java.io.IOException;
3232
import java.io.ObjectOutputStream;
33+
import java.net.MalformedURLException;
3334
import java.net.URL;
3435
import java.util.ArrayList;
3536
import java.util.Arrays;
@@ -140,11 +141,7 @@ private static ClassLoader getVerifySqlLibClassLoader(FlinkVersion flinkVersion)
140141
flinkVersion,
141142
file -> {
142143
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-
}
144+
shimsUrls.add(toUrl(file));
148145
}
149146
});
150147

@@ -223,11 +220,7 @@ private static ClassLoader getFlinkShimsClassLoader(FlinkVersion flinkVersion) {
223220
flinkVersion,
224221
file -> {
225222
if (file != null) {
226-
try {
227-
shimsUrls.add(file.toURI().toURL());
228-
} catch (Exception e) {
229-
throw new RuntimeException(e);
230-
}
223+
shimsUrls.add(toUrl(file));
231224
}
232225
});
233226

@@ -254,13 +247,17 @@ private static List<URL> getFlinkHomeLib(
254247
List<URL> urls = new ArrayList<>();
255248
for (File f : files) {
256249
if (filterFun.test(f)) {
257-
try {
258-
urls.add(f.toURI().toURL());
259-
} catch (Exception e) {
260-
throw new RuntimeException(e);
261-
}
250+
urls.add(toUrl(f));
262251
}
263252
}
264253
return urls;
265254
}
255+
256+
private static URL toUrl(File file) {
257+
try {
258+
return file.toURI().toURL();
259+
} catch (MalformedURLException e) {
260+
throw new IllegalArgumentException("Invalid file URL: " + file, e);
261+
}
262+
}
266263
}

‎streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/java/org/apache/streampark/flink/core/conf/ParameterCli.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@
2727

2828
import java.net.URLClassLoader;
2929
import java.util.ArrayList;
30-
import java.util.Collections;
3130
import java.util.LinkedHashMap;
3231
import java.util.List;
3332
import java.util.Locale;
@@ -96,7 +95,8 @@ private static Map<String, String> loadConfig(String conf) {
9695
"[StreamPark] Usage:flink.conf file error,must be (yml|conf|properties)");
9796
}
9897
} catch (Exception e) {
99-
return Collections.emptyMap();
98+
throw new IllegalArgumentException(
99+
"[StreamPark] Failed to load flink config file: " + conf, e);
100100
}
101101
}
102102

@@ -112,7 +112,7 @@ private static String buildOption(Map<String, String> map, String[] programArgs)
112112
}
113113
}
114114
} catch (ParseException exception) {
115-
exception.printStackTrace();
115+
// Ignore unrecognized CLI tokens; valid options are still collected below.
116116
}
117117
String mainClass = map.get(OPTION_MAIN);
118118
if (mainClass != null) {
@@ -151,7 +151,7 @@ private static String buildDetachedMode(Map<String, String> map, String[] progra
151151
|| line.hasOption(FlinkRunOption.DETACHED_OPTION.getLongOpt());
152152
return detached ? "Detached" : "Attach";
153153
} catch (ParseException e) {
154-
throw new RuntimeException(e);
154+
throw new IllegalArgumentException("Failed to parse Flink detached mode options", e);
155155
}
156156
}
157157

@@ -187,7 +187,7 @@ public static String[] getOption(Map<String, String> map, String[] args) {
187187
}
188188
}
189189
} catch (ParseException e) {
190-
e.printStackTrace();
190+
// Ignore unrecognized CLI tokens merged from program arguments.
191191
}
192192
}
193193

0 commit comments

Comments
 (0)