Skip to content

Commit 02d861c

Browse files
shangeyaocursoragent
shangeyao
and
cursoragent
committed
[Flink] Fix Sonar duplication and code smell findings for PR #4471
Extract shared Yarn jar upload and K8s Docker build helpers to reduce new-code duplication below the 3% quality gate, rename pipeline enum constants to UPPER_CASE, and address SqlSplitter/StringCastUtils Sonar issues. Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent e56b2e0 commit 02d861c

17 files changed

Lines changed: 484 additions & 532 deletions

File tree

‎streampark-common/src/main/java/org/apache/streampark/common/util/HadoopConfigUtils.java‎

Lines changed: 36 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -194,69 +194,47 @@ private static Map.Entry<String, String> findHostMatch(String line, Map<String,
194194

195195
public static Map<String, String> readSystemHadoopConf() {
196196
return getSystemHadoopConfDirOptional()
197-
.map(
198-
confDir -> {
199-
Map<String, String> map = new LinkedHashMap<>();
200-
File[] files = LfsOperator.listDir(confDir);
201-
if (files != null) {
202-
for (File f : files) {
203-
boolean matched = false;
204-
for (String name : HADOOP_CLIENT_CONF_FILES) {
205-
if (name.equals(f.getName())) {
206-
matched = true;
207-
break;
208-
}
209-
}
210-
if (!matched) {
211-
continue;
212-
}
213-
try {
214-
map.put(
215-
f.getName(),
216-
org.apache.commons.io.FileUtils.readFileToString(
217-
f, StandardCharsets.UTF_8));
218-
} catch (IOException e) {
219-
throw new IllegalStateException(
220-
"Failed to read Hadoop configuration file: " + f.getAbsolutePath(), e);
221-
}
222-
}
223-
}
224-
return map;
225-
})
197+
.map(dir -> readSystemConfFiles(dir, HADOOP_CLIENT_CONF_FILES, "Hadoop"))
226198
.orElse(Collections.emptyMap());
227199
}
228200

229201
public static Map<String, String> readSystemHiveConf() {
230202
return getSystemHiveConfDirOptional()
231-
.map(
232-
confDir -> {
233-
Map<String, String> map = new LinkedHashMap<>();
234-
File[] files = LfsOperator.listDir(confDir);
235-
if (files != null) {
236-
for (File f : files) {
237-
boolean matched = false;
238-
for (String name : HIVE_CLIENT_CONF_FILES) {
239-
if (name.equals(f.getName())) {
240-
matched = true;
241-
break;
242-
}
243-
}
244-
if (!matched) {
245-
continue;
246-
}
247-
try {
248-
map.put(
249-
f.getName(),
250-
org.apache.commons.io.FileUtils.readFileToString(
251-
f, StandardCharsets.UTF_8));
252-
} catch (IOException e) {
253-
throw new IllegalStateException(
254-
"Failed to read Hive configuration file: " + f.getAbsolutePath(), e);
255-
}
256-
}
257-
}
258-
return map;
259-
})
203+
.map(dir -> readSystemConfFiles(dir, HIVE_CLIENT_CONF_FILES, "Hive"))
260204
.orElse(Collections.emptyMap());
261205
}
206+
207+
private static Map<String, String> readSystemConfFiles(
208+
String confDir,
209+
String[] confFileNames,
210+
String confLabel) {
211+
Map<String, String> map = new LinkedHashMap<>();
212+
File[] files = LfsOperator.listDir(confDir);
213+
if (files == null) {
214+
return map;
215+
}
216+
for (File f : files) {
217+
if (!matchesConfFile(f.getName(), confFileNames)) {
218+
continue;
219+
}
220+
try {
221+
map.put(
222+
f.getName(),
223+
org.apache.commons.io.FileUtils.readFileToString(f, StandardCharsets.UTF_8));
224+
} catch (IOException e) {
225+
throw new IllegalStateException(
226+
"Failed to read " + confLabel + " configuration file: " + f.getAbsolutePath(), e);
227+
}
228+
}
229+
return map;
230+
}
231+
232+
private static boolean matchesConfFile(String fileName, String[] confFileNames) {
233+
for (String name : confFileNames) {
234+
if (name.equals(fileName)) {
235+
return true;
236+
}
237+
}
238+
return false;
239+
}
262240
}

‎streampark-common/src/main/java/org/apache/streampark/common/util/StringCastUtils.java‎

Lines changed: 28 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -24,26 +24,34 @@ private StringCastUtils() {
2424
}
2525

