You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/02/08 07:21:35 UTC
[incubator-streampark] branch dev updated: [Improve] Rename streampark-flink-submit module (#2304)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new 9a98f12b2 [Improve] Rename streampark-flink-submit module (#2304)
9a98f12b2 is described below
commit 9a98f12b21b9a5fb0985ebe1acc0c9127735cc27
Author: benjobs <be...@apache.org>
AuthorDate: Wed Feb 8 15:21:29 2023 +0800
[Improve] Rename streampark-flink-submit module (#2304)
* [Improve] streampark-flink-submit rename to streampark-flink-client
* FlinkClients rename to FlinkClientHandler
---
.github/labeler.yml | 4 +--
.../streampark-console-service/pom.xml | 6 ++--
.../console/core/entity/FlinkCluster.java | 4 +--
.../core/service/impl/ApplicationServiceImpl.java | 20 ++++++-------
.../core/service/impl/FlinkClusterServiceImpl.java | 16 +++++-----
streampark-flink/pom.xml | 2 +-
.../pom.xml | 8 ++---
.../streampark-flink-client-api}/pom.xml | 6 ++--
.../streampark/flink/client/FlinkClient.scala} | 24 +++++++--------
.../flink/client}/bean/CancelRequest.scala | 2 +-
.../flink/client}/bean/CancelResponse.scala | 2 +-
.../flink/client}/bean/DeployRequest.scala | 4 +--
.../flink/client}/bean/DeployResponse.scala | 2 +-
.../flink/client}/bean/ShutDownRequest.scala | 2 +-
.../flink/client}/bean/ShutDownResponse.scala | 2 +-
.../flink/client}/bean/SubmitRequest.scala | 4 +--
.../flink/client}/bean/SubmitResponse.scala | 2 +-
.../streampark-flink-client-core}/pom.xml | 8 ++---
...flink.client.deployment.StandaloneClientFactory | 0
...flink.kubernetes.KubernetesClusterClientFactory | 0
.../org.apache.flink.yarn.YarnClusterClientFactory | 0
.../flink/client/FlinkClientHandler.scala} | 16 +++++-----
.../impl/KubernetesNativeApplicationSubmit.scala | 6 ++--
.../impl/KubernetesNativeSessionSubmit.scala | 8 ++---
.../flink/client}/impl/LocalSubmit.scala | 6 ++--
.../flink/client}/impl/RemoteSubmit.scala | 8 ++---
.../flink/client}/impl/YarnApplicationSubmit.scala | 6 ++--
.../flink/client}/impl/YarnPerJobSubmit.scala | 6 ++--
.../flink/client}/impl/YarnSessionSubmit.scala | 6 ++--
.../client}/tool/FlinkSessionSubmitHelper.scala | 6 ++--
.../flink/client}/trait/FlinkSubmitTrait.scala | 34 +++++++++++-----------
.../trait/KubernetesNativeSubmitTrait.scala | 6 ++--
.../flink/client}/trait/YarnSubmitTrait.scala | 10 +++----
.../flink/client}/test/ParameterTestCase.scala | 6 ++--
.../flink/client}/test/YarnPerJobTestCase.scala | 4 +--
35 files changed, 123 insertions(+), 123 deletions(-)
diff --git a/.github/labeler.yml b/.github/labeler.yml
index 448708458..19b4b5ae0 100644
--- a/.github/labeler.yml
+++ b/.github/labeler.yml
@@ -54,11 +54,11 @@ SPARK:
KUBERNETES:
- "streampark-flink/streampark-flink-kubernetes/**/*"
- - "streampark-flink-submit/**/Kubernetes*.scala"
+ - "streampark-flink-client/**/Kubernetes*.scala"
YARN:
- "streampark-common/**/Yarn*.scala"
- - "streampark-flink-submit/**/Yarn*.scala"
+ - "streampark-flink-client/**/Yarn*.scala"
WINDOWS:
- "**/*.cmd"
diff --git a/streampark-console/streampark-console-service/pom.xml b/streampark-console/streampark-console-service/pom.xml
index 887d6aaa0..387c89a3c 100644
--- a/streampark-console/streampark-console-service/pom.xml
+++ b/streampark-console/streampark-console-service/pom.xml
@@ -349,7 +349,7 @@
<dependency>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit-api_${scala.binary.version}</artifactId>
+ <artifactId>streampark-flink-client-api_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
@@ -524,7 +524,7 @@
<!-- flink-submit-core -->
<dependency>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit-core_${scala.binary.version}</artifactId>
+ <artifactId>streampark-flink-client-core_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<outputDirectory>${project.build.directory}/lib</outputDirectory>
</dependency>
@@ -604,7 +604,7 @@
<!-- flink-submit-core -->
<dependency>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit-core_${scala.binary.version}</artifactId>
+ <artifactId>streampark-flink-client-core_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<outputDirectory>${project.build.directory}/lib</outputDirectory>
</dependency>
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
index 41512fbc0..f9098a0e2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
@@ -27,7 +27,7 @@ import org.apache.streampark.common.util.YarnUtils;
import org.apache.streampark.console.base.util.CommonUtils;
import org.apache.streampark.console.base.util.JacksonUtils;
import org.apache.streampark.console.core.metrics.flink.Overview;
-import org.apache.streampark.flink.submit.FlinkSubmitter;
+import org.apache.streampark.flink.client.FlinkClient;
import com.baomidou.mybatisplus.annotation.FieldStrategy;
import com.baomidou.mybatisplus.annotation.IdType;
@@ -202,7 +202,7 @@ public class FlinkCluster implements Serializable {
public Map<String, Object> getProperties() {
Map<String, Object> map = new HashMap<>();
Map<String, String> dynamicProperties =
- FlinkSubmitter.extractDynamicPropertiesAsJava(this.getDynamicProperties());
+ FlinkClient.extractDynamicPropertiesAsJava(this.getDynamicProperties());
map.putAll(this.getOptionMap());
map.putAll(dynamicProperties);
ResolveOrder resolveOrder = ResolveOrder.of(this.getResolveOrder());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index f553e7229..8f38e0f2a 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -78,6 +78,12 @@ import org.apache.streampark.console.core.service.SavePointService;
import org.apache.streampark.console.core.service.SettingService;
import org.apache.streampark.console.core.service.VariableService;
import org.apache.streampark.console.core.task.FlinkRESTAPIWatcher;
+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.KubernetesSubmitParam;
+import org.apache.streampark.flink.client.bean.SubmitRequest;
+import org.apache.streampark.flink.client.bean.SubmitResponse;
import org.apache.streampark.flink.core.conf.ParameterCli;
import org.apache.streampark.flink.kubernetes.FlinkK8sWatcher;
import org.apache.streampark.flink.kubernetes.IngressController;
@@ -87,12 +93,6 @@ import org.apache.streampark.flink.kubernetes.model.TrackId;
import org.apache.streampark.flink.packer.pipeline.BuildResult;
import org.apache.streampark.flink.packer.pipeline.DockerImageBuildResponse;
import org.apache.streampark.flink.packer.pipeline.ShadedBuildResponse;
-import org.apache.streampark.flink.submit.FlinkSubmitter;
-import org.apache.streampark.flink.submit.bean.CancelRequest;
-import org.apache.streampark.flink.submit.bean.CancelResponse;
-import org.apache.streampark.flink.submit.bean.KubernetesSubmitParam;
-import org.apache.streampark.flink.submit.bean.SubmitRequest;
-import org.apache.streampark.flink.submit.bean.SubmitResponse;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -1233,7 +1233,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
properties);
CompletableFuture<CancelResponse> cancelFuture =
- CompletableFuture.supplyAsync(() -> FlinkSubmitter.cancel(cancelRequest), executorService);
+ CompletableFuture.supplyAsync(() -> FlinkClient.cancel(cancelRequest), executorService);
cancelFutureMap.put(application.getId(), cancelFuture);
@@ -1528,7 +1528,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
extraParameter);
CompletableFuture<SubmitResponse> future =
- CompletableFuture.supplyAsync(() -> FlinkSubmitter.submit(submitRequest), executorService);
+ CompletableFuture.supplyAsync(() -> FlinkClient.submit(submitRequest), executorService);
startFutureMap.put(application.getId(), future);
@@ -1651,7 +1651,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
Map<String, String> dynamicProperties =
- FlinkSubmitter.extractDynamicPropertiesAsJava(application.getDynamicProperties());
+ FlinkClient.extractDynamicPropertiesAsJava(application.getDynamicProperties());
properties.putAll(dynamicProperties);
ResolveOrder resolveOrder = ResolveOrder.of(application.getResolveOrder());
if (resolveOrder != null) {
@@ -1691,7 +1691,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
// 1) properties have the highest priority, read the properties are set: -Dstate.savepoints.dir
String savepointPath =
- FlinkSubmitter.extractDynamicPropertiesAsJava(application.getDynamicProperties())
+ FlinkClient.extractDynamicPropertiesAsJava(application.getDynamicProperties())
.get(CheckpointingOptions.SAVEPOINT_DIRECTORY.key());
// Application conf configuration has the second priority. If it is a streampark|flinksql type
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
index 2e285cd98..9a586d8dc 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java
@@ -32,12 +32,12 @@ import org.apache.streampark.console.core.service.CommonService;
import org.apache.streampark.console.core.service.FlinkClusterService;
import org.apache.streampark.console.core.service.FlinkEnvService;
import org.apache.streampark.console.core.task.FlinkRESTAPIWatcher;
-import org.apache.streampark.flink.submit.FlinkSubmitter;
-import org.apache.streampark.flink.submit.bean.DeployRequest;
-import org.apache.streampark.flink.submit.bean.DeployResponse;
-import org.apache.streampark.flink.submit.bean.KubernetesDeployParam;
-import org.apache.streampark.flink.submit.bean.ShutDownRequest;
-import org.apache.streampark.flink.submit.bean.ShutDownResponse;
+import org.apache.streampark.flink.client.FlinkClient;
+import org.apache.streampark.flink.client.bean.DeployRequest;
+import org.apache.streampark.flink.client.bean.DeployResponse;
+import org.apache.streampark.flink.client.bean.KubernetesDeployParam;
+import org.apache.streampark.flink.client.bean.ShutDownRequest;
+import org.apache.streampark.flink.client.bean.ShutDownResponse;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
@@ -167,7 +167,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
kubernetesDeployParam);
log.info("deploy cluster request " + deployRequest);
Future<DeployResponse> future =
- executorService.submit(() -> FlinkSubmitter.deploy(deployRequest));
+ executorService.submit(() -> FlinkClient.deploy(deployRequest));
DeployResponse deployResponse = future.get(60, TimeUnit.SECONDS);
if (deployResponse != null) {
if (ExecutionMode.YARN_SESSION.equals(executionModeEnum)) {
@@ -284,7 +284,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
try {
Future<ShutDownResponse> future =
- executorService.submit(() -> FlinkSubmitter.shutdown(stopRequest));
+ executorService.submit(() -> FlinkClient.shutdown(stopRequest));
ShutDownResponse shutDownResponse = future.get(60, TimeUnit.SECONDS);
if (shutDownResponse != null) {
flinkCluster.setAddress(null);
diff --git a/streampark-flink/pom.xml b/streampark-flink/pom.xml
index 5f316eb69..b385fb743 100644
--- a/streampark-flink/pom.xml
+++ b/streampark-flink/pom.xml
@@ -34,7 +34,7 @@
<module>streampark-flink-connector</module>
<module>streampark-flink-sqlclient</module>
<module>streampark-flink-udf</module>
- <module>streampark-flink-submit</module>
+ <module>streampark-flink-client</module>
<module>streampark-flink-proxy</module>
<module>streampark-flink-packer</module>
<module>streampark-flink-kubernetes</module>
diff --git a/streampark-flink/streampark-flink-submit/pom.xml b/streampark-flink/streampark-flink-client/pom.xml
similarity index 88%
rename from streampark-flink/streampark-flink-submit/pom.xml
rename to streampark-flink/streampark-flink-client/pom.xml
index ff96f4af3..b8a9efc7e 100644
--- a/streampark-flink/streampark-flink-submit/pom.xml
+++ b/streampark-flink/streampark-flink-client/pom.xml
@@ -23,13 +23,13 @@
<version>2.1.0</version>
</parent>
- <artifactId>streampark-flink-submit</artifactId>
- <name>StreamPark : Flink Submit</name>
+ <artifactId>streampark-flink-client</artifactId>
+ <name>StreamPark : Flink Client</name>
<packaging>pom</packaging>
<modules>
- <module>streampark-flink-submit-api</module>
- <module>streampark-flink-submit-core</module>
+ <module>streampark-flink-client-api</module>
+ <module>streampark-flink-client-core</module>
</modules>
<profiles>
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/pom.xml b/streampark-flink/streampark-flink-client/streampark-flink-client-api/pom.xml
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/pom.xml
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/pom.xml
index 5ec623800..5ba2b2e3e 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/pom.xml
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/pom.xml
@@ -19,12 +19,12 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit</artifactId>
+ <artifactId>streampark-flink-client</artifactId>
<version>2.1.0</version>
</parent>
- <artifactId>streampark-flink-submit-api_${scala.binary.version}</artifactId>
- <name>StreamPark : Flink Submit Api</name>
+ <artifactId>streampark-flink-client-api_${scala.binary.version}</artifactId>
+ <name>StreamPark : Flink Client Api</name>
<dependencies>
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmitter.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
similarity index 87%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmitter.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
index e6a01356d..bb25d1b97 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmitter.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/FlinkClient.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit
+package org.apache.streampark.flink.client
import java.util
import java.util.{Map => JavaMap}
@@ -29,9 +29,9 @@ import org.apache.commons.lang3.StringUtils
import org.apache.streampark.common.util.{Logger, Utils}
import org.apache.streampark.flink.proxy.FlinkShimsProxy
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.bean._
-object FlinkSubmitter extends Logger {
+object FlinkClient extends Logger {
private[this] lazy val PROPERTY_PATTERN = Pattern.compile("(.*?)=(.*?)")
@@ -39,21 +39,21 @@ object FlinkSubmitter extends Logger {
private[this] lazy val MULTI_PROPERTY_PATTERN = Pattern.compile(MULTI_PROPERTY_REGEXP)
- private[this] val FLINK_SUBMIT_CLASS_NAME = "org.apache.streampark.flink.submit.FlinkSubmit"
+ private[this] val FLINK_CLIENT_HANDLER_CLASS_NAME = "org.apache.streampark.flink.client.FlinkClientHandler"
- private[this] val SUBMIT_REQUEST_CLASS_NAME = "org.apache.streampark.flink.submit.bean.SubmitRequest"
+ private[this] val SUBMIT_REQUEST_CLASS_NAME = "org.apache.streampark.flink.client.bean.SubmitRequest"
- private[this] val DEPLOY_REQUEST_CLASS_NAME = "org.apache.streampark.flink.submit.bean.DeployRequest"
+ private[this] val DEPLOY_REQUEST_CLASS_NAME = "org.apache.streampark.flink.client.bean.DeployRequest"
- private[this] val CANCEL_REQUEST_CLASS_NAME = "org.apache.streampark.flink.submit.bean.CancelRequest"
+ private[this] val CANCEL_REQUEST_CLASS_NAME = "org.apache.streampark.flink.client.bean.CancelRequest"
- private[this] val SHUTDOWN_REQUEST_CLASS_NAME = "org.apache.streampark.flink.submit.bean.ShutDownRequest"
+ private[this] val SHUTDOWN_REQUEST_CLASS_NAME = "org.apache.streampark.flink.client.bean.ShutDownRequest"
def submit(submitRequest: SubmitRequest): SubmitResponse = {
FlinkShimsProxy.proxy(
submitRequest.flinkVersion,
(classLoader: ClassLoader) => {
- val submitClass = classLoader.loadClass(FLINK_SUBMIT_CLASS_NAME)
+ val submitClass = classLoader.loadClass(FLINK_CLIENT_HANDLER_CLASS_NAME)
val requestClass = classLoader.loadClass(SUBMIT_REQUEST_CLASS_NAME)
val method = submitClass.getDeclaredMethod("submit", requestClass)
method.setAccessible(true)
@@ -66,7 +66,7 @@ object FlinkSubmitter extends Logger {
FlinkShimsProxy.proxy(
stopRequest.flinkVersion,
(classLoader: ClassLoader) => {
- val submitClass = classLoader.loadClass(FLINK_SUBMIT_CLASS_NAME)
+ val submitClass = classLoader.loadClass(FLINK_CLIENT_HANDLER_CLASS_NAME)
val requestClass = classLoader.loadClass(CANCEL_REQUEST_CLASS_NAME)
val method = submitClass.getDeclaredMethod("cancel", requestClass)
method.setAccessible(true)
@@ -82,7 +82,7 @@ object FlinkSubmitter extends Logger {
FlinkShimsProxy.proxy(
deployRequest.flinkVersion,
(classLoader: ClassLoader) => {
- val submitClass = classLoader.loadClass(FLINK_SUBMIT_CLASS_NAME)
+ val submitClass = classLoader.loadClass(FLINK_CLIENT_HANDLER_CLASS_NAME)
val requestClass = classLoader.loadClass(DEPLOY_REQUEST_CLASS_NAME)
val method = submitClass.getDeclaredMethod("deploy", requestClass)
method.setAccessible(true)
@@ -95,7 +95,7 @@ object FlinkSubmitter extends Logger {
FlinkShimsProxy.proxy(
shutDownRequest.flinkVersion,
(classLoader: ClassLoader) => {
- val submitClass = classLoader.loadClass(FLINK_SUBMIT_CLASS_NAME)
+ val submitClass = classLoader.loadClass(FLINK_CLIENT_HANDLER_CLASS_NAME)
val requestClass = classLoader.loadClass(SHUTDOWN_REQUEST_CLASS_NAME)
val method = submitClass.getDeclaredMethod("shutdown", requestClass)
method.setAccessible(true)
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelRequest.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
index ca8e541ff..275a378bb 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
import java.util.{Map => JavaMap}
import javax.annotation.Nullable
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelResponse.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelResponse.scala
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelResponse.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelResponse.scala
index bdeb35d99..20f11bd46 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/CancelResponse.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelResponse.scala
@@ -15,6 +15,6 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
case class CancelResponse(savePointDir: String)
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployRequest.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
index 456115777..aee42849e 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
import java.io.File
import java.util.{Map => JavaMap}
@@ -35,7 +35,7 @@ case class DeployRequest(
properties: JavaMap[String, Any],
@Nullable k8sDeployParam: KubernetesDeployParam) {
- private[submit] lazy val hdfsWorkspace = {
+ private[client] lazy val hdfsWorkspace = {
/**
* You must keep the flink version and configuration in the native flink and hdfs exactly the same.
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployResponse.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployResponse.scala
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployResponse.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployResponse.scala
index 800de3059..91b6ecad0 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/DeployResponse.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployResponse.scala
@@ -15,6 +15,6 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
case class DeployResponse(address: String, clusterId: String)
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownRequest.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
index 0dc83307e..fd2b711b3 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
import java.util.{Map => JavaMap}
import javax.annotation.Nullable
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownResponse.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownResponse.scala
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownResponse.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownResponse.scala
index aed5f8637..5c9a14728 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/ShutDownResponse.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownResponse.scala
@@ -15,6 +15,6 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
case class ShutDownResponse()
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
similarity index 98%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitRequest.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
index 91b96160f..b09018778 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
import java.io.File
import java.util.{Map => JavaMap}
@@ -135,7 +135,7 @@ case class SubmitRequest(
}
}
- private[submit] lazy val hdfsWorkspace = {
+ private[client] lazy val hdfsWorkspace = {
/**
* 必须保持本机flink和hdfs里的flink版本和配置都完全一致.
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitResponse.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitResponse.scala
similarity index 95%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitResponse.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitResponse.scala
index df43e9b73..c44987753 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-api/src/main/scala/org/apache/streampark/flink/submit/bean/SubmitResponse.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitResponse.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.bean
+package org.apache.streampark.flink.client.bean
import java.util.{Map => JavaMap}
import javax.annotation.Nullable
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/pom.xml b/streampark-flink/streampark-flink-client/streampark-flink-client-core/pom.xml
similarity index 93%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/pom.xml
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/pom.xml
index ff86d539f..8c4aaa4f5 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/pom.xml
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/pom.xml
@@ -19,18 +19,18 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit</artifactId>
+ <artifactId>streampark-flink-client</artifactId>
<version>2.1.0</version>
</parent>
- <artifactId>streampark-flink-submit-core_${scala.binary.version}</artifactId>
- <name>StreamPark : Flink Submit Core</name>
+ <artifactId>streampark-flink-client-core_${scala.binary.version}</artifactId>
+ <name>StreamPark : Flink Client Core</name>
<dependencies>
<dependency>
<groupId>org.apache.streampark</groupId>
- <artifactId>streampark-flink-submit-api_${scala.binary.version}</artifactId>
+ <artifactId>streampark-flink-client-api_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.client.deployment.StandaloneClientFactory b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.client.deployment.StandaloneClientFactory
similarity index 100%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.client.deployment.StandaloneClientFactory
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.client.deployment.StandaloneClientFactory
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.kubernetes.KubernetesClusterClientFactory b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.kubernetes.KubernetesClusterClientFactory
similarity index 100%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.kubernetes.KubernetesClusterClientFactory
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.kubernetes.KubernetesClusterClientFactory
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.yarn.YarnClusterClientFactory b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.yarn.YarnClusterClientFactory
similarity index 100%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/resources/META-INF/services/org.apache.flink.yarn.YarnClusterClientFactory
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/resources/META-INF/services/org.apache.flink.yarn.YarnClusterClientFactory
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientHandler.scala
similarity index 89%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientHandler.scala
index fc80748c7..86200d2b3 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/FlinkSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientHandler.scala
@@ -15,13 +15,13 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit
+package org.apache.streampark.flink.client
import org.apache.streampark.common.enums.ExecutionMode
-import org.apache.streampark.flink.submit.bean._
-import org.apache.streampark.flink.submit.impl._
+import org.apache.streampark.flink.client.bean._
+import org.apache.streampark.flink.client.impl._
-object FlinkSubmit {
+object FlinkClientHandler {
def submit(submitInfo: SubmitRequest): SubmitResponse = {
submitInfo.executionMode match {
@@ -32,7 +32,7 @@ object FlinkSubmit {
case ExecutionMode.YARN_PER_JOB => YarnPerJobSubmit.submit(submitInfo)
case ExecutionMode.KUBERNETES_NATIVE_SESSION => KubernetesNativeSessionSubmit.submit(submitInfo)
case ExecutionMode.KUBERNETES_NATIVE_APPLICATION => KubernetesNativeApplicationSubmit.submit(submitInfo)
- case _ => throw new UnsupportedOperationException(s"Unsupported ${submitInfo.executionMode} Submit ")
+ case _ => throw new UnsupportedOperationException(s"Unsupported ${submitInfo.executionMode} submit ")
}
}
@@ -45,7 +45,7 @@ object FlinkSubmit {
case ExecutionMode.YARN_PER_JOB | ExecutionMode.YARN_SESSION => YarnPerJobSubmit.cancel(cancelRequest)
case ExecutionMode.KUBERNETES_NATIVE_SESSION => KubernetesNativeSessionSubmit.cancel(cancelRequest)
case ExecutionMode.KUBERNETES_NATIVE_APPLICATION => KubernetesNativeApplicationSubmit.cancel(cancelRequest)
- case _ => throw new UnsupportedOperationException(s"Unsupported ${cancelRequest.executionMode} Submit ")
+ case _ => throw new UnsupportedOperationException(s"Unsupported ${cancelRequest.executionMode} cancel ")
}
}
@@ -53,7 +53,7 @@ object FlinkSubmit {
deployRequest.executionMode match {
case ExecutionMode.YARN_SESSION => YarnSessionSubmit.deploy(deployRequest)
case ExecutionMode.KUBERNETES_NATIVE_SESSION => KubernetesNativeSessionSubmit.deploy(deployRequest)
- case _ => throw new UnsupportedOperationException(s"Unsupported ${deployRequest.executionMode} Submit ")
+ case _ => throw new UnsupportedOperationException(s"Unsupported ${deployRequest.executionMode} deploy cluster ")
}
}
@@ -61,7 +61,7 @@ object FlinkSubmit {
shutDownRequest.executionMode match {
case ExecutionMode.YARN_SESSION => YarnSessionSubmit.shutdown(shutDownRequest)
case ExecutionMode.KUBERNETES_NATIVE_SESSION => KubernetesNativeSessionSubmit.shutdown(shutDownRequest)
- case _ => throw new UnsupportedOperationException(s"Unsupported ${shutDownRequest.executionMode} Submit ")
+ case _ => throw new UnsupportedOperationException(s"Unsupported ${shutDownRequest.executionMode} shutdown cluster ")
}
}
}
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeApplicationSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationSubmit.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeApplicationSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationSubmit.scala
index 935ae2e14..eaa183454 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeApplicationSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import com.google.common.collect.Lists
import org.apache.commons.lang3.StringUtils
@@ -28,8 +28,8 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions
import org.apache.streampark.common.enums.ExecutionMode
import org.apache.streampark.common.util.Utils
import org.apache.streampark.flink.packer.pipeline.DockerImageBuildResponse
-import org.apache.streampark.flink.submit.`trait`.KubernetesNativeSubmitTrait
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.`trait`.KubernetesNativeSubmitTrait
+import org.apache.streampark.flink.client.bean._
/**
* kubernetes native application mode submit
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeSessionSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionSubmit.scala
similarity index 97%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeSessionSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionSubmit.scala
index 23923fe5a..c5132b658 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/KubernetesNativeSessionSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import java.io.File
@@ -38,9 +38,9 @@ import org.apache.streampark.flink.core.FlinkKubernetesClient
import org.apache.streampark.flink.kubernetes.KubernetesRetriever
import org.apache.streampark.flink.kubernetes.enums.FlinkK8sExecuteMode
import org.apache.streampark.flink.kubernetes.model.ClusterKey
-import org.apache.streampark.flink.submit.`trait`.KubernetesNativeSubmitTrait
-import org.apache.streampark.flink.submit.bean._
-import org.apache.streampark.flink.submit.tool.FlinkSessionSubmitHelper
+import org.apache.streampark.flink.client.`trait`.KubernetesNativeSubmitTrait
+import org.apache.streampark.flink.client.bean._
+import org.apache.streampark.flink.client.tool.FlinkSessionSubmitHelper
/**
* kubernetes native session mode submit
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/LocalSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/LocalSubmit.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/LocalSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/LocalSubmit.scala
index 923902aaf..2f05460c7 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/LocalSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/LocalSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import java.lang.{Integer => JavaInt}
@@ -26,8 +26,8 @@ import org.apache.flink.configuration._
import org.apache.flink.runtime.minicluster.{MiniCluster, MiniClusterConfiguration}
import org.apache.streampark.common.util.Utils
-import org.apache.streampark.flink.submit.`trait`.FlinkSubmitTrait
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.`trait`.FlinkSubmitTrait
+import org.apache.streampark.flink.client.bean._
object LocalSubmit extends FlinkSubmitTrait {
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/RemoteSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteSubmit.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/RemoteSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteSubmit.scala
index 0d621950a..7fbb80e82 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/RemoteSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/RemoteSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import java.io.File
import java.lang.{Integer => JavaInt}
@@ -28,9 +28,9 @@ import org.apache.flink.client.program.{ClusterClient, PackagedProgram}
import org.apache.flink.configuration._
import org.apache.streampark.common.util.Utils
-import org.apache.streampark.flink.submit.`trait`.FlinkSubmitTrait
-import org.apache.streampark.flink.submit.bean.{CancelRequest, CancelResponse, SubmitRequest, SubmitResponse}
-import org.apache.streampark.flink.submit.tool.FlinkSessionSubmitHelper
+import org.apache.streampark.flink.client.`trait`.FlinkSubmitTrait
+import org.apache.streampark.flink.client.bean.{CancelRequest, CancelResponse, SubmitRequest, SubmitResponse}
+import org.apache.streampark.flink.client.tool.FlinkSessionSubmitHelper
/**
* Submit Job to Remote Cluster
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnApplicationSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationSubmit.scala
similarity index 97%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnApplicationSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationSubmit.scala
index eed83abb7..0665d585b 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnApplicationSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import java.util.Collections
import java.util.concurrent.Callable
@@ -37,8 +37,8 @@ import org.apache.streampark.common.conf.Workspace
import org.apache.streampark.common.enums.DevelopmentMode
import org.apache.streampark.common.util.{HdfsUtils, Utils}
import org.apache.streampark.flink.packer.pipeline.ShadedBuildResponse
-import org.apache.streampark.flink.submit.`trait`.YarnSubmitTrait
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.`trait`.YarnSubmitTrait
+import org.apache.streampark.flink.client.bean._
/**
* yarn application mode submit
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnPerJobSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnPerJobSubmit.scala
similarity index 97%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnPerJobSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnPerJobSubmit.scala
index 71e6cb15f..050e75bdf 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnPerJobSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnPerJobSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import java.io.File
import java.lang.{Boolean => JavaBool}
@@ -32,8 +32,8 @@ import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.hadoop.yarn.api.records.ApplicationId
import org.apache.streampark.common.util.{FlinkUtils, Utils}
-import org.apache.streampark.flink.submit.`trait`.YarnSubmitTrait
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.`trait`.YarnSubmitTrait
+import org.apache.streampark.flink.client.bean._
/**
* yarn PerJob mode submit
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnSessionSubmit.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnSessionSubmit.scala
similarity index 98%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnSessionSubmit.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnSessionSubmit.scala
index 13097864a..d5e951a55 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/impl/YarnSessionSubmit.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnSessionSubmit.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.impl
+package org.apache.streampark.flink.client.impl
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
@@ -34,8 +34,8 @@ import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException
import org.apache.hadoop.yarn.util.ConverterUtils
import org.apache.streampark.common.util.Utils
-import org.apache.streampark.flink.submit.`trait`.YarnSubmitTrait
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.`trait`.YarnSubmitTrait
+import org.apache.streampark.flink.client.bean._
/**
* Submit Job to YARN Session Cluster
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/tool/FlinkSessionSubmitHelper.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/tool/FlinkSessionSubmitHelper.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.scala
index 372988a5c..a9f9ba039 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/tool/FlinkSessionSubmitHelper.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/tool/FlinkSessionSubmitHelper.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.tool
+package org.apache.streampark.flink.client.tool
import java.io.File
import java.nio.charset.StandardCharsets
@@ -93,7 +93,7 @@ object FlinkSessionSubmitHelper extends Logger {
/**
* refer to https://ci.apache.org/projects/flink/flink-docs-stable/docs/ops/rest_api/#jars-upload
*/
-private[submit] case class JarUploadResponse(filename: String, status: String) {
+private[client] case class JarUploadResponse(filename: String, status: String) {
def isSuccessful: Boolean = "success".equalsIgnoreCase(status)
@@ -103,7 +103,7 @@ private[submit] case class JarUploadResponse(filename: String, status: String) {
/**
* refer to https://ci.apache.org/projects/flink/flink-docs-stable/docs/ops/rest_api/#jars-upload
*/
-private[submit] case class JarRunRequest(entryClass: String, programArgs: String, parallelism: String, savepointPath: String, allowNonRestoredState: Boolean) {
+private[client] case class JarRunRequest(entryClass: String, programArgs: String, parallelism: String, savepointPath: String, allowNonRestoredState: Boolean) {
def this(flinkConf: Configuration) {
this(
entryClass = flinkConf.get(ApplicationConfiguration.APPLICATION_MAIN_CLASS),
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkSubmitTrait.scala
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkSubmitTrait.scala
index fe5ccf22b..2e9c3a508 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/FlinkSubmitTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkSubmitTrait.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.`trait`
+package org.apache.streampark.flink.client.`trait`
import java.io.File
import java.util.{Collections, List => JavaList, Map => JavaMap}
@@ -48,15 +48,15 @@ import org.apache.streampark.common.enums.{ApplicationType, DevelopmentMode, Exe
import org.apache.streampark.common.util.{DeflaterUtils, Logger}
import org.apache.streampark.flink.core.FlinkClusterClient
import org.apache.streampark.flink.core.conf.FlinkRunOption
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.bean._
trait FlinkSubmitTrait extends Logger {
- private[submit] lazy val PARAM_KEY_FLINK_CONF = KEY_FLINK_CONF("--")
- private[submit] lazy val PARAM_KEY_FLINK_SQL = KEY_FLINK_SQL("--")
- private[submit] lazy val PARAM_KEY_APP_CONF = KEY_APP_CONF("--")
- private[submit] lazy val PARAM_KEY_APP_NAME = KEY_APP_NAME("--")
- private[submit] lazy val PARAM_KEY_FLINK_PARALLELISM = KEY_FLINK_PARALLELISM("--")
+ private[client] lazy val PARAM_KEY_FLINK_CONF = KEY_FLINK_CONF("--")
+ private[client] lazy val PARAM_KEY_FLINK_SQL = KEY_FLINK_SQL("--")
+ private[client] lazy val PARAM_KEY_APP_CONF = KEY_APP_CONF("--")
+ private[client] lazy val PARAM_KEY_APP_NAME = KEY_APP_NAME("--")
+ private[client] lazy val PARAM_KEY_FLINK_PARALLELISM = KEY_FLINK_PARALLELISM("--")
@throws[Exception]
def submit(submitRequest: SubmitRequest): SubmitResponse = {
@@ -186,7 +186,7 @@ trait FlinkSubmitTrait extends Logger {
}
}
- private[submit] def getJobGraph(flinkConfig: Configuration, submitRequest: SubmitRequest, jarFile: File): (PackagedProgram, JobGraph) = {
+ private[client] def getJobGraph(flinkConfig: Configuration, submitRequest: SubmitRequest, jarFile: File): (PackagedProgram, JobGraph) = {
val packageProgram = PackagedProgram
.newBuilder
.setJarFile(jarFile)
@@ -206,14 +206,14 @@ trait FlinkSubmitTrait extends Logger {
packageProgram -> jobGraph
}
- private[submit] def getJobID(jobId: String) = Try(JobID.fromHexString(jobId)) match {
+ private[client] def getJobID(jobId: String) = Try(JobID.fromHexString(jobId)) match {
case Success(id) => id
case Failure(e) => throw new CliArgsException(e.getMessage)
}
// ----------Public Method end ------------------
- private[submit] def validateAndGetActiveCommandLine(customCommandLines: JavaList[CustomCommandLine], commandLine: CommandLine): CustomCommandLine = {
+ private[client] def validateAndGetActiveCommandLine(customCommandLines: JavaList[CustomCommandLine], commandLine: CommandLine): CustomCommandLine = {
val line = checkNotNull(commandLine)
logInfo(s"Custom commandline: $customCommandLines")
for (cli <- customCommandLines) {
@@ -224,11 +224,11 @@ trait FlinkSubmitTrait extends Logger {
throw new IllegalStateException("No valid command-line found.")
}
- private[submit] def getFlinkDefaultConfiguration(flinkHome: String): Configuration = {
+ private[client] def getFlinkDefaultConfiguration(flinkHome: String): Configuration = {
Try(GlobalConfiguration.loadConfiguration(s"$flinkHome/conf")).getOrElse(new Configuration())
}
- private[submit] def getOptionFromDefaultFlinkConfig[T](flinkHome: String, option: ConfigOption[T]): T = {
+ private[client] def getOptionFromDefaultFlinkConfig[T](flinkHome: String, option: ConfigOption[T]): T = {
getFlinkDefaultConfiguration(flinkHome).get(option)
}
@@ -242,7 +242,7 @@ trait FlinkSubmitTrait extends Logger {
customCommandLines
}
- private[submit] def getParallelism(submitRequest: SubmitRequest): Integer = {
+ private[client] def getParallelism(submitRequest: SubmitRequest): Integer = {
if (submitRequest.properties.containsKey(KEY_FLINK_PARALLELISM())) {
Integer.valueOf(submitRequest.properties.get(KEY_FLINK_PARALLELISM()).toString)
} else {
@@ -316,7 +316,7 @@ trait FlinkSubmitTrait extends Logger {
}
- private[submit] def getCommandLineOptions(flinkHome: String) = {
+ private[client] def getCommandLineOptions(flinkHome: String) = {
val customCommandLines = getCustomCommandLines(flinkHome)
val customCommandLineOptions = new Options
for (customCommandLine <- customCommandLines) {
@@ -326,7 +326,7 @@ trait FlinkSubmitTrait extends Logger {
FlinkRunOption.mergeOptions(CliFrontendParser.getRunCommandOptions, customCommandLineOptions)
}
- private[submit] def extractConfiguration(flinkHome: String, properties: JavaMap[String, Any]): Configuration = {
+ private[client] def extractConfiguration(flinkHome: String, properties: JavaMap[String, Any]): Configuration = {
val commandLine = {
val commandLineOptions = getCommandLineOptions(flinkHome)
// read and verify user config...
@@ -439,7 +439,7 @@ trait FlinkSubmitTrait extends Logger {
configuration
}
- implicit private[submit] class EnhanceFlinkConfiguration(flinkConfig: Configuration) {
+ implicit private[client] class EnhanceFlinkConfiguration(flinkConfig: Configuration) {
def safeSet[T](option: ConfigOption[T], value: T): Configuration = {
flinkConfig match {
case x if value != null && value.toString.nonEmpty => x.set(option, value)
@@ -448,7 +448,7 @@ trait FlinkSubmitTrait extends Logger {
}
}
- private[submit] def cancelJob(cancelRequest: CancelRequest, jobID: JobID, client: ClusterClient[_]): String = {
+ private[client] def cancelJob(cancelRequest: CancelRequest, jobID: JobID, client: ClusterClient[_]): String = {
val savePointDir = {
if (!cancelRequest.withSavePoint) null;
else {
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/KubernetesNativeSubmitTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeSubmitTrait.scala
similarity index 97%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/KubernetesNativeSubmitTrait.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeSubmitTrait.scala
index fada1719a..325162123 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/KubernetesNativeSubmitTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeSubmitTrait.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.`trait`
+package org.apache.streampark.flink.client.`trait`
import java.io.File
import javax.annotation.Nonnull
@@ -35,14 +35,14 @@ import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.Service
import org.apache.streampark.common.enums.{ExecutionMode, FlinkK8sRestExposedType}
import org.apache.streampark.flink.kubernetes.IngressController
import org.apache.streampark.flink.packer.pipeline.DockerImageBuildResponse
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.bean._
/**
* kubernetes native mode submit
*/
trait KubernetesNativeSubmitTrait extends FlinkSubmitTrait {
- private[submit] val fatJarCached = new mutable.HashMap[String, File]()
+ private[client] val fatJarCached = new mutable.HashMap[String, File]()
override def setConfig(submitRequest: SubmitRequest, flinkConfig: Configuration): Unit = {
// extract from submitRequest
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/YarnSubmitTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/YarnSubmitTrait.scala
similarity index 94%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/YarnSubmitTrait.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/YarnSubmitTrait.scala
index d013094bd..da34f30eb 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/main/scala/org/apache/streampark/flink/submit/trait/YarnSubmitTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/YarnSubmitTrait.scala
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.`trait`
+package org.apache.streampark.flink.client.`trait`
import java.lang.{Boolean => JavaBool}
import java.lang.reflect.Method
@@ -32,7 +32,7 @@ import org.apache.flink.yarn.configuration.YarnConfigOptions
import org.apache.hadoop.yarn.api.records.ApplicationId
import org.apache.streampark.common.util.ExceptionUtils
-import org.apache.streampark.flink.submit.bean._
+import org.apache.streampark.flink.client.bean._
/**
* yarn application mode submit
@@ -74,7 +74,7 @@ trait YarnSubmitTrait extends FlinkSubmitTrait {
deployInternal
}
- private[submit] def deployInternal(
+ private[client] def deployInternal(
clusterDescriptor: YarnClusterDescriptor,
clusterSpecification: ClusterSpecification,
applicationName: String,
@@ -90,7 +90,7 @@ trait YarnSubmitTrait extends FlinkSubmitTrait {
detached).asInstanceOf[ClusterClientProvider[ApplicationId]]
}
- private[submit] def getSessionClusterDescriptor[T <: ClusterDescriptor[ApplicationId]](flinkConfig: Configuration): (ApplicationId, T) = {
+ private[client] def getSessionClusterDescriptor[T <: ClusterDescriptor[ApplicationId]](flinkConfig: Configuration): (ApplicationId, T) = {
val serviceLoader = new DefaultClusterClientServiceLoader
val clientFactory = serviceLoader.getClusterClientFactory[ApplicationId](flinkConfig)
val yarnClusterId: ApplicationId = clientFactory.getClusterId(flinkConfig)
@@ -99,7 +99,7 @@ trait YarnSubmitTrait extends FlinkSubmitTrait {
(yarnClusterId, clusterDescriptor)
}
- private[submit] def getSessionClusterDeployDescriptor[T <: ClusterDescriptor[ApplicationId]](flinkConfig: Configuration): (ClusterSpecification, T) = {
+ private[client] def getSessionClusterDeployDescriptor[T <: ClusterDescriptor[ApplicationId]](flinkConfig: Configuration): (ClusterSpecification, T) = {
val serviceLoader = new DefaultClusterClientServiceLoader
val clientFactory = serviceLoader.getClusterClientFactory[ApplicationId](flinkConfig)
val clusterSpecification = clientFactory.getClusterSpecification(flinkConfig)
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/ParameterTestCase.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/ParameterTestCase.scala
similarity index 96%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/ParameterTestCase.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/ParameterTestCase.scala
index 56a292f02..0cff357eb 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/ParameterTestCase.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/ParameterTestCase.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.test
+package org.apache.streampark.flink.client.test
import scala.collection.mutable.ArrayBuffer
@@ -22,7 +22,7 @@ import org.apache.commons.lang3.StringUtils
import org.apache.flink.api.java.utils.ParameterTool
import org.junit.jupiter.api.{Assertions, Test}
-import org.apache.streampark.flink.submit.FlinkSubmitter
+import org.apache.streampark.flink.client.FlinkClient
class ParameterTestCase {
@@ -121,7 +121,7 @@ class ParameterTestCase {
|
|""".stripMargin
- val map = FlinkSubmitter.extractDynamicProperties(dynamicProperties)
+ val map = FlinkClient.extractDynamicProperties(dynamicProperties)
Assertions.assertEquals(map("env.java.opts1"), "-Dfile.encoding=UTF-8")
Assertions.assertEquals(map("env.java.opts2"), "-Dfile.enc\\\"oding=UTF-8")
Assertions.assertEquals(map("env.java.opts3"), " -Dfile.encoding=UTF-8")
diff --git a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/YarnPerJobTestCase.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/YarnPerJobTestCase.scala
similarity index 98%
rename from streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/YarnPerJobTestCase.scala
rename to streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/YarnPerJobTestCase.scala
index f0e634941..5b3e03560 100644
--- a/streampark-flink/streampark-flink-submit/streampark-flink-submit-core/src/test/scala/org/apache/streampark/flink/submit/test/YarnPerJobTestCase.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/test/scala/org/apache/streampark/flink/client/test/YarnPerJobTestCase.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.streampark.flink.submit.test
+package org.apache.streampark.flink.client.test
import java.io.File
import java.lang.reflect.Method
@@ -41,7 +41,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId
import org.apache.streampark.common.util.Logger
import org.apache.streampark.flink.core.conf.FlinkRunOption
-import org.apache.streampark.flink.submit.bean
+import org.apache.streampark.flink.client.bean
/**
* perJob to submit jobs programmatically,