Skip to content

Commit 940ca34

Browse files
shangeyaoshangeyao
authored
[Flink] Migrate streampark-flink-client from Scala to Java (#4467)
* [Migrate] #4452 Migrate streampark-flink-client from Scala to Java Convert all 27 Scala sources in client-api and client-core to Java, replace json4s with Jackson in session REST submit, and preserve FlinkShimsProxy reflection contracts for console compatibility. Closes #4452 * [CI] Fix Sonar issues for #4467 flink-client migration Address 31 Sonar findings: serializable request fields (S1948), refactor complex methods (S3776/S1141), replace instanceof checks (S1193), remove printStackTrace (S4507), and improve exception/boolean handling (S112/S5411). * [CI] Replace Sonar suppressions with proper fixes for #4467 Use parameter objects (JobClientTarget, SubmitApplicationSpec, etc.) for S107, make BuildResult Serializable for S1948, and standardize client API on FlinkException instead of @SuppressWarnings for S112. * [Flink] Fix remaining Sonar S112/S1141 in flink-client entrypoint and Yarn trait Replace throws Exception with FlinkException in FlinkClientEntrypoint and refactor YarnClientTrait to eliminate nested try blocks without suppressions. * [Flink] Reduce duplicated code in flink-client migration for Sonar Extract callAsFlinkException helpers, AbstractSavepointClientRequest base class, and generic entrypoint routing to lower new-code duplication. * [Flink] Further reduce Sonar duplicated lines in flink-client migration Extract shared logging, job-graph submit, and packer build response base classes to bring new-code duplication below the quality gate threshold. * [Flink] Remove unused import in SimpleBuildResponse * [Flink] Fix Sonar duplication in client-api bean classes Extract shared deploy request base, hdfs workspace helper, and compose SubmitRequest from SubmitApplicationSpec to eliminate 221 duplicated lines. * [Flink] Fix Spotless formatting in flink-client-api bean classes Align constructor parameter indentation with the project formatter so CI spotless:check passes. * [Flink] Fix Spotless formatting in flink-client-core Java classes Apply project formatter to client-core implementations and traits so CI spotless:check passes during backend build.
1 parent 5a3743e commit 940ca34

67 files changed

Lines changed: 4989 additions & 3282 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 28 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,10 @@
7575
import org.apache.streampark.flink.client.FlinkClient;
7676
import org.apache.streampark.flink.client.bean.CancelRequest;
7777
import org.apache.streampark.flink.client.bean.CancelResponse;
78+
import org.apache.streampark.flink.client.bean.JobClientTarget;
79+
import org.apache.streampark.flink.client.bean.SavepointCancelOptions;
80+
import org.apache.streampark.flink.client.bean.SubmitApplicationSpec;
81+
import org.apache.streampark.flink.client.bean.SubmitClusterSpec;
7882
import org.apache.streampark.flink.client.bean.SubmitRequest;
7983
import org.apache.streampark.flink.client.bean.SubmitResponse;
8084
import org.apache.streampark.flink.kubernetes.FlinkK8sWatcher;
@@ -309,13 +313,12 @@ public void cancel(FlinkApplication appParam) throws Exception {
309313
flinkEnv.getFlinkVersion(),
310314
FlinkDeployMode.of(application.getDeployMode()),
311315
properties,
312-
clusterId,
313-
application.getJobId(),
314-
appParam.getRestoreOrTriggerSavepoint(),
315-
appParam.getDrain(),
316-
customSavepoint,
317-
appParam.getNativeFormat(),
318-
namespace);
316+
new JobClientTarget(clusterId, application.getJobId(), namespace),
317+
new SavepointCancelOptions(
318+
appParam.getRestoreOrTriggerSavepoint(),
319+
appParam.getDrain(),
320+
customSavepoint,
321+
appParam.getNativeFormat()));
319322

320323
final Date triggerTime = new Date();
321324
CompletableFuture<CancelResponse> cancelFuture =
@@ -459,22 +462,25 @@ public void start(FlinkApplication appParam, boolean auto) throws Exception {
459462
flinkEnv.getFlinkVersion(),
460463
FlinkDeployMode.of(application.getDeployMode()),
461464
getProperties(application, dynamicProperties),
462-
flinkEnv.getFlinkConf(),
463-
FlinkJobType.of(application.getJobType()),
464-
application.getId(),
465-
new JobID().toHexString(),
466-
application.getJobName(),
467-
appConf,
468-
application.getApplicationType(),
469-
getSavepointPath(appParam),
470-
FlinkRestoreMode.of(appParam.getRestoreMode()),
471-
applicationArgs,
472-
k8sClusterId,
473-
application.getHadoopUser(),
465+
SubmitApplicationSpec.builder()
466+
.flinkYaml(flinkEnv.getFlinkConf())
467+
.jobType(FlinkJobType.of(application.getJobType()))
468+
.id(application.getId())
469+
.jobId(new JobID().toHexString())
470+
.appName(application.getJobName())
471+
.appConf(appConf)
472+
.applicationType(application.getApplicationType())
473+
.savePoint(getSavepointPath(appParam))
474+
.restoreMode(FlinkRestoreMode.of(appParam.getRestoreMode()))
475+
.args(applicationArgs)
476+
.build(),
477+
new SubmitClusterSpec(
478+
k8sClusterId,
479+
application.getHadoopUser(),
480+
k8sNamespace,
481+
exposedType),
474482
buildResult,
475-
extraParameter,
476-
k8sNamespace,
477-
exposedType);
483+
extraParameter);
478484

