[Flink] Migrate streampark-flink-client from Scala to Java

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

@@ -75,6 +75,10 @@
 import org.apache.streampark.flink.client.FlinkClient;
 import org.apache.streampark.flink.client.bean.CancelRequest;
 import org.apache.streampark.flink.client.bean.CancelResponse;
+import org.apache.streampark.flink.client.bean.JobClientTarget;
+import org.apache.streampark.flink.client.bean.SavepointCancelOptions;
+import org.apache.streampark.flink.client.bean.SubmitApplicationSpec;
+import org.apache.streampark.flink.client.bean.SubmitClusterSpec;
 import org.apache.streampark.flink.client.bean.SubmitRequest;
 import org.apache.streampark.flink.client.bean.SubmitResponse;
 import org.apache.streampark.flink.kubernetes.FlinkK8sWatcher;
@@ -309,13 +313,12 @@ public void cancel(FlinkApplication appParam) throws Exception {
                 flinkEnv.getFlinkVersion(),
                 FlinkDeployMode.of(application.getDeployMode()),
                 properties,
-                clusterId,
-                application.getJobId(),
-                appParam.getRestoreOrTriggerSavepoint(),
-                appParam.getDrain(),
-                customSavepoint,
-                appParam.getNativeFormat(),
-                namespace);
+                new JobClientTarget(clusterId, application.getJobId(), namespace),
+                new SavepointCancelOptions(
+                    appParam.getRestoreOrTriggerSavepoint(),
+                    appParam.getDrain(),
+                    customSavepoint,
+                    appParam.getNativeFormat()));
 
         final Date triggerTime = new Date();
         CompletableFuture<CancelResponse> cancelFuture =
@@ -459,22 +462,25 @@ public void start(FlinkApplication appParam, boolean auto) throws Exception {
                 flinkEnv.getFlinkVersion(),
                 FlinkDeployMode.of(application.getDeployMode()),
                 getProperties(application, dynamicProperties),
-                flinkEnv.getFlinkConf(),
-                FlinkJobType.of(application.getJobType()),
-                application.getId(),
-                new JobID().toHexString(),
-                application.getJobName(),
-                appConf,
-                application.getApplicationType(),
-                getSavepointPath(appParam),
-                FlinkRestoreMode.of(appParam.getRestoreMode()),
-                applicationArgs,
-                k8sClusterId,
-                application.getHadoopUser(),
+                SubmitApplicationSpec.builder()
+                    .flinkYaml(flinkEnv.getFlinkConf())
+                    .jobType(FlinkJobType.of(application.getJobType()))
+                    .id(application.getId())
+                    .jobId(new JobID().toHexString())
+                    .appName(application.getJobName())
+                    .appConf(appConf)
+                    .applicationType(application.getApplicationType())
+                    .savePoint(getSavepointPath(appParam))
+                    .restoreMode(FlinkRestoreMode.of(appParam.getRestoreMode()))
+                    .args(applicationArgs)
+                    .build(),
+                new SubmitClusterSpec(
+                    k8sClusterId,
+                    application.getHadoopUser(),
+                    k8sNamespace,
+                    exposedType),
                 buildResult,
-                extraParameter,
-                k8sNamespace,
-                exposedType);
+                extraParameter);
 
         CompletableFuture<SubmitResponse> future =
             CompletableFuture.supplyAsync(() -> FlinkClient.submit(submitRequest), executorService);

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

@@ -46,7 +46,9 @@
 import org.apache.streampark.console.core.util.ServiceHelper;
 import org.apache.streampark.console.core.watcher.FlinkAppHttpWatcher;
 import org.apache.streampark.flink.client.FlinkClient;
+import org.apache.streampark.flink.client.bean.JobClientTarget;
 import org.apache.streampark.flink.client.bean.SavepointResponse;
+import org.apache.streampark.flink.client.bean.SavepointTriggerOptions;
 import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
 import org.apache.streampark.flink.util.FlinkUtils;
 