2626
public static <T> T cast(String v, Class<T> classType) {
27-
Object result;
27+
return classType.cast(parseValue(v, classType));
28+
}
29+
30+
private static Object parseValue(String v, Class<?> classType) {
2831
if (classType == String.class) {
29-
result = v;
30-
} else if (classType == Byte.class || classType == byte.class) {
31-
result = Byte.parseByte(v);
32-
} else if (classType == Integer.class || classType == int.class) {
33-
result = Integer.parseInt(v);
34-
} else if (classType == Long.class || classType == long.class) {
35-
result = Long.parseLong(v);
36-
} else if (classType == Float.class || classType == float.class) {
37-
result = Float.parseFloat(v);
38-
} else if (classType == Double.class || classType == double.class) {
39-
result = Double.parseDouble(v);
40-
} else if (classType == Short.class || classType == short.class) {
41-
result = Short.parseShort(v);
42-
} else if (classType == Boolean.class || classType == boolean.class) {
43-
result = Boolean.parseBoolean(v);
44-
} else {
45-
throw new IllegalArgumentException("Unsupported type: " + classType);
46-
}
47-
return classType.cast(result);
32+
return v;
33+
}
34+
if (classType == Byte.class || classType == byte.class) {
35+
return Byte.parseByte(v);
36+
}
37+
if (classType == Integer.class || classType == int.class) {
38+
return Integer.parseInt(v);
39+
}
40+
if (classType == Long.class || classType == long.class) {
41+
return Long.parseLong(v);
42+
}
43+
if (classType == Float.class || classType == float.class) {
44+
return Float.parseFloat(v);
45+
}
46+
if (classType == Double.class || classType == double.class) {
47+
return Double.parseDouble(v);
48+
}
49+
if (classType == Short.class || classType == short.class) {
50+
return Short.parseShort(v);
51+
}
52+
if (classType == Boolean.class || classType == boolean.class) {
53+
return Boolean.parseBoolean(v);
54+
}
55+
throw new IllegalArgumentException("Unsupported type: " + classType);
4856
}
4957
}

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBuildPipeline.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -304,7 +304,7 @@ public static View of(@Nonnull ApplicationBuildPipeline pipe) {
304304
Step step = new Step()
305305
.setSeq(i)
306306
.setDesc(stepDesc.getOrDefault(i, "unknown step"))
307-
.setStatus(stepStatus.getOrDefault(i, PipelineStepStatusEnum.unknown).getCode());
307+
.setStatus(stepStatus.getOrDefault(i, PipelineStepStatusEnum.UNKNOWN).getCode());
308308
Long st = stepTs.get(i);
309309
if (st != null) {
310310
step.setTs(new Date(st));

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationBuildPipelineServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -630,7 +630,7 @@ public DockerResolvedSnapshot getDockerProgressDetailSnapshot(@Nonnull Long appI
630630
@Override
631631
public boolean allowToBuildNow(@Nonnull Long appId) {
632632
return getCurrentBuildPipeline(appId)
633-
.map(pipeline -> PipelineStatusEnum.running != pipeline.getPipelineStatus())
633+
.map(pipeline -> PipelineStatusEnum.RUNNING != pipeline.getPipelineStatus())
634634
.orElse(true);
635635
}
636636

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -309,11 +309,11 @@ private AppControl getAppControl(FlinkApplication record) {
309309
return new AppControl()
310310
.setAllowBuild(
311311
record.getBuildStatus() == null
312-
|| !PipelineStatusEnum.running.getCode()
312+
|| !PipelineStatusEnum.RUNNING.getCode()
313313
.equals(record.getBuildStatus()))
314314
.setAllowStart(
315315
!record.shouldTracking()
316-
&& PipelineStatusEnum.success.getCode()
316+
&& PipelineStatusEnum.SUCCESS.getCode()
317317
.equals(record.getBuildStatus()))
318318
.setAllowStop(record.isRunning())
319319
.setAllowView(record.shouldTracking());

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/SparkApplicationBuildPipelineServiceImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -479,7 +479,7 @@ public Optional<ApplicationBuildPipeline> getCurrentBuildPipeline(@Nonnull Long
479479
@Override
480480
public boolean allowToBuildNow(@Nonnull Long appId) {
481481
return getCurrentBuildPipeline(appId)
482-
.map(pipeline -> PipelineStatusEnum.running != pipeline.getPipelineStatus())
482+
.map(pipeline -> PipelineStatusEnum.RUNNING != pipeline.getPipelineStatus())
483483
.orElse(true);
484484
}
485485

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/SparkApplicationManageServiceImpl.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -226,12 +226,12 @@ record -> {
226226
AppControl appControl = new AppControl()
227227
.setAllowBuild(
228228
record.getBuildStatus() == null
229-
|| !PipelineStatusEnum.running
229+
|| !PipelineStatusEnum.RUNNING
230230
.getCode()
231231
.equals(record.getBuildStatus()))
232232
.setAllowStart(
233233
!record.shouldTracking()
234-
&& PipelineStatusEnum.success
234+
&& PipelineStatusEnum.SUCCESS
235235
.getCode()
236236
.equals(record.getBuildStatus()))
237237
.setAllowStop(record.isRunning())

‎streampark-flink/streampark-flink-packer/src/main/java/org/apache/streampark/flink/packer/pipeline/BuildPipeline.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ public abstract class BuildPipeline extends LoggerSupport
4646
ThreadUtils.threadFactory("streampark-pipeline-watcher-executor"),
4747
new ThreadPoolExecutor.AbortPolicy());
4848

49-
protected PipelineStatusEnum pipeStatus = PipelineStatusEnum.pending;
49+
protected PipelineStatusEnum pipeStatus = PipelineStatusEnum.PENDING;
5050

5151
protected PipeError error = PipeError.empty();
5252

@@ -64,7 +64,7 @@ protected BuildPipeline() {
6464
(seq, desc) -> stepsStatus.put(
6565
seq,
6666
new AbstractMap.SimpleEntry<>(
67-
PipelineStepStatusEnum.waiting, System.currentTimeMillis())));
67+
PipelineStepStatusEnum.WAITING, System.currentTimeMillis())));
6868
}
6969

7070
/** use to identify the log record that belongs to which pipeline instance */
@@ -107,7 +107,7 @@ protected <R> java.util.Optional<R> execStep(int seq, Callable<R> process) {
107107
stepsStatus.put(
108108
seq,
109109
new AbstractMap.SimpleEntry<>(
110-
PipelineStepStatusEnum.running, System.currentTimeMillis()));
110+
PipelineStepStatusEnum.RUNNING, System.currentTimeMillis()));
111111
logInfo(
112112
"Building pipeline step["
113113
+ seq
@@ -120,16 +120,16 @@ protected <R> java.util.Optional<R> execStep(int seq, Callable<R> process) {
120120
stepsStatus.put(
121121
seq,
122122
new AbstractMap.SimpleEntry<>(
123-
PipelineStepStatusEnum.success, System.currentTimeMillis()));
123+
PipelineStepStatusEnum.SUCCESS, System.currentTimeMillis()));
124124
logInfo("Building pipeline step[" + seq + "/" + allSteps() + "] success");
125125
notifyStepChange();
126126
return java.util.Optional.of(result);
127127
} catch (Exception cause) {
128128
stepsStatus.put(
129129
seq,
130130
new AbstractMap.SimpleEntry<>(
131-
PipelineStepStatusEnum.failure, System.currentTimeMillis()));
132-
pipeStatus = PipelineStatusEnum.failure;
131+
PipelineStepStatusEnum.FAILURE, System.currentTimeMillis()));
132+
pipeStatus = PipelineStatusEnum.FAILURE;
133133
error = PipeError.of(cause.getMessage(), cause);
134134
logInfo(
135135
"Building pipeline step["
@@ -148,7 +148,7 @@ protected void skipStep(int step) {
148148
stepsStatus.put(
149149
step,
150150
new AbstractMap.SimpleEntry<>(
151-
PipelineStepStatusEnum.skipped, System.currentTimeMillis()));
151+
PipelineStepStatusEnum.SKIPPED, System.currentTimeMillis()));
152152
logInfo(
153153
"Building pipeline step["
154154
+ step
@@ -162,26 +162,26 @@ protected void skipStep(int step) {
162162
/** Launch the building pipeline. */
163163
@Override
164164
public BuildResult launch() {
165-
pipeStatus = PipelineStatusEnum.running;
165+
pipeStatus = PipelineStatusEnum.RUNNING;
166166
try {
167167
notifyStart();
168168
logInfo("Building pipeline is launching, params=" + offerBuildParam());
169169
BuildResult result =
170170
EXEC_POOL.submit(this::buildProcess).get(20, TimeUnit.MINUTES);
171-
pipeStatus = PipelineStatusEnum.success;
171+
pipeStatus = PipelineStatusEnum.SUCCESS;
172172
logInfo("Building pipeline has finished successfully.");
173173
notifyFinish(result);
174174
return result;
175175
} catch (InterruptedException e) {
176176
Thread.currentThread().interrupt();
177-
pipeStatus = PipelineStatusEnum.failure;
177+
pipeStatus = PipelineStatusEnum.FAILURE;
178178
error = PipeError.of(e.getMessage(), e);
179179
logError("Building pipeline has failed.", e);
180180
BuildResult result = new ErrorResult();
181181
notifyFinish(result);
182182
return result;
183183
} catch (ExecutionException e) {
184-
pipeStatus = PipelineStatusEnum.failure;
184+
pipeStatus = PipelineStatusEnum.FAILURE;
185185
Throwable cause = e.getCause();
186186
if (cause == null) {
187187
cause = e;
@@ -192,7 +192,7 @@ public BuildResult launch() {
192192
notifyFinish(result);
193193
return result;
194194
} catch (TimeoutException e) {
195-
pipeStatus = PipelineStatusEnum.failure;
195+
pipeStatus = PipelineStatusEnum.FAILURE;
196196
error = PipeError.of(e.getMessage(), e);
197197
logError("Building pipeline has failed.", e);
198198
BuildResult result = new ErrorResult();

0 commit comments

Comments
 (0)