479485
CompletableFuture<SubmitResponse> future =
480486
CompletableFuture.supplyAsync(() -> FlinkClient.submit(submitRequest), executorService);

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

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,9 @@
4646
import org.apache.streampark.console.core.util.ServiceHelper;
4747
import org.apache.streampark.console.core.watcher.FlinkAppHttpWatcher;
4848
import org.apache.streampark.flink.client.FlinkClient;
49+
import org.apache.streampark.flink.client.bean.JobClientTarget;
4950
import org.apache.streampark.flink.client.bean.SavepointResponse;
51+
import org.apache.streampark.flink.client.bean.SavepointTriggerOptions;
5052
import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
5153
import org.apache.streampark.flink.util.FlinkUtils;
5254

@@ -503,10 +505,7 @@ private TriggerSavepointRequest renderTriggerSavepointRequest(
503505
flinkEnv.getFlinkVersion(),
504506
application.getDeployModeEnum(),
505507
properties,
506-
clusterId,
507-
application.getJobId(),
508-
customSavepoint,
509-
nativeFormat,
510-
application.getK8sNamespace());
508+
new JobClientTarget(clusterId, application.getJobId(), application.getK8sNamespace()),
509+
new SavepointTriggerOptions(customSavepoint, nativeFormat));
511510
}
512511
}

‎streampark-flink/streampark-flink-client/streampark-flink-client-api/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,12 @@
7272
<artifactId>hadoop-client-runtime</artifactId>
7373
<scope>provided</scope>
7474
</dependency>
75+
76+
<dependency>
77+
<groupId>org.projectlombok</groupId>
78+
<artifactId>lombok</artifactId>
79+
<scope>provided</scope>
80+
</dependency>
7581
</dependencies>
7682

7783
<profiles>

streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequestTrait.scala renamed to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/java/org/apache/streampark/flink/client/ExitSecurityManager.java

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -15,20 +15,23 @@
1515
* limitations under the License.
1616
*/
1717

18-
package org.apache.streampark.flink.client.bean
18+
package org.apache.streampark.flink.client;
1919

20-
import org.apache.streampark.common.conf.FlinkVersion
21-
import org.apache.streampark.common.enums.FlinkDeployMode
22-
import org.apache.streampark.common.util.Implicits.JavaMap
20+
import java.security.Permission;
2321