@@ -503,10 +505,7 @@ private TriggerSavepointRequest renderTriggerSavepointRequest(
             flinkEnv.getFlinkVersion(),
             application.getDeployModeEnum(),
             properties,
-            clusterId,
-            application.getJobId(),
-            customSavepoint,
-            nativeFormat,
-            application.getK8sNamespace());
+            new JobClientTarget(clusterId, application.getJobId(), application.getK8sNamespace()),
+            new SavepointTriggerOptions(customSavepoint, nativeFormat));
     }
 }

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

@@ -72,6 +72,12 @@
             <artifactId>hadoop-client-runtime</artifactId>
             <scope>provided</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <profiles>

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

@@ -15,20 +15,23 @@
  * limitations under the License.
  */
 
-package org.apache.streampark.flink.client.bean
+package org.apache.streampark.flink.client;
 
-import org.apache.streampark.common.conf.FlinkVersion
-import org.apache.streampark.common.enums.FlinkDeployMode
-import org.apache.streampark.common.util.Implicits.JavaMap
+import java.security.Permission;
 
-import javax.annotation.Nullable
+/** Used to mask JVM requests for external operations. */
+public class ExitSecurityManager extends SecurityManager {
 
-trait DeployRequestTrait {
+    @Override
+    public void checkExit(int status) {
+        throw new SecurityException(
+            "System.exit("
+                + status
+                + ") was called in your flink job, The job has been stopped, please check your program...");
+    }
 
-  val flinkVersion: FlinkVersion
-  val deployMode: FlinkDeployMode
-  val properties: JavaMap[String, Any]
-  val clusterId: String
-  val id: Long
-  @Nullable val k8sParam: KubernetesDeployParam
+    @Override
+    public void checkPermission(Permission perm) {
+        // no-op
+    }
 }

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

@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.client;
+
+import org.apache.streampark.common.conf.FlinkVersion;
+import org.apache.streampark.common.util.LoggerSupport;
+import org.apache.streampark.flink.client.bean.CancelRequest;
+import org.apache.streampark.flink.client.bean.CancelResponse;
+import org.apache.streampark.flink.client.bean.DeployRequest;
+import org.apache.streampark.flink.client.bean.DeployResponse;
+import org.apache.streampark.flink.client.bean.SavepointResponse;
+import org.apache.streampark.flink.client.bean.ShutDownRequest;
+import org.apache.streampark.flink.client.bean.ShutDownResponse;
+import org.apache.streampark.flink.client.bean.SubmitRequest;
+import org.apache.streampark.flink.client.bean.SubmitResponse;
+import org.apache.streampark.flink.client.bean.TriggerSavepointRequest;
+import org.apache.streampark.flink.proxy.FlinkShimsProxy;
+
+import java.util.function.Function;
+
+public final class FlinkClient extends LoggerSupport {
+
+    private static final String FLINK_CLIENT_ENTRYPOINT_CLASS =
+        "org.apache.streampark.flink.client.FlinkClientEntrypoint";
+
+    private static final String SUBMIT_REQUEST =
+        "org.apache.streampark.flink.client.bean.SubmitRequest";
+
+    private static final String DEPLOY_REQUEST =
+        "org.apache.streampark.flink.client.bean.DeployRequest";
+
+    private static final String CANCEL_REQUEST =
+        "org.apache.streampark.flink.client.bean.CancelRequest";
+
+    private static final String SHUTDOWN_REQUEST =
+        "org.apache.streampark.flink.client.bean.ShutDownRequest";
+
+    private static final String SAVEPOINT_REQUEST =
+        "org.apache.streampark.flink.client.bean.TriggerSavepointRequest";
+
+    private FlinkClient() {
+    }
+
+    public static SubmitResponse submit(SubmitRequest submitRequest) {
+        SecurityManager securityManager = System.getSecurityManager();
+        try {
+            System.setSecurityManager(new ExitSecurityManager());
+            return proxy(submitRequest, submitRequest.flinkVersion(), SUBMIT_REQUEST, "submit");
+        } finally {
+            System.setSecurityManager(securityManager);
+        }
+    }
+
+    public static CancelResponse cancel(CancelRequest stopRequest) {
+        return proxy(stopRequest, stopRequest.flinkVersion(), CANCEL_REQUEST, "cancel");
+    }
+
+    public static DeployResponse deploy(DeployRequest deployRequest) {
+        return proxy(deployRequest, deployRequest.flinkVersion(), DEPLOY_REQUEST, "deploy");
+    }
+
+    public static ShutDownResponse shutdown(ShutDownRequest shutDownRequest) {
+        return proxy(shutDownRequest, shutDownRequest.flinkVersion(), SHUTDOWN_REQUEST, "shutdown");
+    }
+
+    public static SavepointResponse triggerSavepoint(TriggerSavepointRequest savepointRequest) {
+        return proxy(
+            savepointRequest, savepointRequest.flinkVersion(), SAVEPOINT_REQUEST, "triggerSavepoint");
+    }
+
+    @SuppressWarnings("unchecked")
+    private static <T> T proxy(
+                               Object request,
+                               FlinkVersion flinkVersion,
+                               String requestClassName,
+                               String methodName) {
+        flinkVersion.checkVersion();
+        return FlinkShimsProxy.proxy(
+            flinkVersion,
+            (Function<ClassLoader, T>) classLoader -> {
+                try {
+                    Class<?> submitClass = classLoader.loadClass(FLINK_CLIENT_ENTRYPOINT_CLASS);
+                    Class<?> requestClass = classLoader.loadClass(requestClassName);
+                    java.lang.reflect.Method method =
+                        submitClass.getDeclaredMethod(methodName, requestClass);
+                    method.setAccessible(true);
+                    Object obj =
+                        method.invoke(
+                            null, FlinkShimsProxy.getObject(classLoader, request));
+                    if (obj == null) {
+                        return null;
+                    }
+                    return FlinkShimsProxy.getObject(FlinkClient.class.getClassLoader(), obj);
+                } catch (RuntimeException e) {
+                    throw e;
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            });
+    }
+}

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

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.flink.client.bean;
+
+import org.apache.streampark.common.conf.FlinkVersion;
+import org.apache.streampark.common.enums.FlinkDeployMode;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/** Shared fields for deploy and shutdown client requests. */
+abstract class AbstractDeployClientRequest implements DeployRequestTrait, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final FlinkVersion flinkVersion;
+    private final FlinkDeployMode deployMode;
+    @Nullable
+    private final Map<String, Serializable> properties;
+    private final String clusterId;
+    private final long id;
+    @Nullable
+    private final KubernetesDeployParam k8sParam;
+
+    AbstractDeployClientRequest(
+                                FlinkVersion flinkVersion,
+                                FlinkDeployMode deployMode,
+                                @Nullable Map<String, Object> properties,
+                                String clusterId,
+                                long id,
+                                @Nullable KubernetesDeployParam k8sParam) {
+        this.flinkVersion = flinkVersion;
+        this.deployMode = deployMode;
+        this.properties = ClientBeanUtils.toSerializableMap(properties);
+        this.clusterId = clusterId;
+        this.id = id;
+        this.k8sParam = k8sParam;
+    }
+
+    @Override
+    public FlinkVersion flinkVersion() {
+        return flinkVersion;
+    }
+
+    @Override
+    public FlinkDeployMode deployMode() {
+        return deployMode;
+    }
+
+    @Override
+    @Nullable
+    public Map<String, Object> properties() {
+        return ClientBeanUtils.copyPropertiesMap(properties);
+    }
+
+    @Override
+    public String clusterId() {
+        return clusterId;
+    }
+
+    @Override
+    public long id() {
+        return id;
+    }
+
+    @Override
+    @Nullable
+    public KubernetesDeployParam k8sParam() {
+        return k8sParam;
+    }
+}