24-
import javax.annotation.Nullable
22+
/** Used to mask JVM requests for external operations. */
23+
public class ExitSecurityManager extends SecurityManager {
2524

26-
trait DeployRequestTrait {
25+
@Override
26+
public void checkExit(int status) {
27+
throw new SecurityException(
28+
"System.exit("
29+
+ status
30+
+ ") was called in your flink job, The job has been stopped, please check your program...");
31+
}
2732

28-
val flinkVersion: FlinkVersion
29-
val deployMode: FlinkDeployMode
30-
val properties: JavaMap[String, Any]
31-
val clusterId: String
32-
val id: Long
33-
@Nullable val k8sParam: KubernetesDeployParam
33+
@Override
34+
public void checkPermission(Permission perm) {
35+
// no-op
36+
}
3437
}

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

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
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.client;
19+
20+
import org.apache.streampark.common.conf.FlinkVersion;
21+
import org.apache.streampark.common.util.LoggerSupport;
22+
import org.apache.streampark.flink.client.bean.CancelRequest;
23+
import org.apache.streampark.flink.client.bean.CancelResponse;
24+
import org.apache.streampark.flink.client.bean.DeployRequest;
25+
import org.apache.streampark.flink.client.bean.DeployResponse;
26+
import org.apache.streampark.flink.client.bean.SavepointResponse;
27+
import org.apache.streampark.flink.client.bean.ShutDownRequest;
28+
import org.apache.streampark.flink.client.bean.ShutDownResponse;
29+
import org.apache.streampark.flink.client.bean.SubmitRequest;
30+
import org.apache.streampark.flink.client.bean.SubmitResponse;
31+
import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
32+
import org.apache.streampark.flink.proxy.FlinkShimsProxy;
33+
34+
import java.util.function.Function;
35+
36+
public final class FlinkClient extends LoggerSupport {
37+
38+
private static final String FLINK_CLIENT_ENTRYPOINT_CLASS =
39+
"org.apache.streampark.flink.client.FlinkClientEntrypoint";
40+
41+
private static final String SUBMIT_REQUEST =
42+
"org.apache.streampark.flink.client.bean.SubmitRequest";
43+
44+
private static final String DEPLOY_REQUEST =
45+
"org.apache.streampark.flink.client.bean.DeployRequest";
46+
47+
private static final String CANCEL_REQUEST =
48+
"org.apache.streampark.flink.client.bean.CancelRequest";
49+
50+
private static final String SHUTDOWN_REQUEST =
51+
"org.apache.streampark.flink.client.bean.ShutDownRequest";
52+
53+
private static final String SAVEPOINT_REQUEST =
54+
"org.apache.streampark.flink.client.bean.TriggerSavepointRequest";
55+
56+
private FlinkClient() {
57+
}
58+
59+
public static SubmitResponse submit(SubmitRequest submitRequest) {
60+
SecurityManager securityManager = System.getSecurityManager();
61+
try {
62+
System.setSecurityManager(new ExitSecurityManager());
63+
return proxy(submitRequest, submitRequest.flinkVersion(), SUBMIT_REQUEST, "submit");
64+
} finally {
65+
System.setSecurityManager(securityManager);
66+
}
67+
}
68+
69+
public static CancelResponse cancel(CancelRequest stopRequest) {
70+
return proxy(stopRequest, stopRequest.flinkVersion(), CANCEL_REQUEST, "cancel");
71+
}
72+
73+
public static DeployResponse deploy(DeployRequest deployRequest) {
74+
return proxy(deployRequest, deployRequest.flinkVersion(), DEPLOY_REQUEST, "deploy");
75+
}
76+
77+
public static ShutDownResponse shutdown(ShutDownRequest shutDownRequest) {
78+
return proxy(shutDownRequest, shutDownRequest.flinkVersion(), SHUTDOWN_REQUEST, "shutdown");
79+
}
80+
81+
public static SavepointResponse triggerSavepoint(TriggerSavepointRequest savepointRequest) {
82+
return proxy(
83+
savepointRequest, savepointRequest.flinkVersion(), SAVEPOINT_REQUEST, "triggerSavepoint");
84+
}
85+
86+
@SuppressWarnings("unchecked")
87+
private static <T> T proxy(
88+
Object request,
89+
FlinkVersion flinkVersion,
90+
String requestClassName,
91+
String methodName) {
92+
flinkVersion.checkVersion();
93+
return FlinkShimsProxy.proxy(
94+
flinkVersion,
95+
(Function<ClassLoader, T>) classLoader -> {
96+
try {
97+
Class<?> submitClass = classLoader.loadClass(FLINK_CLIENT_ENTRYPOINT_CLASS);
98+
Class<?> requestClass = classLoader.loadClass(requestClassName);
99+
java.lang.reflect.Method method =
100+
submitClass.getDeclaredMethod(methodName, requestClass);
101+
method.setAccessible(true);
102+
Object obj =
103+
method.invoke(
104+
null, FlinkShimsProxy.getObject(classLoader, request));
105+
if (obj == null) {
106+
return null;
107+
}
108+
return FlinkShimsProxy.getObject(FlinkClient.class.getClassLoader(), obj);
109+
} catch (RuntimeException e) {
110+
throw e;
111+
} catch (Exception e) {
112+
throw new RuntimeException(e);
113+
}
114+
});
115+
}
116+
}

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

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
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.client.bean;
19+
20+
import org.apache.streampark.common.conf.FlinkVersion;
21+
import org.apache.streampark.common.enums.FlinkDeployMode;
22+
23+
import javax.annotation.Nullable;
24+
25+
import java.io.Serializable;
26+
import java.util.Map;
27+
28+
/** Shared fields for deploy and shutdown client requests. */
29+
abstract class AbstractDeployClientRequest implements DeployRequestTrait, Serializable {
30+
31+
private static final long serialVersionUID = 1L;
32+
33+
private final FlinkVersion flinkVersion;
34+
private final FlinkDeployMode deployMode;
35+
@Nullable
36+
private final Map<String, Serializable> properties;
37+
private final String clusterId;
38+
private final long id;
39+
@Nullable
40+
private final KubernetesDeployParam k8sParam;
41+
42+
AbstractDeployClientRequest(
43+
FlinkVersion flinkVersion,
44+
FlinkDeployMode deployMode,
45+
@Nullable Map<String, Object> properties,
46+
String clusterId,
47+
long id,
48+
@Nullable KubernetesDeployParam k8sParam) {
49+
this.flinkVersion = flinkVersion;
50+
this.deployMode = deployMode;
51+
this.properties = ClientBeanUtils.toSerializableMap(properties);
52+
this.clusterId = clusterId;
53+
this.id = id;
54+
this.k8sParam = k8sParam;
55+
}
56+
57+
@Override
58+
public FlinkVersion flinkVersion() {
59+
return flinkVersion;
60+
}
61+
62+
@Override
63+
public FlinkDeployMode deployMode() {
64+
return deployMode;
65+
}
66+
67+
@Override
68+
@Nullable
69+
public Map<String, Object> properties() {
70+
return ClientBeanUtils.copyPropertiesMap(properties);
71+
}
72+
73+
@Override
74+
public String clusterId() {
75+
return clusterId;
76+
}
77+
78+
@Override
79+
public long id() {
80+
return id;
81+
}
82+
83+
@Override
84+
@Nullable
85+
public KubernetesDeployParam k8sParam() {
86+
return k8sParam;
87+
}
88+
}

0 commit comments

Comments
 (0)