You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by go...@apache.org on 2023/09/24 12:28:16 UTC
[incubator-streampark] branch dev updated: [Improve] Enumeration improvement (#3182)
This is an automated email from the ASF dual-hosted git repository.
gongzhongqiang 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 093a57dd5 [Improve] Enumeration improvement (#3182)
093a57dd5 is described below
commit 093a57dd5ed100b4ef134797e75b8e69ac939c73
Author: benjobs <be...@apache.org>
AuthorDate: Sun Sep 24 07:28:10 2023 -0500
[Improve] Enumeration improvement (#3182)
* [Improve] Enumeration improvement
* enum minor improvement
---
...plicationTypeEnum.java => ApplicationType.java} | 8 +--
.../{ClusterStateEnum.java => ClusterState.java} | 26 ++++-----
...mentModeEnum.java => FlinkDevelopmentMode.java} | 12 ++--
...cutionModeEnum.java => FlinkExecutionMode.java} | 32 +++++------
...dTypeEnum.java => FlinkK8sRestExposedType.java} | 8 +--
...{RestoreModeEnum.java => FlinkRestoreMode.java} | 12 ++--
...Enum.java => FlinkSqlValidationFailedType.java} | 8 +--
.../{ResolveOrderEnum.java => ResolveOrder.java} | 8 +--
.../enums/{SemanticEnum.java => Semantic.java} | 10 ++--
.../{StorageTypeEnum.java => StorageType.java} | 8 +--
.../apache/streampark/common/conf/Workspace.scala | 14 ++---
.../enums/{ApiTypeEnum.scala => ApiType.scala} | 2 +-
.../{PlannerTypeEnum.scala => PlannerType.scala} | 2 +-
.../enums/{TableModeEnum.scala => TableMode.scala} | 2 +-
.../apache/streampark/common/fs/FsOperator.scala | 12 ++--
.../interceptor/UploadFileTypeInterceptor.java | 2 +-
.../console/core/annotation/AppUpdated.java | 6 +-
.../{StreamParkAspect.java => ConsoleAspect.java} | 4 +-
.../console/core/bean/AlertTemplate.java | 25 ++++----
.../controller/ApplicationHistoryController.java | 4 +-
.../core/controller/FlinkClusterController.java | 6 +-
.../console/core/entity/Application.java | 66 +++++++++++-----------
.../console/core/entity/ApplicationBackUp.java | 4 +-
.../streampark/console/core/entity/Effective.java | 2 +-
.../console/core/entity/FlinkCluster.java | 28 ++++-----
.../streampark/console/core/entity/Resource.java | 4 +-
.../console/core/runner/EnvInitializer.java | 24 ++++----
.../console/core/service/FlinkClusterService.java | 8 +--
.../console/core/service/YarnQueueService.java | 4 +-
.../application/ApplicationManageService.java | 4 +-
.../impl/ApplicationActionServiceImpl.java | 50 ++++++++--------
.../impl/ApplicationInfoServiceImpl.java | 18 +++---
.../impl/ApplicationManageServiceImpl.java | 35 ++++++------
.../core/service/impl/AppBuildPipeServiceImpl.java | 24 ++++----
.../core/service/impl/FlinkClusterServiceImpl.java | 65 ++++++++++-----------
.../core/service/impl/ResourceServiceImpl.java | 8 +--
.../core/service/impl/SavePointServiceImpl.java | 14 ++---
.../core/service/impl/SqlWorkBenchServiceImpl.java | 4 +-
.../core/service/impl/YarnQueueServiceImpl.java | 11 ++--
.../console/core/task/FlinkAppHttpWatcher.java | 32 +++++------
.../console/core/task/FlinkClusterWatcher.java | 60 ++++++++++----------
.../core/task/FlinkK8sChangeEventListener.java | 6 +-
.../console/core/task/FlinkK8sWatcherWrapper.java | 8 +--
.../console/system/controller/UserController.java | 2 +-
.../streampark/console/system/entity/User.java | 4 +-
.../system/security/impl/AuthenticatorImpl.java | 14 ++---
.../system/service/impl/MenuServiceImpl.java | 4 +-
.../system/service/impl/TeamServiceImpl.java | 2 +-
.../system/service/impl/UserServiceImpl.java | 4 +-
.../resources/mapper/core/AlertConfigMapper.xml | 2 +-
.../resources/mapper/core/ApplicationMapper.xml | 16 +++---
.../resources/mapper/core/FlinkClusterMapper.xml | 6 +-
.../main/resources/mapper/core/ProjectMapper.xml | 8 +--
.../main/resources/mapper/system/UserMapper.xml | 2 +-
.../console/core/task/FlinkK8sObserverBroker.scala | 14 ++---
.../core/utils/FlinkK8sDataTypeConverter.scala | 26 ++++-----
.../streampark/console/SpringUnitTestBase.java | 6 +-
.../service/ApplicationManageServiceITest.java | 6 +-
.../core/service/ApplicationManageServiceTest.java | 10 ++--
.../core/service/FlinkClusterServiceTest.java | 4 +-
.../console/core/service/SavePointServiceTest.java | 22 ++++----
.../console/core/service/UserServiceTest.java | 14 ++---
.../console/core/service/YarnQueueServiceTest.java | 12 ++--
.../src/views/base/login/LoginForm.vue | 12 ++--
.../src/views/flink/app/Add.vue | 8 +--
.../src/views/flink/app/Detail.vue | 2 +-
.../src/views/flink/app/EditFlink.vue | 12 ++--
.../src/views/flink/app/EditStreamPark.vue | 16 +++---
.../src/views/flink/app/View.vue | 6 +-
.../components/AppView/StartApplicationModal.vue | 6 +-
.../src/views/flink/project/Add.vue | 2 +-
.../src/views/flink/project/Edit.vue | 10 ++--
.../src/views/flink/project/View.vue | 12 ++--
.../views/flink/project/components/ListItem.vue | 14 ++---
.../flink/project/components/RepositoryGroup.tsx | 10 ++--
.../views/setting/Alarm/components/AlertModal.vue | 24 ++++----
.../setting/Alarm/components/AlertTypeInfo.vue | 18 +++---
.../src/views/setting/Alarm/index.data.ts | 2 +-
.../src/views/setting/Alarm/index.vue | 14 ++---
.../src/views/setting/FlinkCluster/EditCluster.vue | 4 +-
.../src/views/setting/FlinkCluster/index.vue | 18 +++---
.../src/views/system/user/User.vue | 2 +-
.../flink/client/bean/CancelRequest.scala | 4 +-
.../flink/client/bean/DeployRequest.scala | 7 +--
.../flink/client/bean/KubernetesSubmitParam.scala | 6 +-
.../flink/client/bean/SavepointRequestTrait.scala | 4 +-
.../flink/client/bean/ShutDownRequest.scala | 4 +-
.../flink/client/bean/SubmitRequest.scala | 16 +++---
.../client/bean/TriggerSavepointRequest.scala | 4 +-
.../flink/client/FlinkClientEndpoint.scala | 6 +-
.../impl/KubernetesNativeApplicationClient.scala | 8 ++-
.../impl/KubernetesNativeSessionClient.scala | 8 ++-
.../flink/client/impl/YarnApplicationClient.scala | 6 +-
.../flink/client/trait/FlinkClientTrait.scala | 20 +++----
.../client/trait/KubernetesNativeClientTrait.scala | 13 ++---
.../kubernetes/enums/FlinkK8sExecuteModeEnum.scala | 14 ++---
.../flink/packer/pipeline/BuildRequest.scala | 22 ++++----
.../impl/FlinkK8sApplicationBuildPipeline.scala | 6 +-
.../impl/FlinkK8sApplicationBuildPipelineV2.scala | 6 +-
.../pipeline/impl/FlinkRemoteBuildPipeline.scala | 8 +--
.../impl/FlinkYarnApplicationBuildPipeline.scala | 6 +-
.../streampark/flink/core/FlinkSqlValidator.scala | 8 +--
.../flink/core/FlinkStreamingInitializer.scala | 12 ++--
.../flink/core/FlinkTableInitializer.scala | 30 +++++-----
.../streampark/flink/core/SqlCommandParser.scala | 10 ++--
105 files changed, 646 insertions(+), 642 deletions(-)
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationTypeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationType.java
similarity index 87%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationTypeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationType.java
index 557dd4cf8..70c992d82 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationTypeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/ApplicationType.java
@@ -17,7 +17,7 @@
package org.apache.streampark.common.enums;
-public enum ApplicationTypeEnum {
+public enum ApplicationType {
/** StreamPark Flink */
STREAMPARK_FLINK(1, "StreamPark Flink"),
@@ -34,7 +34,7 @@ public enum ApplicationTypeEnum {
private final int type;
private final String name;
- ApplicationTypeEnum(int type, String name) {
+ ApplicationType(int type, String name) {
this.type = type;
this.name = name;
}
@@ -47,8 +47,8 @@ public enum ApplicationTypeEnum {
return name;
}
- public static ApplicationTypeEnum of(int type) {
- for (ApplicationTypeEnum appType : ApplicationTypeEnum.values()) {
+ public static ApplicationType of(int type) {
+ for (ApplicationType appType : ApplicationType.values()) {
if (appType.getType() == type) {
return appType;
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterStateEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
similarity index 71%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterStateEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
index 3cd992f87..b2599b74c 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterStateEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
@@ -18,7 +18,7 @@
package org.apache.streampark.common.enums;
/** @since 1.2.3 */
-public enum ClusterStateEnum {
+public enum ClusterState {
/** The cluster was just created but not started */
CREATED(0),
@@ -49,33 +49,33 @@ public enum ClusterStateEnum {
private final Integer state;
- ClusterStateEnum(Integer state) {
+ ClusterState(Integer state) {
this.state = state;
}
- public static ClusterStateEnum of(Integer value) {
- for (ClusterStateEnum clusterStateEnum : values()) {
- if (clusterStateEnum.state.equals(value)) {
- return clusterStateEnum;
+ public static ClusterState of(Integer value) {
+ for (ClusterState clusterState : values()) {
+ if (clusterState.state.equals(value)) {
+ return clusterState;
}
}
- return ClusterStateEnum.UNKNOWN;
+ return ClusterState.UNKNOWN;
}
- public static ClusterStateEnum of(String name) {
- for (ClusterStateEnum clusterStateEnum : values()) {
- if (clusterStateEnum.name().equals(name)) {
- return clusterStateEnum;
+ public static ClusterState of(String name) {
+ for (ClusterState clusterState : values()) {
+ if (clusterState.name().equals(name)) {
+ return clusterState;
}
}
- return ClusterStateEnum.UNKNOWN;
+ return ClusterState.UNKNOWN;
}
public Integer getState() {
return state;
}
- public static boolean isRunning(ClusterStateEnum state) {
+ public static boolean isRunning(ClusterState state) {
return RUNNING.equals(state);
}
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentModeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkDevelopmentMode.java
similarity index 80%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentModeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkDevelopmentMode.java
index 31377c1f7..38c545e91 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentModeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkDevelopmentMode.java
@@ -17,7 +17,7 @@
package org.apache.streampark.common.enums;
-public enum DevelopmentModeEnum {
+public enum FlinkDevelopmentMode {
/** custom code */
CUSTOM_CODE("Custom Code", 1),
@@ -32,15 +32,15 @@ public enum DevelopmentModeEnum {
private final Integer mode;
- DevelopmentModeEnum(String name, Integer mode) {
+ FlinkDevelopmentMode(String name, Integer mode) {
this.name = name;
this.mode = mode;
}
- public static DevelopmentModeEnum of(Integer value) {
- for (DevelopmentModeEnum developmentModeEnum : values()) {
- if (developmentModeEnum.mode.equals(value)) {
- return developmentModeEnum;
+ public static FlinkDevelopmentMode of(Integer value) {
+ for (FlinkDevelopmentMode flinkDevelopmentMode : values()) {
+ if (flinkDevelopmentMode.mode.equals(value)) {
+ return flinkDevelopmentMode;
}
}
return null;
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/ExecutionModeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkExecutionMode.java
similarity index 76%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/ExecutionModeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkExecutionMode.java
index 06bedf34b..47632bca5 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/ExecutionModeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkExecutionMode.java
@@ -21,7 +21,7 @@ import com.google.common.collect.Lists;
import java.util.List;
-public enum ExecutionModeEnum {
+public enum FlinkExecutionMode {
/** Local mode */
LOCAL(0, "local"),
@@ -48,24 +48,24 @@ public enum ExecutionModeEnum {
private final String name;
- ExecutionModeEnum(Integer mode, String name) {
+ FlinkExecutionMode(Integer mode, String name) {
this.mode = mode;
this.name = name;
}
- public static ExecutionModeEnum of(Integer value) {
- for (ExecutionModeEnum executionModeEnum : values()) {
- if (executionModeEnum.mode.equals(value)) {
- return executionModeEnum;
+ public static FlinkExecutionMode of(Integer value) {
+ for (FlinkExecutionMode mode : values()) {
+ if (mode.mode.equals(value)) {
+ return mode;
}
}
return null;
}
- public static ExecutionModeEnum of(String name) {
- for (ExecutionModeEnum executionModeEnum : values()) {
- if (executionModeEnum.name.equals(name)) {
- return executionModeEnum;
+ public static FlinkExecutionMode of(String name) {
+ for (FlinkExecutionMode mode : values()) {
+ if (mode.name.equals(name)) {
+ return mode;
}
}
return null;
@@ -79,17 +79,17 @@ public enum ExecutionModeEnum {
return name;
}
- public static boolean isYarnMode(ExecutionModeEnum mode) {
+ public static boolean isYarnMode(FlinkExecutionMode mode) {
return YARN_PER_JOB == mode || YARN_APPLICATION == mode || YARN_SESSION == mode;
}
// TODO: We'll inline this method back to the corresponding caller lines
// after dropping the yarn perjob mode.
- public static boolean isYarnPerJobOrAppMode(ExecutionModeEnum mode) {
+ public static boolean isYarnPerJobOrAppMode(FlinkExecutionMode mode) {
return YARN_PER_JOB == mode || YARN_APPLICATION == mode;
}
- public static boolean isYarnSessionMode(ExecutionModeEnum mode) {
+ public static boolean isYarnSessionMode(FlinkExecutionMode mode) {
return YARN_SESSION == mode;
}
@@ -101,7 +101,7 @@ public enum ExecutionModeEnum {
return KUBERNETES_NATIVE_SESSION == of(value);
}
- public static boolean isKubernetesMode(ExecutionModeEnum mode) {
+ public static boolean isKubernetesMode(FlinkExecutionMode mode) {
return KUBERNETES_NATIVE_SESSION == mode || KUBERNETES_NATIVE_APPLICATION == mode;
}
@@ -118,7 +118,7 @@ public enum ExecutionModeEnum {
KUBERNETES_NATIVE_SESSION.getMode(), KUBERNETES_NATIVE_APPLICATION.getMode());
}
- public static boolean isSessionMode(ExecutionModeEnum mode) {
+ public static boolean isSessionMode(FlinkExecutionMode mode) {
return KUBERNETES_NATIVE_SESSION == mode || YARN_SESSION == mode;
}
@@ -126,7 +126,7 @@ public enum ExecutionModeEnum {
return isRemoteMode(of(value));
}
- public static boolean isRemoteMode(ExecutionModeEnum mode) {
+ public static boolean isRemoteMode(FlinkExecutionMode mode) {
return REMOTE == mode;
}
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedTypeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
similarity index 85%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedTypeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
index df6504453..c87c5a3a6 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedTypeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
@@ -18,7 +18,7 @@
package org.apache.streampark.common.enums;
/** kubernetes.rest-service.exposed.type */
-public enum FlinkK8sRestExposedTypeEnum {
+public enum FlinkK8sRestExposedType {
/** LoadBalancer */
LOAD_BALANCER("LoadBalancer", 0),
@@ -33,13 +33,13 @@ public enum FlinkK8sRestExposedTypeEnum {
private final Integer type;
- FlinkK8sRestExposedTypeEnum(String name, Integer type) {
+ FlinkK8sRestExposedType(String name, Integer type) {
this.name = name;
this.type = type;
}
- public static FlinkK8sRestExposedTypeEnum of(Integer value) {
- for (FlinkK8sRestExposedTypeEnum order : values()) {
+ public static FlinkK8sRestExposedType of(Integer value) {
+ for (FlinkK8sRestExposedType order : values()) {
if (order.type.equals(value)) {
return order;
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/RestoreModeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkRestoreMode.java
similarity index 88%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/RestoreModeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkRestoreMode.java
index bb48c1d04..ae88a4216 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/RestoreModeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkRestoreMode.java
@@ -19,7 +19,7 @@ package org.apache.streampark.common.enums;
import java.util.Objects;
-public enum RestoreModeEnum {
+public enum FlinkRestoreMode {
/**
* In this mode Flink claims ownership of the snapshot and essentially treats it like a
@@ -51,7 +51,7 @@ public enum RestoreModeEnum {
return this.mode;
}
- RestoreModeEnum(int mode) {
+ FlinkRestoreMode(int mode) {
this.mode = mode;
}
@@ -59,10 +59,10 @@ public enum RestoreModeEnum {
return this.toString();
}
- public static RestoreModeEnum of(Integer value) {
- for (RestoreModeEnum restoreModeEnum : values()) {
- if (Objects.equals(restoreModeEnum.mode, value)) {
- return restoreModeEnum;
+ public static FlinkRestoreMode of(Integer value) {
+ for (FlinkRestoreMode flinkRestoreModeEnum : values()) {
+ if (Objects.equals(flinkRestoreModeEnum.mode, value)) {
+ return flinkRestoreModeEnum;
}
}
return null;
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedTypeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedType.java
similarity index 85%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedTypeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedType.java
index 8c32a55b1..6f9b8a067 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedTypeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkSqlValidationFailedType.java
@@ -17,7 +17,7 @@
package org.apache.streampark.common.enums;
-public enum FlinkSqlValidationFailedTypeEnum {
+public enum FlinkSqlValidationFailedType {
/** Basic test failed (such as null, etc.) */
VERIFY_FAILED(1),
@@ -36,12 +36,12 @@ public enum FlinkSqlValidationFailedTypeEnum {
private final int failedType;
- FlinkSqlValidationFailedTypeEnum(int failedType) {
+ FlinkSqlValidationFailedType(int failedType) {
this.failedType = failedType;
}
- public static FlinkSqlValidationFailedTypeEnum of(Integer value) {
- for (FlinkSqlValidationFailedTypeEnum type : values()) {
+ public static FlinkSqlValidationFailedType of(Integer value) {
+ for (FlinkSqlValidationFailedType type : values()) {
if (type.failedType == value) {
return type;
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrderEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrder.java
similarity index 87%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrderEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrder.java
index 0ab2a8821..e69823e27 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrderEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/ResolveOrder.java
@@ -18,7 +18,7 @@
package org.apache.streampark.common.enums;
/** classloader.resolve-order */
-public enum ResolveOrderEnum {
+public enum ResolveOrder {
/** parent-first */
PARENT_FIRST("parent-first", 0),
@@ -30,13 +30,13 @@ public enum ResolveOrderEnum {
private final Integer order;
- ResolveOrderEnum(String name, Integer order) {
+ ResolveOrder(String name, Integer order) {
this.name = name;
this.order = order;
}
- public static ResolveOrderEnum of(Integer value) {
- for (ResolveOrderEnum order : values()) {
+ public static ResolveOrder of(Integer value) {
+ for (ResolveOrder order : values()) {
if (order.order.equals(value)) {
return order;
}
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/SemanticEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/Semantic.java
similarity index 85%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/SemanticEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/Semantic.java
index 9f5ab9f47..f405f2818 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/SemanticEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/Semantic.java
@@ -18,7 +18,7 @@
package org.apache.streampark.common.enums;
/** Flink consistency semantics */
-public enum SemanticEnum {
+public enum Semantic {
/**
* Ensure that the counting results obtained after a fault are consistent with the correct values.
@@ -31,10 +31,10 @@ public enum SemanticEnum {
/** After the fault occurs, the counting results may be lost. */
NONE;
- public static SemanticEnum of(String name) {
- for (SemanticEnum semanticEnum : SemanticEnum.values()) {
- if (name.equals(semanticEnum.name())) {
- return semanticEnum;
+ public static Semantic of(String name) {
+ for (Semantic semantic : Semantic.values()) {
+ if (name.equals(semantic.name())) {
+ return semantic;
}
}
return null;
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/StorageTypeEnum.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/StorageType.java
similarity index 88%
rename from streampark-common/src/main/java/org/apache/streampark/common/enums/StorageTypeEnum.java
rename to streampark-common/src/main/java/org/apache/streampark/common/enums/StorageType.java
index 8d49a5f98..11be2cbec 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/StorageTypeEnum.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/StorageType.java
@@ -19,7 +19,7 @@ package org.apache.streampark.common.enums;
import org.apache.commons.lang3.StringUtils;
-public enum StorageTypeEnum {
+public enum StorageType {
/** hdfs */
HDFS("hdfs"),
@@ -29,7 +29,7 @@ public enum StorageTypeEnum {
private final String type;
- StorageTypeEnum(String type) {
+ StorageType(String type) {
this.type = type;
}
@@ -37,11 +37,11 @@ public enum StorageTypeEnum {
return type;
}
- public static StorageTypeEnum of(String identifier) {
+ public static StorageType of(String identifier) {
if (StringUtils.isBlank(identifier)) {
return LFS;
}
- for (StorageTypeEnum type : values()) {
+ for (StorageType type : values()) {
if (type.type.equals(identifier)) {
return type;
}
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala
index d7d7ee13e..8f3e7b10c 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/conf/Workspace.scala
@@ -16,7 +16,7 @@
*/
package org.apache.streampark.common.conf
-import org.apache.streampark.common.enums.StorageTypeEnum
+import org.apache.streampark.common.enums.StorageType
import org.apache.streampark.common.util.{HdfsUtils, SystemPropertyUtils}
import org.apache.streampark.common.util.ImplicitsUtils._
@@ -24,11 +24,11 @@ import java.net.URI
object Workspace {
- def of(storageType: StorageTypeEnum): Workspace = Workspace(storageType)
+ def of(storageType: StorageType): Workspace = Workspace(storageType)
- lazy val local: Workspace = Workspace.of(StorageTypeEnum.LFS)
+ lazy val local: Workspace = Workspace.of(StorageType.LFS)
- lazy val remote: Workspace = Workspace.of(StorageTypeEnum.HDFS)
+ lazy val remote: Workspace = Workspace.of(StorageType.HDFS)
private[this] lazy val localWorkspace = local.WORKSPACE
@@ -52,7 +52,7 @@ object Workspace {
}
-case class Workspace(storageType: StorageTypeEnum) {
+case class Workspace(storageType: StorageType) {
private[this] def getConfigValue[T](option: InternalOption): T = {
val s = SystemPropertyUtils.get(option.key)
@@ -68,11 +68,11 @@ case class Workspace(storageType: StorageTypeEnum) {
lazy val WORKSPACE: String = {
storageType match {
- case StorageTypeEnum.LFS =>
+ case StorageType.LFS =>
val path: String = getConfigValue[String](CommonConfig.STREAMPARK_WORKSPACE_LOCAL)
require(path != null, "[StreamPark] streampark.workspace.local must not be null")
path
- case StorageTypeEnum.HDFS =>
+ case StorageType.HDFS =>
val path: String = getConfigValue[String](CommonConfig.STREAMPARK_WORKSPACE_REMOTE)
path match {
case p if p.isEmpty =>
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiTypeEnum.scala b/streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiType.scala
similarity index 95%
rename from streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiTypeEnum.scala
rename to streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiType.scala
index 22578a37e..b0055ad19 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiTypeEnum.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/enums/ApiType.scala
@@ -16,7 +16,7 @@
*/
package org.apache.streampark.common.enums
-object ApiTypeEnum extends Enumeration {
+object ApiType extends Enumeration {
type ApiType = Value
val JAVA, SCALA = Value
}
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerTypeEnum.scala b/streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerType.scala
similarity index 95%
rename from streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerTypeEnum.scala
rename to streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerType.scala
index 2f4e0f006..bdf93a961 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerTypeEnum.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/enums/PlannerType.scala
@@ -16,7 +16,7 @@
*/
package org.apache.streampark.common.enums
-object PlannerTypeEnum extends Enumeration {
+object PlannerType extends Enumeration {
type PlannerType = Value
val BLINK, OLD, ANY = Value
}
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/enums/TableModeEnum.scala b/streampark-common/src/main/scala/org/apache/streampark/common/enums/TableMode.scala
similarity index 95%
rename from streampark-common/src/main/scala/org/apache/streampark/common/enums/TableModeEnum.scala
rename to streampark-common/src/main/scala/org/apache/streampark/common/enums/TableMode.scala
index 88a78188c..1df8cca97 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/enums/TableModeEnum.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/enums/TableMode.scala
@@ -16,7 +16,7 @@
*/
package org.apache.streampark.common.enums
-object TableModeEnum extends Enumeration {
+object TableMode extends Enumeration {
type TableMode = Value
val BATCH, STREAMING = Value
}
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/fs/FsOperator.scala b/streampark-common/src/main/scala/org/apache/streampark/common/fs/FsOperator.scala
index d54014af4..b7565371c 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/fs/FsOperator.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/fs/FsOperator.scala
@@ -17,18 +17,18 @@
package org.apache.streampark.common.fs
-import org.apache.streampark.common.enums.StorageTypeEnum
+import org.apache.streampark.common.enums.StorageType
object FsOperator {
- lazy val lfs: FsOperator = FsOperator.of(StorageTypeEnum.LFS)
+ lazy val lfs: FsOperator = FsOperator.of(StorageType.LFS)
- lazy val hdfs: FsOperator = FsOperator.of(StorageTypeEnum.HDFS)
+ lazy val hdfs: FsOperator = FsOperator.of(StorageType.HDFS)
- def of(storageType: StorageTypeEnum): FsOperator = {
+ def of(storageType: StorageType): FsOperator = {
storageType match {
- case StorageTypeEnum.HDFS => HdfsOperator
- case StorageTypeEnum.LFS => LfsOperator
+ case StorageType.HDFS => HdfsOperator
+ case StorageType.LFS => LfsOperator
case _ => throw new UnsupportedOperationException(s"Unsupported storageType:$storageType")
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
index 2dde6014c..e575c7824 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
@@ -75,7 +75,7 @@ public class UploadFileTypeInterceptor implements HandlerInterceptor {
try {
Metadata metadata = new Metadata();
AutoDetectParser parser = new AutoDetectParser();
- parser.parse(stream, new DefaultHandler(), metadata, new ParseContext());
+ parser.parse(input, new DefaultHandler(), metadata, new ParseContext());
String mimeType = metadata.get(HttpHeaders.CONTENT_TYPE);
return contentType.contains("text/x-python")
&& MediaType.TEXT_PLAIN.toString().equals(mimeType);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/annotation/AppUpdated.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/annotation/AppUpdated.java
index cfb2a5563..19e83e2fa 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/annotation/AppUpdated.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/annotation/AppUpdated.java
@@ -17,6 +17,7 @@
package org.apache.streampark.console.core.annotation;
+import org.apache.streampark.console.core.aspect.ConsoleAspect;
import org.apache.streampark.console.core.task.FlinkAppHttpWatcher;
import org.aspectj.lang.ProceedingJoinPoint;
@@ -30,9 +31,8 @@ import java.lang.annotation.Target;
* In the controller({@link org.apache.streampark.console.core.controller}), If some method causes
* application state update, need to add this annotation, This annotation marks which methods will
* cause the application to be updated, Will work together with {@link
- * org.apache.streampark.console.core.aspect.StreamParkAspect#appUpdated(ProceedingJoinPoint)}, The
- * final purpose will be refresh {@link FlinkAppHttpWatcher#WATCHING_APPS}, Make the state of the
- * job consistent with the database
+ * ConsoleAspect#appUpdated(ProceedingJoinPoint)}, The final purpose will be refresh {@link
+ * FlinkAppHttpWatcher#WATCHING_APPS}, Make the state of the job consistent with the database
*/
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/StreamParkAspect.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/ConsoleAspect.java
similarity index 98%
rename from streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/StreamParkAspect.java
rename to streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/ConsoleAspect.java
index a16765d80..87f85c143 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/StreamParkAspect.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/aspect/ConsoleAspect.java
@@ -56,7 +56,7 @@ import java.util.Objects;
@Slf4j
@Component
@Aspect
-public class StreamParkAspect {
+public class ConsoleAspect {
@Autowired private FlinkAppHttpWatcher flinkAppHttpWatcher;
@Autowired private CommonService commonService;
@@ -109,7 +109,7 @@ public class StreamParkAspect {
User currentUser = commonService.getCurrentUser();
ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
- boolean isAdmin = currentUser.getUserTypeEnum() == UserTypeEnum.ADMIN;
+ boolean isAdmin = currentUser.getUserType() == UserTypeEnum.ADMIN;
if (!isAdmin) {
PermissionTypeEnum permissionTypeEnum = permissionAction.type();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/AlertTemplate.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/AlertTemplate.java
index e05195637..f7fb06e64 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/AlertTemplate.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/bean/AlertTemplate.java
@@ -17,8 +17,8 @@
package org.apache.streampark.console.core.bean;
-import org.apache.streampark.common.enums.ClusterStateEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ClusterState;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.DateUtils;
import org.apache.streampark.common.util.YarnUtils;
import org.apache.streampark.console.core.entity.Application;
@@ -61,7 +61,7 @@ public class AlertTemplate implements Serializable {
return new AlertTemplateBuilder()
.setDuration(application.getStartTime(), application.getEndTime())
.setJobName(application.getJobName())
- .setLink(application.getExecutionModeEnum(), application.getAppId())
+ .setLink(application.getFlinkExecutionMode(), application.getAppId())
.setStartTime(application.getStartTime())
.setEndTime(application.getEndTime())
.setRestart(application.isNeedRestartOnFailed(), application.getRestartCount())
@@ -74,12 +74,11 @@ public class AlertTemplate implements Serializable {
.build();
}
- public static AlertTemplate of(
- Application application, CheckPointStatusEnum checkPointStatusEnum) {
+ public static AlertTemplate of(Application application, CheckPointStatusEnum statusEnum) {
return new AlertTemplateBuilder()
.setDuration(application.getStartTime(), application.getEndTime())
.setJobName(application.getJobName())
- .setLink(application.getExecutionModeEnum(), application.getAppId())
+ .setLink(application.getFlinkExecutionMode(), application.getAppId())
.setStartTime(application.getStartTime())
.setType(2)
.setCpFailureRateInterval(
@@ -91,18 +90,18 @@ public class AlertTemplate implements Serializable {
.build();
}
- public static AlertTemplate of(FlinkCluster cluster, ClusterStateEnum clusterStateEnum) {
+ public static AlertTemplate of(FlinkCluster cluster, ClusterState clusterState) {
return new AlertTemplateBuilder()
.setDuration(cluster.getStartTime(), cluster.getEndTime())
.setJobName(cluster.getClusterName())
- .setLink(cluster.getExecutionModeEnum(), cluster.getClusterId())
+ .setLink(cluster.getFlinkExecutionModeEnum(), cluster.getClusterId())
.setStartTime(cluster.getStartTime())
.setEndTime(cluster.getEndTime())
.setType(3)
- .setTitle(String.format("Notify: %s %s", cluster.getClusterName(), clusterStateEnum.name()))
+ .setTitle(String.format("Notify: %s %s", cluster.getClusterName(), clusterState.name()))
.setSubject(
- String.format("StreamPark Alert: %s %s", cluster.getClusterName(), clusterStateEnum))
- .setStatus(clusterStateEnum.name())
+ String.format("StreamPark Alert: %s %s", cluster.getClusterName(), clusterState))
+ .setStatus(clusterState.name())
.setAllJobs(cluster.getAllJobs())
.setAffectedJobs(cluster.getAffectedJobs())
.build();
@@ -187,8 +186,8 @@ public class AlertTemplate implements Serializable {
return this;
}
- public AlertTemplateBuilder setLink(ExecutionModeEnum mode, String appId) {
- if (ExecutionModeEnum.isYarnMode(mode)) {
+ public AlertTemplateBuilder setLink(FlinkExecutionMode mode, String appId) {
+ if (FlinkExecutionMode.isYarnMode(mode)) {
String format = "%s/proxy/%s/";
String url = String.format(format, YarnUtils.getRMWebAppURL(false), appId);
alertTemplate.setLink(url);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationHistoryController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationHistoryController.java
index 4a09164da..13b22edfa 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationHistoryController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationHistoryController.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.controller;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.core.service.application.ApplicationInfoService;
@@ -65,7 +65,7 @@ public class ApplicationHistoryController {
@RequiresPermissions("app:create")
public RestResponse listSessionClusterId(int executionMode) {
List<String> clusterIds;
- switch (ExecutionModeEnum.of(executionMode)) {
+ switch (FlinkExecutionMode.of(executionMode)) {
case KUBERNETES_NATIVE_SESSION:
case YARN_SESSION:
case REMOTE:
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
index 64f207214..a2b0048c2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.controller;
-import org.apache.streampark.common.enums.ClusterStateEnum;
+import org.apache.streampark.common.enums.ClusterState;
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.base.exception.InternalException;
import org.apache.streampark.console.core.bean.ResponseResult;
@@ -93,7 +93,7 @@ public class FlinkClusterController {
@Operation(summary = "Start flink cluster")
@PostMapping("start")
public RestResponse start(FlinkCluster cluster) {
- flinkClusterService.updateClusterState(cluster.getId(), ClusterStateEnum.STARTING);
+ flinkClusterService.updateClusterState(cluster.getId(), ClusterState.STARTING);
flinkClusterService.start(cluster);
return RestResponse.success();
}
@@ -102,7 +102,7 @@ public class FlinkClusterController {
@PostMapping("shutdown")
public RestResponse shutdown(FlinkCluster cluster) {
if (flinkClusterService.allowShutdownCluster(cluster)) {
- flinkClusterService.updateClusterState(cluster.getId(), ClusterStateEnum.CANCELLING);
+ flinkClusterService.updateClusterState(cluster.getId(), ClusterState.CANCELLING);
flinkClusterService.shutdown(cluster);
}
return RestResponse.success();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
index b92c3dd62..960683594 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Application.java
@@ -20,11 +20,11 @@ package org.apache.streampark.console.core.entity;
import org.apache.streampark.common.conf.ConfigConst;
import org.apache.streampark.common.conf.K8sFlinkConfig;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.ApplicationTypeEnum;
-import org.apache.streampark.common.enums.DevelopmentModeEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
-import org.apache.streampark.common.enums.FlinkK8sRestExposedTypeEnum;
-import org.apache.streampark.common.enums.StorageTypeEnum;
+import org.apache.streampark.common.enums.ApplicationType;
+import org.apache.streampark.common.enums.FlinkDevelopmentMode;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
+import org.apache.streampark.common.enums.FlinkK8sRestExposedType;
+import org.apache.streampark.common.enums.StorageType;
import org.apache.streampark.common.fs.FsOperator;
import org.apache.streampark.console.base.util.JacksonUtils;
import org.apache.streampark.console.core.bean.AppControl;
@@ -276,8 +276,8 @@ public class Application implements Serializable {
}
public void setYarnQueueByHotParams() {
- if (!(ExecutionModeEnum.YARN_APPLICATION == this.getExecutionModeEnum()
- || ExecutionModeEnum.YARN_PER_JOB == this.getExecutionModeEnum())) {
+ if (!(FlinkExecutionMode.YARN_APPLICATION == this.getFlinkExecutionMode()
+ || FlinkExecutionMode.YARN_PER_JOB == this.getFlinkExecutionMode())) {
return;
}
@@ -341,8 +341,8 @@ public class Application implements Serializable {
}
@JsonIgnore
- public DevelopmentModeEnum getDevelopmentMode() {
- return DevelopmentModeEnum.of(jobType);
+ public FlinkDevelopmentMode getDevelopmentMode() {
+ return FlinkDevelopmentMode.of(jobType);
}
@JsonIgnore
@@ -351,13 +351,13 @@ public class Application implements Serializable {
}
@JsonIgnore
- public FlinkK8sRestExposedTypeEnum getK8sRestExposedTypeEnum() {
- return FlinkK8sRestExposedTypeEnum.of(this.k8sRestExposedType);
+ public FlinkK8sRestExposedType getK8sRestExposedTypeEnum() {
+ return FlinkK8sRestExposedType.of(this.k8sRestExposedType);
}
@JsonIgnore
- public ExecutionModeEnum getExecutionModeEnum() {
- return ExecutionModeEnum.of(executionMode);
+ public FlinkExecutionMode getFlinkExecutionMode() {
+ return FlinkExecutionMode.of(executionMode);
}
public boolean cpFailedTrigger() {
@@ -398,10 +398,10 @@ public class Application implements Serializable {
return path;
}
- /** Automatically identify remoteAppHome or localAppHome based on app ExecutionModeEnum */
+ /** Automatically identify remoteAppHome or localAppHome based on app FlinkExecutionMode */
@JsonIgnore
public String getAppHome() {
- switch (this.getExecutionModeEnum()) {
+ switch (this.getFlinkExecutionMode()) {
case KUBERNETES_NATIVE_APPLICATION:
case KUBERNETES_NATIVE_SESSION:
case YARN_PER_JOB:
@@ -413,7 +413,7 @@ public class Application implements Serializable {
return getRemoteAppHome();
default:
throw new UnsupportedOperationException(
- "unsupported executionMode ".concat(getExecutionModeEnum().getName()));
+ "unsupported executionMode ".concat(getFlinkExecutionMode().getName()));
}
}
@@ -423,8 +423,8 @@ public class Application implements Serializable {
}
@JsonIgnore
- public ApplicationTypeEnum getApplicationType() {
- return ApplicationTypeEnum.of(appType);
+ public ApplicationType getApplicationType() {
+ return ApplicationType.of(appType);
}
@JsonIgnore
@@ -441,24 +441,24 @@ public class Application implements Serializable {
@JsonIgnore
public boolean isFlinkSqlJob() {
- return DevelopmentModeEnum.FLINK_SQL.getMode().equals(this.getJobType());
+ return FlinkDevelopmentMode.FLINK_SQL.getMode().equals(this.getJobType());
}
@JsonIgnore
public boolean isFlinkSqlJobOrPyFlinkJob() {
- return DevelopmentModeEnum.FLINK_SQL.getMode().equals(this.getJobType())
- || DevelopmentModeEnum.PYFLINK.getMode().equals(this.getJobType());
+ return FlinkDevelopmentMode.FLINK_SQL.getMode().equals(this.getJobType())
+ || FlinkDevelopmentMode.PYFLINK.getMode().equals(this.getJobType());
}
@JsonIgnore
public boolean isCustomCodeJob() {
- return DevelopmentModeEnum.CUSTOM_CODE.getMode().equals(this.getJobType());
+ return FlinkDevelopmentMode.CUSTOM_CODE.getMode().equals(this.getJobType());
}
@JsonIgnore
public boolean isCustomCodeOrPyFlinkJob() {
- return DevelopmentModeEnum.CUSTOM_CODE.getMode().equals(this.getJobType())
- || DevelopmentModeEnum.PYFLINK.getMode().equals(this.getJobType());
+ return FlinkDevelopmentMode.CUSTOM_CODE.getMode().equals(this.getJobType())
+ || FlinkDevelopmentMode.PYFLINK.getMode().equals(this.getJobType());
}
@JsonIgnore
@@ -474,7 +474,7 @@ public class Application implements Serializable {
}
public boolean isStreamParkJob() {
- return this.getAppType() == ApplicationTypeEnum.STREAMPARK_FLINK.getType();
+ return this.getAppType() == ApplicationType.STREAMPARK_FLINK.getType();
}
@JsonIgnore
@@ -507,21 +507,21 @@ public class Application implements Serializable {
}
@JsonIgnore
- public StorageTypeEnum getStorageType() {
+ public StorageType getStorageType() {
return getStorageType(getExecutionMode());
}
- public static StorageTypeEnum getStorageType(Integer execMode) {
- ExecutionModeEnum executionModeEnum = ExecutionModeEnum.of(execMode);
+ public static StorageType getStorageType(Integer execMode) {
+ FlinkExecutionMode executionModeEnum = FlinkExecutionMode.of(execMode);
switch (Objects.requireNonNull(executionModeEnum)) {
case YARN_APPLICATION:
- return StorageTypeEnum.HDFS;
+ return StorageType.HDFS;
case YARN_PER_JOB:
case YARN_SESSION:
case KUBERNETES_NATIVE_SESSION:
case KUBERNETES_NATIVE_APPLICATION:
case REMOTE:
- return StorageTypeEnum.LFS;
+ return StorageType.LFS;
default:
throw new UnsupportedOperationException("Unsupported ".concat(executionModeEnum.getName()));
}
@@ -559,7 +559,7 @@ public class Application implements Serializable {
if (appParam != this) {
this.hotParams = null;
}
- ExecutionModeEnum executionModeEnum = appParam.getExecutionModeEnum();
+ FlinkExecutionMode executionModeEnum = appParam.getFlinkExecutionMode();
Map<String, String> hotParams = new HashMap<>(0);
if (needFillYarnQueueLabel(executionModeEnum)) {
hotParams.putAll(YarnQueueLabelExpression.getQueueLabelMap(appParam.getYarnQueue()));
@@ -569,8 +569,8 @@ public class Application implements Serializable {
}
}
- private boolean needFillYarnQueueLabel(ExecutionModeEnum mode) {
- return ExecutionModeEnum.YARN_PER_JOB == mode || ExecutionModeEnum.YARN_APPLICATION == mode;
+ private boolean needFillYarnQueueLabel(FlinkExecutionMode mode) {
+ return FlinkExecutionMode.YARN_PER_JOB == mode || FlinkExecutionMode.YARN_APPLICATION == mode;
}
@Override
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java
index 9d966a125..873588cfe 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java
@@ -55,7 +55,7 @@ public class ApplicationBackUp {
this.configId = application.getConfigId();
this.description = application.getBackUpDescription();
this.createTime = new Date();
- switch (application.getExecutionModeEnum()) {
+ switch (application.getFlinkExecutionMode()) {
case KUBERNETES_NATIVE_APPLICATION:
case KUBERNETES_NATIVE_SESSION:
case YARN_PER_JOB:
@@ -75,7 +75,7 @@ public class ApplicationBackUp {
break;
default:
throw new UnsupportedOperationException(
- "unsupported executionMode ".concat(application.getExecutionModeEnum().getName()));
+ "unsupported executionMode ".concat(application.getFlinkExecutionMode().getName()));
}
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Effective.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Effective.java
index d5befb948..29b2744bd 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Effective.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Effective.java
@@ -45,7 +45,7 @@ public class Effective {
private Long targetId;
private Date createTime;
- private transient EffectiveTypeEnum effectiveTypeEnum;
+ private transient EffectiveTypeEnum effectiveType;
public Effective() {}
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 0c3ef209a..48f3d8b15 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
@@ -18,10 +18,10 @@
package org.apache.streampark.console.core.entity;
import org.apache.streampark.common.conf.ConfigConst;
-import org.apache.streampark.common.enums.ClusterStateEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
-import org.apache.streampark.common.enums.FlinkK8sRestExposedTypeEnum;
-import org.apache.streampark.common.enums.ResolveOrderEnum;
+import org.apache.streampark.common.enums.ClusterState;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
+import org.apache.streampark.common.enums.FlinkK8sRestExposedType;
+import org.apache.streampark.common.enums.ResolveOrder;
import org.apache.streampark.common.util.HttpClientUtils;
import org.apache.streampark.common.util.PropertiesUtils;
import org.apache.streampark.console.base.util.JacksonUtils;
@@ -118,16 +118,16 @@ public class FlinkCluster implements Serializable {
private transient Integer affectedJobs = 0;
@JsonIgnore
- public FlinkK8sRestExposedTypeEnum getK8sRestExposedTypeEnum() {
- return FlinkK8sRestExposedTypeEnum.of(this.k8sRestExposedType);
+ public FlinkK8sRestExposedType getK8sRestExposedTypeEnum() {
+ return FlinkK8sRestExposedType.of(this.k8sRestExposedType);
}
- public ExecutionModeEnum getExecutionModeEnum() {
- return ExecutionModeEnum.of(this.executionMode);
+ public FlinkExecutionMode getFlinkExecutionModeEnum() {
+ return FlinkExecutionMode.of(this.executionMode);
}
- public ClusterStateEnum getClusterStateEnum() {
- return ClusterStateEnum.of(this.clusterState);
+ public ClusterState getClusterStateEnum() {
+ return ClusterState.of(this.clusterState);
}
@JsonIgnore
@@ -137,7 +137,7 @@ public class FlinkCluster implements Serializable {
return Collections.emptyMap();
}
Map<String, Object> map = JacksonUtils.read(this.options, Map.class);
- if (ExecutionModeEnum.YARN_SESSION == getExecutionModeEnum()) {
+ if (FlinkExecutionMode.YARN_SESSION == getFlinkExecutionModeEnum()) {
map.put(ConfigConst.KEY_YARN_APP_NAME(), this.clusterName);
map.putAll(YarnQueueLabelExpression.getQueueLabelMap(yarnQueue));
}
@@ -181,9 +181,9 @@ public class FlinkCluster implements Serializable {
PropertiesUtils.extractDynamicPropertiesAsJava(this.getDynamicProperties());
map.putAll(this.getOptionMap());
map.putAll(dynamicProperties);
- ResolveOrderEnum resolveOrderEnum = ResolveOrderEnum.of(this.getResolveOrder());
- if (resolveOrderEnum != null) {
- map.put(CoreOptions.CLASSLOADER_RESOLVE_ORDER.key(), resolveOrderEnum.getName());
+ ResolveOrder resolveOrder = ResolveOrder.of(this.getResolveOrder());
+ if (resolveOrder != null) {
+ map.put(CoreOptions.CLASSLOADER_RESOLVE_ORDER.key(), resolveOrder.getName());
}
return map;
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Resource.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Resource.java
index 87903a486..2c6159dc4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Resource.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Resource.java
@@ -54,9 +54,9 @@ public class Resource implements Serializable {
/** user id of creator */
private Long creatorId;
- private ResourceTypeEnum resourceTypeEnum;
+ private ResourceTypeEnum resourceType;
- private EngineTypeEnum engineTypeEnum;
+ private EngineTypeEnum engineType;
// for flink app
private String mainClass;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
index e3f685a2f..c4cb68a27 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
@@ -22,7 +22,7 @@ import org.apache.streampark.common.conf.ConfigConst;
import org.apache.streampark.common.conf.InternalConfigHolder;
import org.apache.streampark.common.conf.InternalOption;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.StorageTypeEnum;
+import org.apache.streampark.common.enums.StorageType;
import org.apache.streampark.common.fs.FsOperator;
import org.apache.streampark.common.util.SystemPropertyUtils;
import org.apache.streampark.common.util.Utils;
@@ -53,7 +53,7 @@ import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import static org.apache.streampark.common.enums.StorageTypeEnum.LFS;
+import static org.apache.streampark.common.enums.StorageType.LFS;
@Order(1)
@Slf4j
@@ -64,7 +64,7 @@ public class EnvInitializer implements ApplicationRunner {
@Autowired private SettingService settingService;
- private final Set<StorageTypeEnum> initialized = new HashSet<>(2);
+ private final Set<StorageType> initialized = new HashSet<>(2);
private final FileFilter fileFilter = p -> !".gitkeep".equals(p.getName());
@@ -126,17 +126,17 @@ public class EnvInitializer implements ApplicationRunner {
SystemPropertyUtils.set(key, value);
}
- public synchronized void storageInitialize(StorageTypeEnum storageTypeEnum) {
+ public synchronized void storageInitialize(StorageType storageType) {
- if (initialized.contains(storageTypeEnum)) {
+ if (initialized.contains(storageType)) {
return;
}
- FsOperator fsOperator = FsOperator.of(storageTypeEnum);
- Workspace workspace = Workspace.of(storageTypeEnum);
+ FsOperator fsOperator = FsOperator.of(storageType);
+ Workspace workspace = Workspace.of(storageType);
// 1. prepare workspace dir
- if (LFS == storageTypeEnum) {
+ if (LFS == storageType) {
fsOperator.mkdirsIfNotExists(Workspace.APP_LOCAL_DIST());
}
Arrays.asList(
@@ -198,18 +198,18 @@ public class EnvInitializer implements ApplicationRunner {
FsOperator.lfs().mkdirs(localMavenRepo);
}
- initialized.add(storageTypeEnum);
+ initialized.add(storageType);
}
- public void checkFlinkEnv(StorageTypeEnum storageTypeEnum, FlinkEnv flinkEnv) throws IOException {
+ public void checkFlinkEnv(StorageType storageType, FlinkEnv flinkEnv) throws IOException {
String flinkLocalHome = flinkEnv.getFlinkHome();
if (flinkLocalHome == null) {
throw new ExceptionInInitializerError(
"[StreamPark] FLINK_HOME is undefined,Make sure that Flink is installed.");
}
- Workspace workspace = Workspace.of(storageTypeEnum);
+ Workspace workspace = Workspace.of(storageType);
String appFlink = workspace.APP_FLINK();
- FsOperator fsOperator = FsOperator.of(storageTypeEnum);
+ FsOperator fsOperator = FsOperator.of(storageType);
if (!fsOperator.exists(appFlink)) {
log.info("checkFlinkEnv, now mkdir [{}] starting ...", appFlink);
fsOperator.mkdirs(appFlink);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkClusterService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkClusterService.java
index 2c8ed7ff2..e15ba77e2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkClusterService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkClusterService.java
@@ -17,8 +17,8 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ClusterStateEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ClusterState;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.core.bean.ResponseResult;
import org.apache.streampark.console.core.entity.FlinkCluster;
@@ -49,7 +49,7 @@ public interface FlinkClusterService extends IService<FlinkCluster> {
Boolean existsByFlinkEnvId(Long id);
- List<FlinkCluster> getByExecutionModes(Collection<ExecutionModeEnum> executionModeEnums);
+ List<FlinkCluster> getByExecutionModes(Collection<FlinkExecutionMode> executionModeEnums);
- void updateClusterState(Long id, ClusterStateEnum state);
+ void updateClusterState(Long id, ClusterState state);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/YarnQueueService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/YarnQueueService.java
index eaaafd649..f6594901b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/YarnQueueService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/YarnQueueService.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.core.bean.ResponseResult;
import org.apache.streampark.console.core.entity.YarnQueue;
@@ -37,7 +37,7 @@ public interface YarnQueueService extends IService<YarnQueue> {
void deleteYarnQueue(YarnQueue yarnQueue);
- void checkQueueLabel(ExecutionModeEnum executionModeEnum, String queueLabel);
+ void checkQueueLabel(FlinkExecutionMode executionModeEnum, String queueLabel);
boolean isDefaultQueue(String queueLabel);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/ApplicationManageService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/ApplicationManageService.java
index 4ea3708e7..c26ce4b56 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/ApplicationManageService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/ApplicationManageService.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service.application;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.core.entity.Application;
@@ -148,7 +148,7 @@ public interface ApplicationManageService extends IService<Application> {
* execution modes
*/
List<Application> getByTeamIdAndExecutionModes(
- Long teamId, Collection<ExecutionModeEnum> executionModeEnums);
+ Long teamId, Collection<FlinkExecutionMode> executionModeEnums);
/**
* Retrieves a list of applications be probing or need to probe.
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java
index 7b5c63900..c44815f4e 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java
@@ -20,10 +20,10 @@ package org.apache.streampark.console.core.service.application.impl;
import org.apache.streampark.common.conf.ConfigConst;
import org.apache.streampark.common.conf.K8sFlinkConfig;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.DevelopmentModeEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
-import org.apache.streampark.common.enums.ResolveOrderEnum;
-import org.apache.streampark.common.enums.RestoreModeEnum;
+import org.apache.streampark.common.enums.FlinkDevelopmentMode;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
+import org.apache.streampark.common.enums.FlinkRestoreMode;
+import org.apache.streampark.common.enums.ResolveOrder;
import org.apache.streampark.common.fs.FsOperator;
import org.apache.streampark.common.util.CompletableFutureUtils;
import org.apache.streampark.common.util.DeflaterUtils;
@@ -266,10 +266,10 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
}
String clusterId = null;
- if (ExecutionModeEnum.isKubernetesMode(application.getExecutionMode())) {
+ if (FlinkExecutionMode.isKubernetesMode(application.getExecutionMode())) {
clusterId = application.getClusterId();
- } else if (ExecutionModeEnum.isYarnMode(application.getExecutionMode())) {
- if (ExecutionModeEnum.YARN_SESSION == application.getExecutionModeEnum()) {
+ } else if (FlinkExecutionMode.isYarnMode(application.getExecutionMode())) {
+ if (FlinkExecutionMode.YARN_SESSION == application.getFlinkExecutionMode()) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
ApiAlertException.throwIfNull(
cluster,
@@ -284,7 +284,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
Map<String, Object> properties = new HashMap<>();
- if (ExecutionModeEnum.isRemoteMode(application.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isRemoteMode(application.getFlinkExecutionMode())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
ApiAlertException.throwIfNull(
cluster,
@@ -301,7 +301,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
new CancelRequest(
application.getId(),
flinkEnv.getFlinkVersion(),
- ExecutionModeEnum.of(application.getExecutionMode()),
+ FlinkExecutionMode.of(application.getExecutionMode()),
properties,
clusterId,
application.getJobId(),
@@ -439,7 +439,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
String appConf = userJarAndAppConf.f1;
BuildResult buildResult = buildPipeline.getBuildResult();
- if (ExecutionModeEnum.YARN_APPLICATION == application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.YARN_APPLICATION == application.getFlinkExecutionMode()) {
buildResult = new ShadedBuildResponse(null, flinkUserJar, true);
}
@@ -450,10 +450,10 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
SubmitRequest submitRequest =
new SubmitRequest(
flinkEnv.getFlinkVersion(),
- ExecutionModeEnum.of(application.getExecutionMode()),
+ FlinkExecutionMode.of(application.getExecutionMode()),
getProperties(application),
flinkEnv.getFlinkConf(),
- DevelopmentModeEnum.of(application.getJobType()),
+ FlinkDevelopmentMode.of(application.getJobType()),
application.getId(),
jobId,
application.getJobName(),
@@ -462,7 +462,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
getSavePointed(appParam),
appParam.getRestoreMode() == null
? null
- : RestoreModeEnum.of(appParam.getRestoreMode()),
+ : FlinkRestoreMode.of(appParam.getRestoreMode()),
applicationArgs,
buildResult,
kubernetesSubmitParam,
@@ -540,7 +540,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
.whenComplete(
(t, e) -> {
if (!K8sFlinkConfig.isV2Enabled()
- && ExecutionModeEnum.isKubernetesApplicationMode(
+ && FlinkExecutionMode.isKubernetesApplicationMode(
application.getExecutionMode())) {
String domainName = settingService.getIngressModeDefault();
if (StringUtils.isNotBlank(domainName)) {
@@ -574,7 +574,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
}
private Tuple2<String, String> getUserJarAndAppConf(FlinkEnv flinkEnv, Application application) {
- ExecutionModeEnum executionModeEnum = application.getExecutionModeEnum();
+ FlinkExecutionMode executionModeEnum = application.getFlinkExecutionMode();
ApplicationConfig applicationConfig = configService.getEffective(application.getId());
ApiAlertException.throwIfNull(
@@ -595,7 +595,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
? null
: String.format("yaml://%s", applicationConfig.getContent());
// 3) client
- if (ExecutionModeEnum.YARN_APPLICATION == executionModeEnum) {
+ if (FlinkExecutionMode.YARN_APPLICATION == executionModeEnum) {
String clientPath = Workspace.remote().APP_CLIENT();
flinkUserJar = String.format("%s/%s", clientPath, sqlDistJar);
}
@@ -649,7 +649,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
}
}
- if (ExecutionModeEnum.YARN_APPLICATION == executionModeEnum) {
+ if (FlinkExecutionMode.YARN_APPLICATION == executionModeEnum) {
switch (application.getApplicationType()) {
case STREAMPARK_FLINK:
flinkUserJar =
@@ -682,7 +682,7 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
private Map<String, Object> getProperties(Application application) {
Map<String, Object> properties = new HashMap<>(application.getOptionMap());
- if (ExecutionModeEnum.isRemoteMode(application.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isRemoteMode(application.getFlinkExecutionMode())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
ApiAlertException.throwIfNull(
cluster,
@@ -693,8 +693,8 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
URI activeAddress = cluster.getRemoteURI();
properties.put(RestOptions.ADDRESS.key(), activeAddress.getHost());
properties.put(RestOptions.PORT.key(), activeAddress.getPort());
- } else if (ExecutionModeEnum.isYarnMode(application.getExecutionModeEnum())) {
- if (ExecutionModeEnum.YARN_SESSION == application.getExecutionModeEnum()) {
+ } else if (FlinkExecutionMode.isYarnMode(application.getFlinkExecutionMode())) {
+ if (FlinkExecutionMode.YARN_SESSION == application.getFlinkExecutionMode()) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
ApiAlertException.throwIfNull(
cluster,
@@ -713,11 +713,11 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
Optional.ofNullable(yarnLabelExpr)
.ifPresent(yLabel -> properties.put(ConfigConst.KEY_YARN_APP_NODE_LABEL(), yLabel));
}
- } else if (ExecutionModeEnum.isKubernetesMode(application.getExecutionModeEnum())) {
+ } else if (FlinkExecutionMode.isKubernetesMode(application.getFlinkExecutionMode())) {
properties.put(ConfigConst.KEY_K8S_IMAGE_PULL_POLICY(), "Always");
}
- if (ExecutionModeEnum.isKubernetesApplicationMode(application.getExecutionMode())) {
+ if (FlinkExecutionMode.isKubernetesApplicationMode(application.getExecutionMode())) {
try {
HadoopUtils.yarnClient();
properties.put(JobManagerOptions.ARCHIVE_DIR.key(), Workspace.ARCHIVES_FILE_PATH());
@@ -733,9 +733,9 @@ public class ApplicationActionServiceImpl extends ServiceImpl<ApplicationMapper,
Map<String, String> dynamicProperties =
PropertiesUtils.extractDynamicPropertiesAsJava(application.getDynamicProperties());
properties.putAll(dynamicProperties);
- ResolveOrderEnum resolveOrderEnum = ResolveOrderEnum.of(application.getResolveOrder());
- if (resolveOrderEnum != null) {
- properties.put(CoreOptions.CLASSLOADER_RESOLVE_ORDER.key(), resolveOrderEnum.getName());
+ ResolveOrder resolveOrder = ResolveOrder.of(application.getResolveOrder());
+ if (resolveOrder != null) {
+ properties.put(CoreOptions.CLASSLOADER_RESOLVE_ORDER.key(), resolveOrder.getName());
}
return properties;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java
index 1be928ea6..27b2e8b00 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java
@@ -19,7 +19,7 @@ package org.apache.streampark.console.core.service.application.impl;
import org.apache.streampark.common.conf.K8sFlinkConfig;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.fs.LfsOperator;
import org.apache.streampark.common.util.ExceptionUtils;
import org.apache.streampark.common.util.Utils;
@@ -73,7 +73,7 @@ import java.util.jar.Manifest;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-import static org.apache.streampark.common.enums.StorageTypeEnum.LFS;
+import static org.apache.streampark.common.enums.StorageType.LFS;
import static org.apache.streampark.console.core.task.FlinkK8sWatcherWrapper.Bridge.toTrackId;
import static org.apache.streampark.console.core.task.FlinkK8sWatcherWrapper.isKubernetesApp;
@@ -203,8 +203,8 @@ public class ApplicationInfoServiceImpl extends ServiceImpl<ApplicationMapper, A
envInitializer.checkFlinkEnv(application.getStorageType(), flinkEnv);
envInitializer.storageInitialize(application.getStorageType());
- if (ExecutionModeEnum.YARN_SESSION == application.getExecutionModeEnum()
- || ExecutionModeEnum.REMOTE == application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.YARN_SESSION == application.getFlinkExecutionMode()
+ || FlinkExecutionMode.REMOTE == application.getFlinkExecutionMode()) {
FlinkCluster flinkCluster = flinkClusterService.getById(application.getFlinkClusterId());
boolean conned = flinkClusterWatcher.verifyClusterConnection(flinkCluster);
if (!conned) {
@@ -322,7 +322,7 @@ public class ApplicationInfoServiceImpl extends ServiceImpl<ApplicationMapper, A
Application application = getById(id);
ApiAlertException.throwIfNull(
application, String.format("The application id=%s can't be found.", id));
- if (ExecutionModeEnum.isKubernetesMode(application.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isKubernetesMode(application.getFlinkExecutionMode())) {
CompletableFuture<String> future =
CompletableFuture.supplyAsync(
() ->
@@ -399,12 +399,12 @@ public class ApplicationInfoServiceImpl extends ServiceImpl<ApplicationMapper, A
// has stopped status
if (FlinkAppStateEnum.isEndState(app.getState())) {
// check whether jobName exists on yarn
- if (ExecutionModeEnum.isYarnMode(appParam.getExecutionMode())
+ if (FlinkExecutionMode.isYarnMode(appParam.getExecutionMode())
&& YarnUtils.isContains(appParam.getJobName())) {
return AppExistsStateEnum.IN_YARN;
}
// check whether clusterId, namespace, jobId on kubernetes
- else if (ExecutionModeEnum.isKubernetesMode(appParam.getExecutionMode())
+ else if (FlinkExecutionMode.isKubernetesMode(appParam.getExecutionMode())
&& k8SFlinkTrackMonitor.checkIsInRemoteCluster(toTrackId(appParam))) {
return AppExistsStateEnum.IN_KUBERNETES;
}
@@ -415,12 +415,12 @@ public class ApplicationInfoServiceImpl extends ServiceImpl<ApplicationMapper, A
}
// check whether jobName exists on yarn
- if (ExecutionModeEnum.isYarnMode(appParam.getExecutionMode())
+ if (FlinkExecutionMode.isYarnMode(appParam.getExecutionMode())
&& YarnUtils.isContains(appParam.getJobName())) {
return AppExistsStateEnum.IN_YARN;
}
// check whether clusterId, namespace, jobId on kubernetes
- else if (ExecutionModeEnum.isKubernetesMode(appParam.getExecutionMode())
+ else if (FlinkExecutionMode.isKubernetesMode(appParam.getExecutionMode())
&& k8SFlinkTrackMonitor.checkIsInRemoteCluster(toTrackId(appParam))) {
return AppExistsStateEnum.IN_KUBERNETES;
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationManageServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationManageServiceImpl.java
index 48d3638ed..104870327 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationManageServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationManageServiceImpl.java
@@ -19,8 +19,8 @@ package org.apache.streampark.console.core.service.application.impl;
import org.apache.streampark.common.conf.K8sFlinkConfig;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
-import org.apache.streampark.common.enums.StorageTypeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
+import org.apache.streampark.common.enums.StorageType;
import org.apache.streampark.common.fs.HdfsOperator;
import org.apache.streampark.common.util.DeflaterUtils;
import org.apache.streampark.console.base.domain.RestRequest;
@@ -203,7 +203,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
.delete(application.getWorkspace().APP_WORKSPACE().concat("/").concat(appId.toString()));
// try to delete yarn-application, and leave no trouble.
String path =
- Workspace.of(StorageTypeEnum.HDFS).APP_WORKSPACE().concat("/").concat(appId.toString());
+ Workspace.of(StorageType.HDFS).APP_WORKSPACE().concat("/").concat(appId.toString());
if (HdfsOperator.exists(path)) {
HdfsOperator.delete(path);
}
@@ -319,7 +319,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
}
if (shouldHandleK8sName(appParam)) {
- switch (appParam.getExecutionModeEnum()) {
+ switch (appParam.getFlinkExecutionMode()) {
case KUBERNETES_NATIVE_APPLICATION:
appParam.setK8sName(appParam.getClusterId());
break;
@@ -346,7 +346,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
private boolean shouldHandleK8sName(Application app) {
return K8sFlinkConfig.isV2Enabled()
- && ExecutionModeEnum.isKubernetesMode(app.getExecutionMode());
+ && FlinkExecutionMode.isKubernetesMode(app.getExecutionMode());
}
private boolean existsByJobName(String jobName) {
@@ -370,7 +370,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
newApp.setJobName(jobName);
newApp.setClusterId(
- ExecutionModeEnum.isSessionMode(oldApp.getExecutionModeEnum())
+ FlinkExecutionMode.isSessionMode(oldApp.getFlinkExecutionMode())
? oldApp.getClusterId()
: jobName);
newApp.setArgs(appParam.getArgs() != null ? appParam.getArgs() : oldApp.getArgs());
@@ -473,7 +473,8 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
}
// 2) k8s podTemplate changed.
- if (application.getBuild() && ExecutionModeEnum.isKubernetesMode(appParam.getExecutionMode())) {
+ if (application.getBuild()
+ && FlinkExecutionMode.isKubernetesMode(appParam.getExecutionMode())) {
if (ObjectUtils.trimNoEquals(
application.getK8sRestExposedType(), appParam.getK8sRestExposedType())
|| ObjectUtils.trimNoEquals(
@@ -498,8 +499,8 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
// 4) yarn application mode change
if (!application.getBuild()) {
if (!application.getExecutionMode().equals(appParam.getExecutionMode())) {
- if (ExecutionModeEnum.YARN_APPLICATION == appParam.getExecutionModeEnum()
- || ExecutionModeEnum.YARN_APPLICATION == application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.YARN_APPLICATION == appParam.getFlinkExecutionMode()
+ || FlinkExecutionMode.YARN_APPLICATION == application.getFlinkExecutionMode()) {
application.setBuild(true);
}
}
@@ -534,7 +535,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
application.setCpMaxFailureInterval(appParam.getCpMaxFailureInterval());
application.setTags(appParam.getTags());
- switch (appParam.getExecutionModeEnum()) {
+ switch (appParam.getFlinkExecutionMode()) {
case YARN_APPLICATION:
case YARN_PER_JOB:
case KUBERNETES_NATIVE_APPLICATION:
@@ -550,7 +551,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
}
if (shouldHandleK8sName(appParam)) {
- switch (appParam.getExecutionModeEnum()) {
+ switch (appParam.getFlinkExecutionMode()) {
case KUBERNETES_NATIVE_APPLICATION:
application.setK8sName(appParam.getClusterId());
break;
@@ -676,7 +677,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
@Override
public List<Application> getByTeamIdAndExecutionModes(
- Long teamId, @Nonnull Collection<ExecutionModeEnum> executionModeEnums) {
+ Long teamId, @Nonnull Collection<FlinkExecutionMode> executionModeEnums) {
return getBaseMapper()
.selectList(
new LambdaQueryWrapper<Application>()
@@ -684,7 +685,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
.in(
Application::getExecutionMode,
executionModeEnums.stream()
- .map(ExecutionModeEnum::getMode)
+ .map(FlinkExecutionMode::getMode)
.collect(Collectors.toSet())));
}
@@ -780,7 +781,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
*/
@VisibleForTesting
public boolean validateQueueIfNeeded(Application appParam) {
- yarnQueueService.checkQueueLabel(appParam.getExecutionModeEnum(), appParam.getYarnQueue());
+ yarnQueueService.checkQueueLabel(appParam.getFlinkExecutionMode(), appParam.getYarnQueue());
if (!isYarnNotDefaultQueue(appParam)) {
return true;
}
@@ -796,13 +797,13 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
*/
@VisibleForTesting
public boolean validateQueueIfNeeded(Application oldApp, Application newApp) {
- yarnQueueService.checkQueueLabel(newApp.getExecutionModeEnum(), newApp.getYarnQueue());
+ yarnQueueService.checkQueueLabel(newApp.getFlinkExecutionMode(), newApp.getYarnQueue());
if (!isYarnNotDefaultQueue(newApp)) {
return true;
}
oldApp.setYarnQueueByHotParams();
- if (ExecutionModeEnum.isYarnPerJobOrAppMode(newApp.getExecutionModeEnum())
+ if (FlinkExecutionMode.isYarnPerJobOrAppMode(newApp.getFlinkExecutionMode())
&& StringUtils.equals(oldApp.getYarnQueue(), newApp.getYarnQueue())) {
return true;
}
@@ -818,7 +819,7 @@ public class ApplicationManageServiceImpl extends ServiceImpl<ApplicationMapper,
* (empty or default), return true, false else.
*/
private boolean isYarnNotDefaultQueue(Application application) {
- return ExecutionModeEnum.isYarnPerJobOrAppMode(application.getExecutionModeEnum())
+ return FlinkExecutionMode.isYarnPerJobOrAppMode(application.getFlinkExecutionMode())
&& !yarnQueueService.isDefaultQueue(application.getYarnQueue());
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index c4716db2d..42355f408 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -20,9 +20,9 @@ package org.apache.streampark.console.core.service.impl;
import org.apache.streampark.common.conf.ConfigConst;
import org.apache.streampark.common.conf.K8sFlinkConfig;
import org.apache.streampark.common.conf.Workspace;
-import org.apache.streampark.common.enums.ApplicationTypeEnum;
-import org.apache.streampark.common.enums.DevelopmentModeEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ApplicationType;
+import org.apache.streampark.common.enums.FlinkDevelopmentMode;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.fs.FsOperator;
import org.apache.streampark.common.util.ExceptionUtils;
import org.apache.streampark.common.util.FileUtils;
@@ -439,14 +439,14 @@ public class AppBuildPipeServiceImpl
}
}
- ExecutionModeEnum executionModeEnum = app.getExecutionModeEnum();
+ FlinkExecutionMode executionModeEnum = app.getFlinkExecutionMode();
String mainClass = ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS();
switch (executionModeEnum) {
case YARN_APPLICATION:
String yarnProvidedPath = app.getAppLib();
String localWorkspace = app.getLocalAppHome().concat("/lib");
- if (DevelopmentModeEnum.CUSTOM_CODE == app.getDevelopmentMode()
- && ApplicationTypeEnum.APACHE_FLINK == app.getApplicationType()) {
+ if (FlinkDevelopmentMode.CUSTOM_CODE == app.getDevelopmentMode()
+ && ApplicationType.APACHE_FLINK == app.getApplicationType()) {
yarnProvidedPath = app.getAppHome();
localWorkspace = app.getLocalAppHome();
}
@@ -470,7 +470,7 @@ public class AppBuildPipeServiceImpl
mainClass,
flinkUserJar,
app.isCustomCodeJob(),
- app.getExecutionModeEnum(),
+ app.getFlinkExecutionMode(),
app.getDevelopmentMode(),
flinkEnv.getFlinkVersion(),
getMergedDependencyInfo(app));
@@ -483,7 +483,7 @@ public class AppBuildPipeServiceImpl
app.getLocalAppHome(),
mainClass,
flinkUserJar,
- app.getExecutionModeEnum(),
+ app.getFlinkExecutionMode(),
app.getDevelopmentMode(),
flinkEnv.getFlinkVersion(),
getMergedDependencyInfo(app),
@@ -499,7 +499,7 @@ public class AppBuildPipeServiceImpl
app.getLocalAppHome(),
mainClass,
flinkUserJar,
- app.getExecutionModeEnum(),
+ app.getFlinkExecutionMode(),
app.getDevelopmentMode(),
flinkEnv.getFlinkVersion(),
getMergedDependencyInfo(app),
@@ -522,7 +522,7 @@ public class AppBuildPipeServiceImpl
}
default:
throw new UnsupportedOperationException(
- "Unsupported Building Application for ExecutionMode: " + app.getExecutionModeEnum());
+ "Unsupported Building Application for ExecutionMode: " + app.getFlinkExecutionMode());
}
}
@@ -544,7 +544,7 @@ public class AppBuildPipeServiceImpl
return String.format("%s/%s", app.getAppHome(), app.getJar());
case FLINK_SQL:
String sqlDistJar = commonService.getSqlClientJar(flinkEnv);
- if (app.getExecutionModeEnum() == ExecutionModeEnum.YARN_APPLICATION) {
+ if (app.getFlinkExecutionMode() == FlinkExecutionMode.YARN_APPLICATION) {
String clientPath = Workspace.remote().APP_CLIENT();
return String.format("%s/%s", clientPath, sqlDistJar);
}
@@ -635,7 +635,7 @@ public class AppBuildPipeServiceImpl
resourceId -> {
Resource resource = resourceService.getById(resourceId);
- if (resource.getResourceTypeEnum() != ResourceTypeEnum.GROUP) {
+ if (resource.getResourceType() != ResourceTypeEnum.GROUP) {
mergeDependency(application, mvnArtifacts, jarLibs, resource);
} else {
try {
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 affe3c246..4e885c689 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
@@ -18,8 +18,8 @@
package org.apache.streampark.console.core.service.impl;
import org.apache.streampark.common.conf.K8sFlinkConfig;
-import org.apache.streampark.common.enums.ClusterStateEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ClusterState;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.ThreadUtils;
import org.apache.streampark.common.util.YarnUtils;
import org.apache.streampark.console.base.exception.ApiAlertException;
@@ -123,13 +123,13 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
}
// 3) Check connection
- if (ExecutionModeEnum.isRemoteMode(cluster.getExecutionModeEnum())
+ if (FlinkExecutionMode.isRemoteMode(cluster.getFlinkExecutionModeEnum())
&& !flinkClusterWatcher.verifyClusterConnection(cluster)) {
result.setMsg("The remote cluster connection failed, please check!");
result.setStatus(3);
return result;
}
- if (ExecutionModeEnum.isYarnMode(cluster.getExecutionModeEnum())
+ if (FlinkExecutionMode.isYarnMode(cluster.getFlinkExecutionModeEnum())
&& cluster.getClusterId() != null
&& !flinkClusterWatcher.verifyClusterConnection(cluster)) {
result.setMsg("The flink cluster connection failed, please check!");
@@ -152,15 +152,15 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
ApiAlertException.throwIfFalse(
successful, String.format(ERROR_CLUSTER_QUEUE_HINT, flinkCluster.getYarnQueue()));
flinkCluster.setCreateTime(new Date());
- if (ExecutionModeEnum.isRemoteMode(flinkCluster.getExecutionModeEnum())) {
- flinkCluster.setClusterState(ClusterStateEnum.RUNNING.getState());
+ if (FlinkExecutionMode.isRemoteMode(flinkCluster.getFlinkExecutionModeEnum())) {
+ flinkCluster.setClusterState(ClusterState.RUNNING.getState());
flinkCluster.setStartTime(new Date());
flinkCluster.setEndTime(null);
} else {
- flinkCluster.setClusterState(ClusterStateEnum.CREATED.getState());
+ flinkCluster.setClusterState(ClusterState.CREATED.getState());
}
boolean ret = save(flinkCluster);
- if (ret && ExecutionModeEnum.isRemoteMode(flinkCluster.getExecutionMode())) {
+ if (ret && FlinkExecutionMode.isRemoteMode(flinkCluster.getExecutionMode())) {
FlinkClusterWatcher.addWatching(flinkCluster);
}
if (shouldWatchForK8s(flinkCluster)) {
@@ -178,7 +178,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
ApiAlertException.throwIfNull(
deployResponse,
"Deploy cluster failed, unknown reason，please check you params or StreamPark error log");
- if (ExecutionModeEnum.isYarnSessionMode(flinkCluster.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isYarnSessionMode(flinkCluster.getFlinkExecutionModeEnum())) {
String address =
String.format(
"%s/proxy/%s/", YarnUtils.getRMWebAppURL(true), deployResponse.clusterId());
@@ -188,7 +188,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
flinkCluster.setAddress(deployResponse.address());
}
flinkCluster.setClusterId(deployResponse.clusterId());
- flinkCluster.setClusterState(ClusterStateEnum.RUNNING.getState());
+ flinkCluster.setClusterState(ClusterState.RUNNING.getState());
flinkCluster.setException(null);
flinkCluster.setEndTime(null);
updateById(flinkCluster);
@@ -198,7 +198,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
}
} catch (Exception e) {
log.error(e.getMessage(), e);
- flinkCluster.setClusterState(ClusterStateEnum.FAILED.getState());
+ flinkCluster.setClusterState(ClusterState.FAILED.getState());
flinkCluster.setException(e.toString());
updateById(flinkCluster);
throw new ApiDetailException(e);
@@ -215,9 +215,9 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
flinkCluster.setClusterName(paramOfCluster.getClusterName());
flinkCluster.setAlertId(paramOfCluster.getAlertId());
flinkCluster.setDescription(paramOfCluster.getDescription());
- if (ExecutionModeEnum.isRemoteMode(flinkCluster.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isRemoteMode(flinkCluster.getFlinkExecutionModeEnum())) {
flinkCluster.setAddress(paramOfCluster.getAddress());
- flinkCluster.setClusterState(ClusterStateEnum.RUNNING.getState());
+ flinkCluster.setClusterState(ClusterState.RUNNING.getState());
flinkCluster.setStartTime(new Date());
flinkCluster.setEndTime(null);
FlinkClusterWatcher.addWatching(flinkCluster);
@@ -249,7 +249,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
ShutDownResponse shutDownResponse =
shutdownInternal(flinkCluster, flinkCluster.getClusterId());
ApiAlertException.throwIfNull(shutDownResponse, "Get shutdown response failed");
- flinkCluster.setClusterState(ClusterStateEnum.CANCELED.getState());
+ flinkCluster.setClusterState(ClusterState.CANCELED.getState());
flinkCluster.setEndTime(new Date());
updateById(flinkCluster);
FlinkClusterWatcher.unWatching(flinkCluster);
@@ -305,19 +305,19 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
}
@Override
- public List<FlinkCluster> getByExecutionModes(Collection<ExecutionModeEnum> executionModeEnums) {
+ public List<FlinkCluster> getByExecutionModes(Collection<FlinkExecutionMode> executionModeEnums) {
return getBaseMapper()
.selectList(
new LambdaQueryWrapper<FlinkCluster>()
.in(
FlinkCluster::getExecutionMode,
executionModeEnums.stream()
- .map(ExecutionModeEnum::getMode)
+ .map(FlinkExecutionMode::getMode)
.collect(Collectors.toSet())));
}
@Override
- public void updateClusterState(Long id, ClusterStateEnum state) {
+ public void updateClusterState(Long id, ClusterState state) {
LambdaUpdateWrapper<FlinkCluster> updateWrapper =
new LambdaUpdateWrapper<FlinkCluster>()
.eq(FlinkCluster::getId, id)
@@ -347,10 +347,10 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
FlinkCluster flinkCluster = getById(id);
ApiAlertException.throwIfNull(flinkCluster, "Flink cluster not exist, please check.");
- if (ExecutionModeEnum.isYarnSessionMode(flinkCluster.getExecutionModeEnum())
- || ExecutionModeEnum.isKubernetesSessionMode(flinkCluster.getExecutionMode())) {
+ if (FlinkExecutionMode.isYarnSessionMode(flinkCluster.getFlinkExecutionModeEnum())
+ || FlinkExecutionMode.isKubernetesSessionMode(flinkCluster.getExecutionMode())) {
ApiAlertException.throwIfTrue(
- ClusterStateEnum.isRunning(flinkCluster.getClusterStateEnum()),
+ ClusterState.isRunning(flinkCluster.getClusterStateEnum()),
"Flink cluster is running, cannot be delete, please check.");
}
if (shouldWatchForK8s(flinkCluster)) {
@@ -372,7 +372,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
@VisibleForTesting
public boolean validateQueueIfNeeded(FlinkCluster clusterInfo) {
yarnQueueService.checkQueueLabel(
- clusterInfo.getExecutionModeEnum(), clusterInfo.getYarnQueue());
+ clusterInfo.getFlinkExecutionModeEnum(), clusterInfo.getYarnQueue());
if (!isYarnNotDefaultQueue(clusterInfo)) {
return true;
}
@@ -388,12 +388,13 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
*/
@VisibleForTesting
public boolean validateQueueIfNeeded(FlinkCluster oldCluster, FlinkCluster newCluster) {
- yarnQueueService.checkQueueLabel(newCluster.getExecutionModeEnum(), newCluster.getYarnQueue());
+ yarnQueueService.checkQueueLabel(
+ newCluster.getFlinkExecutionModeEnum(), newCluster.getYarnQueue());
if (!isYarnNotDefaultQueue(newCluster)) {
return true;
}
- if (ExecutionModeEnum.isYarnSessionMode(newCluster.getExecutionModeEnum())
+ if (FlinkExecutionMode.isYarnSessionMode(newCluster.getFlinkExecutionModeEnum())
&& StringUtils.equals(oldCluster.getYarnQueue(), newCluster.getYarnQueue())) {
return true;
}
@@ -409,7 +410,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
* default), return true, false else.
*/
private boolean isYarnNotDefaultQueue(FlinkCluster cluster) {
- return ExecutionModeEnum.isYarnSessionMode(cluster.getExecutionModeEnum())
+ return FlinkExecutionMode.isYarnSessionMode(cluster.getFlinkExecutionModeEnum())
&& !yarnQueueService.isDefaultQueue(cluster.getYarnQueue());
}
@@ -418,7 +419,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
ShutDownRequest stopRequest =
new ShutDownRequest(
flinkEnvService.getById(flinkCluster.getVersionId()).getFlinkVersion(),
- flinkCluster.getExecutionModeEnum(),
+ flinkCluster.getFlinkExecutionModeEnum(),
flinkCluster.getProperties(),
clusterId,
getKubernetesDeployDesc(flinkCluster, "shutdown"));
@@ -432,7 +433,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
DeployRequest deployRequest =
new DeployRequest(
flinkEnvService.getById(flinkCluster.getVersionId()).getFlinkVersion(),
- flinkCluster.getExecutionModeEnum(),
+ flinkCluster.getFlinkExecutionModeEnum(),
flinkCluster.getProperties(),
flinkCluster.getClusterId(),
getKubernetesDeployDesc(flinkCluster, "start"));
@@ -442,12 +443,12 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
}
private void checkActiveIfNeeded(FlinkCluster flinkCluster) {
- if (ExecutionModeEnum.isYarnSessionMode(flinkCluster.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isYarnSessionMode(flinkCluster.getFlinkExecutionModeEnum())) {
ApiAlertException.throwIfFalse(
- ClusterStateEnum.isRunning(flinkCluster.getClusterStateEnum()),
+ ClusterState.isRunning(flinkCluster.getClusterStateEnum()),
"Current cluster is not active, please check!");
if (!flinkClusterWatcher.verifyClusterConnection(flinkCluster)) {
- flinkCluster.setClusterState(ClusterStateEnum.LOST.getState());
+ flinkCluster.setClusterState(ClusterState.LOST.getState());
updateById(flinkCluster);
throw new ApiAlertException("Current cluster is not active, please check!");
}
@@ -457,7 +458,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
@Nullable
private KubernetesDeployParam getKubernetesDeployDesc(
@Nonnull FlinkCluster flinkCluster, String action) {
- ExecutionModeEnum executionModeEnum = flinkCluster.getExecutionModeEnum();
+ FlinkExecutionMode executionModeEnum = flinkCluster.getFlinkExecutionModeEnum();
switch (executionModeEnum) {
case YARN_SESSION:
break;
@@ -472,13 +473,13 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
default:
throw new ApiAlertException(
String.format(
- "The ExecutionModeEnum %s can't %s!", executionModeEnum.getName(), action));
+ "The FlinkExecutionMode %s can't %s!", executionModeEnum.getName(), action));
}
return null;
}
private boolean shouldWatchForK8s(FlinkCluster flinkCluster) {
return K8sFlinkConfig.isV2Enabled()
- && ExecutionModeEnum.isKubernetesSessionMode(flinkCluster.getExecutionMode());
+ && FlinkExecutionMode.isKubernetesSessionMode(flinkCluster.getExecutionMode());
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java
index 3a5e885ff..b09cf590e 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ResourceServiceImpl.java
@@ -131,13 +131,13 @@ public class ResourceServiceImpl extends ServiceImpl<ResourceMapper, Resource>
jars.isEmpty() && poms.isEmpty(), "Please add pom or jar resource.");
ApiAlertException.throwIfTrue(
- resource.getResourceTypeEnum() == ResourceTypeEnum.FLINK_APP && jars.isEmpty(),
+ resource.getResourceType() == ResourceTypeEnum.FLINK_APP && jars.isEmpty(),
"Please upload jar for Flink_App resource");
ApiAlertException.throwIfTrue(
jars.size() + poms.size() > 1, "Please do not add multi dependency at one time.");
- if (resource.getResourceTypeEnum() != ResourceTypeEnum.CONNECTOR) {
+ if (resource.getResourceType() != ResourceTypeEnum.CONNECTOR) {
ApiAlertException.throwIfNull(resource.getResourceName(), "The resourceName is required.");
} else {
String connector = resource.getConnector();
@@ -273,7 +273,7 @@ public class ResourceServiceImpl extends ServiceImpl<ResourceMapper, Resource>
@Override
public RestResponse checkResource(Resource resourceParam) throws JsonProcessingException {
- ResourceTypeEnum type = resourceParam.getResourceTypeEnum();
+ ResourceTypeEnum type = resourceParam.getResourceType();
Map<String, Serializable> resp = new HashMap<>(0);
resp.put("state", 0);
switch (type) {
@@ -305,7 +305,7 @@ public class ResourceServiceImpl extends ServiceImpl<ResourceMapper, Resource>
FlinkConnector connectorResource;
ApiAlertException.throwIfFalse(
- ResourceTypeEnum.CONNECTOR == resourceParam.getResourceTypeEnum(),
+ ResourceTypeEnum.CONNECTOR == resourceParam.getResourceType(),
"getConnectorId method error, resource not flink connector.");
List<File> jars;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
index d8aa71a9c..ff2384d06 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service.impl;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.CompletableFutureUtils;
import org.apache.streampark.common.util.ExceptionUtils;
import org.apache.streampark.common.util.ThreadUtils;
@@ -290,7 +290,7 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
private Map<String, Object> tryGetRestProps(Application application, FlinkCluster cluster) {
Map<String, Object> properties = new HashMap<>();
- if (ExecutionModeEnum.isRemoteMode(application.getExecutionModeEnum())) {
+ if (FlinkExecutionMode.isRemoteMode(application.getFlinkExecutionMode())) {
Utils.notNull(
cluster,
String.format(
@@ -304,10 +304,10 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
}
private String getClusterId(Application application, FlinkCluster cluster) {
- if (ExecutionModeEnum.isKubernetesMode(application.getExecutionMode())) {
+ if (FlinkExecutionMode.isKubernetesMode(application.getExecutionMode())) {
return application.getClusterId();
- } else if (ExecutionModeEnum.isYarnMode(application.getExecutionMode())) {
- if (ExecutionModeEnum.YARN_SESSION == application.getExecutionModeEnum()) {
+ } else if (FlinkExecutionMode.isYarnMode(application.getExecutionMode())) {
+ if (FlinkExecutionMode.YARN_SESSION == application.getFlinkExecutionMode()) {
Utils.notNull(
cluster,
String.format(
@@ -368,7 +368,7 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
public String getSavepointFromDeployLayer(Application application)
throws JsonProcessingException {
// At the yarn or k8s mode, then read the savepoint in flink-conf.yml in the bound flink
- if (!ExecutionModeEnum.isRemoteMode(application.getExecutionMode())) {
+ if (!FlinkExecutionMode.isRemoteMode(application.getExecutionMode())) {
FlinkEnv flinkEnv = flinkEnvService.getById(application.getVersionId());
return flinkEnv.convertFlinkYamlAsMap().get(SAVEPOINT_DIRECTORY.key());
}
@@ -496,7 +496,7 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
return new TriggerSavepointRequest(
application.getId(),
flinkEnv.getFlinkVersion(),
- application.getExecutionModeEnum(),
+ application.getFlinkExecutionMode(),
properties,
clusterId,
application.getJobId(),
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlWorkBenchServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlWorkBenchServiceImpl.java
index 79ac65ca1..f891c4449 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlWorkBenchServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlWorkBenchServiceImpl.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service.impl;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.HadoopConfigUtils;
import org.apache.streampark.console.core.entity.FlinkCluster;
import org.apache.streampark.console.core.entity.FlinkEnv;
@@ -99,7 +99,7 @@ public class SqlWorkBenchServiceImpl implements SqlWorkBenchService {
String port = String.valueOf(remoteURI.getPort());
String clusterId = flinkCluster.getClusterId();
- ExecutionModeEnum executionModeEnum = ExecutionModeEnum.of(flinkCluster.getExecutionMode());
+ FlinkExecutionMode executionModeEnum = FlinkExecutionMode.of(flinkCluster.getExecutionMode());
if (executionModeEnum == null) {
throw new IllegalArgumentException("executionMode is null");
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java
index 1eed75625..e536bdbb9 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service.impl;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.ApiAlertException;
@@ -176,8 +176,8 @@ public class YarnQueueServiceImpl extends ServiceImpl<YarnQueueMapper, YarnQueue
* @param queueLabel queueLabel expression.
*/
@Override
- public void checkQueueLabel(ExecutionModeEnum executionModeEnum, String queueLabel) {
- if (ExecutionModeEnum.isYarnMode(executionModeEnum)) {
+ public void checkQueueLabel(FlinkExecutionMode executionModeEnum, String queueLabel) {
+ if (FlinkExecutionMode.isYarnMode(executionModeEnum)) {
ApiAlertException.throwIfFalse(isValid(queueLabel, true), ERR_FORMAT_HINTS);
}
}
@@ -217,7 +217,7 @@ public class YarnQueueServiceImpl extends ServiceImpl<YarnQueueMapper, YarnQueue
public void checkNotReferencedByFlinkClusters(
@Nonnull String queueLabel, @Nonnull String operation) {
List<FlinkCluster> clustersReferenceYarnQueueLabel =
- flinkClusterService.getByExecutionModes(Sets.newHashSet(ExecutionModeEnum.YARN_SESSION))
+ flinkClusterService.getByExecutionModes(Sets.newHashSet(FlinkExecutionMode.YARN_SESSION))
.stream()
.filter(flinkCluster -> StringUtils.equals(flinkCluster.getYarnQueue(), queueLabel))
.collect(Collectors.toList());
@@ -233,7 +233,8 @@ public class YarnQueueServiceImpl extends ServiceImpl<YarnQueueMapper, YarnQueue
applicationManageService
.getByTeamIdAndExecutionModes(
teamId,
- Sets.newHashSet(ExecutionModeEnum.YARN_APPLICATION, ExecutionModeEnum.YARN_PER_JOB))
+ Sets.newHashSet(
+ FlinkExecutionMode.YARN_APPLICATION, FlinkExecutionMode.YARN_PER_JOB))
.stream()
.filter(
application -> {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkAppHttpWatcher.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkAppHttpWatcher.java
index ff334915d..2e9098306 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkAppHttpWatcher.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkAppHttpWatcher.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.task;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.HttpClientUtils;
import org.apache.streampark.common.util.ThreadUtils;
import org.apache.streampark.common.util.YarnUtils;
@@ -173,7 +173,7 @@ public class FlinkAppHttpWatcher {
new LambdaQueryWrapper<Application>()
.eq(Application::getTracking, 1)
.ne(Application::getState, FlinkAppStateEnum.LOST.getValue())
- .notIn(Application::getExecutionMode, ExecutionModeEnum.getKubernetesMode()));
+ .notIn(Application::getExecutionMode, FlinkExecutionMode.getKubernetesMode()));
applications.forEach(
(app) -> {
WATCHING_APPS.put(app.getId(), app);
@@ -244,9 +244,9 @@ public class FlinkAppHttpWatcher {
private void getStateFromFlink(Application application) throws Exception {
JobsOverview jobsOverview = httpJobsOverview(application);
Optional<JobsOverview.Job> optional;
- ExecutionModeEnum execMode = application.getExecutionModeEnum();
- if (ExecutionModeEnum.YARN_APPLICATION == execMode
- || ExecutionModeEnum.YARN_PER_JOB == execMode) {
+ FlinkExecutionMode execMode = application.getFlinkExecutionMode();
+ if (FlinkExecutionMode.YARN_APPLICATION == execMode
+ || FlinkExecutionMode.YARN_PER_JOB == execMode) {
optional =
!jobsOverview.getJobs().isEmpty()
? jobsOverview.getJobs().stream()
@@ -319,7 +319,7 @@ public class FlinkAppHttpWatcher {
// query the status from the yarn rest Api
YarnAppInfo yarnAppInfo = httpYarnAppInfo(application);
if (yarnAppInfo == null) {
- if (ExecutionModeEnum.REMOTE != application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.REMOTE != application.getFlinkExecutionMode()) {
throw new RuntimeException("FlinkAppHttpWatcher getStateFromYarn failed ");
}
} else {
@@ -356,7 +356,7 @@ public class FlinkAppHttpWatcher {
}
}
} catch (Exception e) {
- if (ExecutionModeEnum.REMOTE != application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.REMOTE != application.getFlinkExecutionMode()) {
throw new RuntimeException("FlinkAppHttpWatcher getStateFromYarn error,", e);
}
}
@@ -662,8 +662,8 @@ public class FlinkAppHttpWatcher {
private Overview httpOverview(Application application) throws IOException {
String appId = application.getAppId();
if (appId != null) {
- if (ExecutionModeEnum.YARN_APPLICATION == application.getExecutionModeEnum()
- || ExecutionModeEnum.YARN_PER_JOB == application.getExecutionModeEnum()) {
+ if (FlinkExecutionMode.YARN_APPLICATION == application.getFlinkExecutionMode()
+ || FlinkExecutionMode.YARN_PER_JOB == application.getFlinkExecutionMode()) {
String reqURL;
if (StringUtils.isBlank(application.getJobManagerUrl())) {
String format = "proxy/%s/overview";
@@ -680,8 +680,8 @@ public class FlinkAppHttpWatcher {
private JobsOverview httpJobsOverview(Application application) throws Exception {
final String flinkUrl = "jobs/overview";
- ExecutionModeEnum execMode = application.getExecutionModeEnum();
- if (ExecutionModeEnum.isYarnMode(execMode)) {
+ FlinkExecutionMode execMode = application.getFlinkExecutionMode();
+ if (FlinkExecutionMode.isYarnMode(execMode)) {
String reqURL;
if (StringUtils.isBlank(application.getJobManagerUrl())) {
String format = "proxy/%s/" + flinkUrl;
@@ -693,7 +693,7 @@ public class FlinkAppHttpWatcher {
return yarnRestRequest(reqURL, JobsOverview.class);
}
- if (application.getJobId() != null && ExecutionModeEnum.isRemoteMode(execMode)) {
+ if (application.getJobId() != null && FlinkExecutionMode.isRemoteMode(execMode)) {
return httpRemoteCluster(
application.getFlinkClusterId(),
cluster -> {
@@ -714,8 +714,8 @@ public class FlinkAppHttpWatcher {
private CheckPoints httpCheckpoints(Application application) throws Exception {
final String flinkUrl = "jobs/%s/checkpoints";
- ExecutionModeEnum execMode = application.getExecutionModeEnum();
- if (ExecutionModeEnum.isYarnMode(execMode)) {
+ FlinkExecutionMode execMode = application.getFlinkExecutionMode();
+ if (FlinkExecutionMode.isYarnMode(execMode)) {
String reqURL;
if (StringUtils.isBlank(application.getJobManagerUrl())) {
String format = "proxy/%s/" + flinkUrl;
@@ -727,7 +727,7 @@ public class FlinkAppHttpWatcher {
return yarnRestRequest(reqURL, CheckPoints.class);
}
- if (application.getJobId() != null && ExecutionModeEnum.isRemoteMode(execMode)) {
+ if (application.getJobId() != null && FlinkExecutionMode.isRemoteMode(execMode)) {
return httpRemoteCluster(
application.getFlinkClusterId(),
cluster -> {
@@ -811,7 +811,7 @@ public class FlinkAppHttpWatcher {
log.info("application with id {} is probing, don't send alert", app.getId());
return;
}
- switch (app.getExecutionModeEnum()) {
+ switch (app.getFlinkExecutionMode()) {
case YARN_APPLICATION:
case YARN_PER_JOB:
alertService.alert(app.getAlertId(), AlertTemplate.of(app, appState));
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkClusterWatcher.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkClusterWatcher.java
index 0d697319e..80e9547b2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkClusterWatcher.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkClusterWatcher.java
@@ -19,8 +19,8 @@ package org.apache.streampark.console.core.task;
import org.apache.streampark.common.conf.CommonConfig;
import org.apache.streampark.common.conf.InternalConfigHolder;
-import org.apache.streampark.common.enums.ClusterStateEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ClusterState;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.HadoopUtils;
import org.apache.streampark.common.util.HttpClientUtils;
import org.apache.streampark.common.util.ThreadUtils;
@@ -77,7 +77,7 @@ public class FlinkClusterWatcher {
/** Watcher cluster lists */
private static final Map<Long, FlinkCluster> WATCHER_CLUSTERS = new ConcurrentHashMap<>(8);
- private static final Cache<Long, ClusterStateEnum> FAILED_STATES =
+ private static final Cache<Long, ClusterState> FAILED_STATES =
Caffeine.newBuilder().expireAfterWrite(WATCHER_INTERVAL).build();
private boolean immediateWatch = false;
@@ -99,9 +99,9 @@ public class FlinkClusterWatcher {
List<FlinkCluster> flinkClusters =
flinkClusterService.list(
new LambdaQueryWrapper<FlinkCluster>()
- .eq(FlinkCluster::getClusterState, ClusterStateEnum.RUNNING.getState())
+ .eq(FlinkCluster::getClusterState, ClusterState.RUNNING.getState())
// excluding flink clusters on kubernetes
- .notIn(FlinkCluster::getExecutionMode, ExecutionModeEnum.getKubernetesMode()));
+ .notIn(FlinkCluster::getExecutionMode, FlinkExecutionMode.getKubernetesMode()));
flinkClusters.forEach(cluster -> WATCHER_CLUSTERS.put(cluster.getId(), cluster));
}
@@ -115,7 +115,7 @@ public class FlinkClusterWatcher {
(aLong, flinkCluster) ->
EXECUTOR.execute(
() -> {
- ClusterStateEnum state = getClusterState(flinkCluster);
+ ClusterState state = getClusterState(flinkCluster);
switch (state) {
case FAILED:
case LOST:
@@ -132,7 +132,7 @@ public class FlinkClusterWatcher {
}
}
- private void alert(FlinkCluster cluster, ClusterStateEnum state) {
+ private void alert(FlinkCluster cluster, ClusterState state) {
if (cluster.getAlertId() != null) {
cluster.setAllJobs(applicationInfoService.countByClusterId(cluster.getId()));
cluster.setAffectedJobs(
@@ -150,13 +150,13 @@ public class FlinkClusterWatcher {
* @param flinkCluster The FlinkCluster object representing the cluster.
* @return The ClusterState object representing the state of the cluster.
*/
- public ClusterStateEnum getClusterState(FlinkCluster flinkCluster) {
- ClusterStateEnum state = FAILED_STATES.getIfPresent(flinkCluster.getId());
+ public ClusterState getClusterState(FlinkCluster flinkCluster) {
+ ClusterState state = FAILED_STATES.getIfPresent(flinkCluster.getId());
if (state != null) {
return state;
}
state = httpClusterState(flinkCluster);
- if (ClusterStateEnum.isRunning(state)) {
+ if (ClusterState.isRunning(state)) {
FAILED_STATES.invalidate(flinkCluster.getId());
} else {
immediateWatch = true;
@@ -171,7 +171,7 @@ public class FlinkClusterWatcher {
* @param flinkCluster The FlinkCluster object representing the cluster.
* @return The ClusterState object representing the state of the cluster.
*/
- private ClusterStateEnum httpRemoteClusterState(FlinkCluster flinkCluster) {
+ private ClusterState httpRemoteClusterState(FlinkCluster flinkCluster) {
return getStateFromFlinkRestApi(flinkCluster);
}
@@ -181,9 +181,9 @@ public class FlinkClusterWatcher {
* @param flinkCluster
* @return
*/
- private ClusterStateEnum httpYarnSessionClusterState(FlinkCluster flinkCluster) {
- ClusterStateEnum state = getStateFromFlinkRestApi(flinkCluster);
- if (ClusterStateEnum.LOST == state) {
+ private ClusterState httpYarnSessionClusterState(FlinkCluster flinkCluster) {
+ ClusterState state = getStateFromFlinkRestApi(flinkCluster);
+ if (ClusterState.LOST == state) {
return getStateFromYarnRestApi(flinkCluster);
}
return state;
@@ -195,14 +195,14 @@ public class FlinkClusterWatcher {
* @param flinkCluster
* @return
*/
- private ClusterStateEnum httpClusterState(FlinkCluster flinkCluster) {
- switch (flinkCluster.getExecutionModeEnum()) {
+ private ClusterState httpClusterState(FlinkCluster flinkCluster) {
+ switch (flinkCluster.getFlinkExecutionModeEnum()) {
case REMOTE:
return httpRemoteClusterState(flinkCluster);
case YARN_SESSION:
return httpYarnSessionClusterState(flinkCluster);
default:
- return ClusterStateEnum.UNKNOWN;
+ return ClusterState.UNKNOWN;
}
}
@@ -212,7 +212,7 @@ public class FlinkClusterWatcher {
* @param flinkCluster
* @return
*/
- private ClusterStateEnum getStateFromFlinkRestApi(FlinkCluster flinkCluster) {
+ private ClusterState getStateFromFlinkRestApi(FlinkCluster flinkCluster) {
String address = flinkCluster.getAddress();
String jobManagerUrl = flinkCluster.getJobManagerUrl();
String flinkUrl =
@@ -225,11 +225,11 @@ public class FlinkClusterWatcher {
flinkUrl,
RequestConfig.custom().setConnectTimeout(5000, TimeUnit.MILLISECONDS).build());
JacksonUtils.read(res, Overview.class);
- return ClusterStateEnum.RUNNING;
+ return ClusterState.RUNNING;
} catch (Exception ignored) {
log.error("cluster id:{} get state from flink api failed", flinkCluster.getId());
}
- return ClusterStateEnum.LOST;
+ return ClusterState.LOST;
}
/**
@@ -238,12 +238,12 @@ public class FlinkClusterWatcher {
* @param flinkCluster
* @return
*/
- private ClusterStateEnum getStateFromYarnRestApi(FlinkCluster flinkCluster) {
+ private ClusterState getStateFromYarnRestApi(FlinkCluster flinkCluster) {
String yarnUrl = "ws/v1/cluster/apps/".concat(flinkCluster.getClusterId());
try {
String result = YarnUtils.restRequest(yarnUrl);
if (null == result) {
- return ClusterStateEnum.UNKNOWN;
+ return ClusterState.UNKNOWN;
}
YarnAppInfo yarnAppInfo = JacksonUtils.read(result, YarnAppInfo.class);
YarnApplicationState status = HadoopUtils.toYarnState(yarnAppInfo.getApp().getState());
@@ -251,11 +251,11 @@ public class FlinkClusterWatcher {
log.error(
"cluster id:{} final application status convert failed, invalid string ",
flinkCluster.getId());
- return ClusterStateEnum.UNKNOWN;
+ return ClusterState.UNKNOWN;
}
return yarnStateConvertClusterState(status);
} catch (Exception e) {
- return ClusterStateEnum.LOST;
+ return ClusterState.LOST;
}
}
@@ -265,7 +265,7 @@ public class FlinkClusterWatcher {
* @param flinkCluster
*/
public static void addWatching(FlinkCluster flinkCluster) {
- if (!ExecutionModeEnum.isKubernetesMode(flinkCluster.getExecutionModeEnum())
+ if (!FlinkExecutionMode.isKubernetesMode(flinkCluster.getFlinkExecutionModeEnum())
&& !WATCHER_CLUSTERS.containsKey(flinkCluster.getId())) {
log.info("add the cluster with id:{} to watcher cluster cache", flinkCluster.getId());
WATCHER_CLUSTERS.put(flinkCluster.getId(), flinkCluster);
@@ -286,10 +286,10 @@ public class FlinkClusterWatcher {
* @param state
* @return
*/
- private ClusterStateEnum yarnStateConvertClusterState(YarnApplicationState state) {
+ private ClusterState yarnStateConvertClusterState(YarnApplicationState state) {
return state == YarnApplicationState.FINISHED
- ? ClusterStateEnum.CANCELED
- : ClusterStateEnum.of(state.toString());
+ ? ClusterState.CANCELED
+ : ClusterState.of(state.toString());
}
/**
@@ -298,7 +298,7 @@ public class FlinkClusterWatcher {
* @return <code>false</code> if the connection of the cluster is invalid, <code>true</code> else.
*/
public Boolean verifyClusterConnection(FlinkCluster flinkCluster) {
- ClusterStateEnum clusterStateEnum = httpClusterState(flinkCluster);
- return ClusterStateEnum.isRunning(clusterStateEnum);
+ ClusterState clusterStateEnum = httpClusterState(flinkCluster);
+ return ClusterState.isRunning(clusterStateEnum);
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sChangeEventListener.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sChangeEventListener.java
index c24bcd1d7..0cd056a4b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sChangeEventListener.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sChangeEventListener.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.task;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.ThreadUtils;
import org.apache.streampark.console.core.bean.AlertTemplate;
import org.apache.streampark.console.core.entity.Application;
@@ -129,9 +129,9 @@ public class FlinkK8sChangeEventListener {
@Subscribe
public void subscribeMetricsChange(FlinkClusterMetricChangeEvent event) {
TrackId trackId = event.trackId();
- ExecutionModeEnum mode = FlinkK8sExecuteModeEnum.toExecutionMode(trackId.executeMode());
+ FlinkExecutionMode mode = FlinkK8sExecuteModeEnum.toExecutionMode(trackId.executeMode());
// discard session mode change
- if (ExecutionModeEnum.KUBERNETES_NATIVE_SESSION == mode) {
+ if (FlinkExecutionMode.KUBERNETES_NATIVE_SESSION == mode) {
return;
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
index ca6dd12a8..5efa209b7 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkK8sWatcherWrapper.java
@@ -18,7 +18,7 @@
package org.apache.streampark.console.core.task;
import org.apache.streampark.common.conf.K8sFlinkConfig;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.enums.FlinkAppStateEnum;
import org.apache.streampark.console.core.service.application.ApplicationManageService;
@@ -100,7 +100,7 @@ public class FlinkK8sWatcherWrapper {
queryWrapper
.eq(Application::getTracking, 1)
.ne(Application::getState, FlinkAppStateEnum.LOST.getValue())
- .in(Application::getExecutionMode, ExecutionModeEnum.getKubernetesMode());
+ .in(Application::getExecutionMode, FlinkExecutionMode.getKubernetesMode());
List<Application> k8sApplication = applicationManageService.list(queryWrapper);
if (CollectionUtils.isEmpty(k8sApplication)) {
@@ -126,7 +126,7 @@ public class FlinkK8sWatcherWrapper {
// covert Application to TrackId
public static TrackId toTrackId(@Nonnull Application app) {
- Enumeration.Value mode = FlinkK8sExecuteModeEnum.of(app.getExecutionModeEnum());
+ Enumeration.Value mode = FlinkK8sExecuteModeEnum.of(app.getFlinkExecutionMode());
if (FlinkK8sExecuteModeEnum.APPLICATION() == mode) {
return TrackId.onApplication(
app.getK8sNamespace(),
@@ -153,6 +153,6 @@ public class FlinkK8sWatcherWrapper {
if (application == null) {
return false;
}
- return ExecutionModeEnum.isKubernetesMode(application.getExecutionMode());
+ return FlinkExecutionMode.isKubernetesMode(application.getExecutionMode());
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
index 4f24d8fd8..4608157fd 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
@@ -77,7 +77,7 @@ public class UserController {
@PostMapping("post")
@RequiresPermissions("user:add")
public RestResponse addUser(@Valid User user) throws Exception {
- user.setLoginTypeEnum(LoginTypeEnum.PASSWORD);
+ user.setLoginType(LoginTypeEnum.PASSWORD);
this.userService.createUser(user);
return RestResponse.success();
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/entity/User.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/entity/User.java
index 9259fc98f..88a2f5733 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/entity/User.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/entity/User.java
@@ -63,9 +63,9 @@ public class User implements Serializable {
@Email(message = "{email}")
private String email;
- private UserTypeEnum userTypeEnum;
+ private UserTypeEnum userType;
- private LoginTypeEnum loginTypeEnum;
+ private LoginTypeEnum loginType;
@NotBlank(message = "{required}")
private String status;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/security/impl/AuthenticatorImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/security/impl/AuthenticatorImpl.java
index 9ff76aa8b..55810c7a7 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/security/impl/AuthenticatorImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/security/impl/AuthenticatorImpl.java
@@ -62,7 +62,7 @@ public class AuthenticatorImpl implements Authenticator {
if (user == null) {
throw new ApiAlertException(String.format("user [%s] does not exist", username));
}
- if (user.getLoginTypeEnum() != LoginTypeEnum.PASSWORD) {
+ if (user.getLoginType() != LoginTypeEnum.PASSWORD) {
throw new ApiAlertException(String.format("user [%s] can not login with PASSWORD", username));
}
String salt = user.getSalt();
@@ -82,9 +82,9 @@ public class AuthenticatorImpl implements Authenticator {
User user = usersService.findByName(username);
if (user != null) {
- if (user.getLoginTypeEnum() != LoginTypeEnum.LDAP) {
+ if (user.getLoginType() != LoginTypeEnum.LDAP) {
throw new ApiAlertException(
- String.format("user [%s] can only sign in with %s", username, user.getLoginTypeEnum()));
+ String.format("user [%s] can only sign in with %s", username, user.getLoginType()));
}
return user;
}
@@ -95,9 +95,9 @@ public class AuthenticatorImpl implements Authenticator {
// check if user exist
User user = usersService.findByName(username);
if (user != null) {
- if (user.getLoginTypeEnum() != LoginTypeEnum.SSO) {
+ if (user.getLoginType() != LoginTypeEnum.SSO) {
throw new ApiAlertException(
- String.format("user [%s] can only sign in with %s", username, user.getLoginTypeEnum()));
+ String.format("user [%s] can only sign in with %s", username, user.getLoginType()));
}
return user;
}
@@ -109,8 +109,8 @@ public class AuthenticatorImpl implements Authenticator {
newUser.setCreateTime(new Date());
newUser.setUsername(username);
newUser.setNickName(username);
- newUser.setLoginTypeEnum(loginTypeEnum);
- newUser.setUserTypeEnum(UserTypeEnum.USER);
+ newUser.setLoginType(loginTypeEnum);
+ newUser.setUserType(UserTypeEnum.USER);
newUser.setStatus(User.STATUS_VALID);
newUser.setSex(User.SEX_UNKNOWN);
usersService.createUser(newUser);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MenuServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MenuServiceImpl.java
index 15a38a745..699d99685 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MenuServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MenuServiceImpl.java
@@ -66,7 +66,7 @@ public class MenuServiceImpl extends ServiceImpl<MenuMapper, Menu> implements Me
new IllegalArgumentException(
String.format("The userId [%s] not found", userId)));
// Admin has the permission for all menus.
- if (UserTypeEnum.ADMIN == user.getUserTypeEnum()) {
+ if (UserTypeEnum.ADMIN == user.getUserType()) {
return this.list().stream().map(Menu::getPerms).collect(Collectors.toList());
}
return this.baseMapper.findUserPermissions(userId, teamId);
@@ -81,7 +81,7 @@ public class MenuServiceImpl extends ServiceImpl<MenuMapper, Menu> implements Me
new IllegalArgumentException(
String.format("The userId:[%s] not found", userId)));
// Admin has the permission for all menus.
- if (UserTypeEnum.ADMIN == user.getUserTypeEnum()) {
+ if (UserTypeEnum.ADMIN == user.getUserType()) {
LambdaQueryWrapper<Menu> queryWrapper =
new LambdaQueryWrapper<Menu>().eq(Menu::getType, "0").orderByAsc(Menu::getOrderNum);
return this.list(queryWrapper);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
index 676f1be80..c59a5037d 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
@@ -143,7 +143,7 @@ public class TeamServiceImpl extends ServiceImpl<TeamMapper, Team> implements Te
.orElseThrow(
() -> new ApiAlertException(String.format("The userId [%s] not found.", userId)));
// Admin has the permission for all teams.
- if (UserTypeEnum.ADMIN == user.getUserTypeEnum()) {
+ if (UserTypeEnum.ADMIN == user.getUserType()) {
return this.list();
}
return baseMapper.findUserTeams(userId);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
index 8c0a5a174..e24eb6686 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
@@ -129,7 +129,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
@Transactional(rollbackFor = Exception.class)
public RestResponse updateUser(User user) {
User existsUser = getById(user.getUserId());
- user.setLoginTypeEnum(null);
+ user.setLoginType(null);
user.setPassword(null);
user.setModifyTime(new Date());
if (needTransferResource(existsUser, user)) {
@@ -154,7 +154,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
User user = getById(userParam.getUserId());
ApiAlertException.throwIfNull(user, "User is null. Update password failed.");
ApiAlertException.throwIfFalse(
- user.getLoginTypeEnum() == LoginTypeEnum.PASSWORD,
+ user.getLoginType() == LoginTypeEnum.PASSWORD,
"Can only update password for user who sign in with PASSWORD");
String saltPassword = ShaHashUtils.encrypt(user.getSalt(), userParam.getOldPassword());
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/AlertConfigMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/AlertConfigMapper.xml
index eb962865e..935da7b96 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/AlertConfigMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/AlertConfigMapper.xml
@@ -22,7 +22,7 @@
<id column="id" jdbcType="BIGINT" property="id"/>
<result column="user_id" jdbcType="BIGINT" property="userId"/>
<result column="alert_name" jdbcType="VARCHAR" property="alertName"/>
- <result column="alert_type" jdbcType="INTEGER" property="alertTypeEnum"/>
+ <result column="alert_type" jdbcType="INTEGER" property="alertType"/>
<result column="email_params" jdbcType="VARCHAR" property="emailParams"/>
<result column="ding_talk_params" jdbcType="VARCHAR" property="dingTalkParams"/>
<result column="we_com_params" jdbcType="VARCHAR" property="weComParams"/>
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
index d0b1e9c47..f57b8efe7 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationMapper.xml
@@ -37,7 +37,7 @@
<result column="app_type" jdbcType="INTEGER" property="appType"/>
<result column="job_type" jdbcType="INTEGER" property="jobType"/>
<result column="resource_from" jdbcType="INTEGER" property="resourceFrom"/>
- <result column="execution_mode" jdbcType="INTEGER" property="executionModeEnum"/>
+ <result column="execution_mode" jdbcType="INTEGER" property="executionMode"/>
<result column="tracking" jdbcType="INTEGER" property="tracking"/>
<result column="jar" jdbcType="VARCHAR" property="jar"/>
<result column="jar_check_sum" jdbcType="VARCHAR" property="jarCheckSum"/>
@@ -56,14 +56,14 @@
<result column="restart_count" jdbcType="INTEGER" property="restartCount"/>
<result column="release" jdbcType="INTEGER" property="release"/>
<result column="build" jdbcType="BOOLEAN" property="build"/>
- <result column="resolve_order" jdbcType="INTEGER" property="resolveOrderEnum"/>
+ <result column="resolve_order" jdbcType="INTEGER" property="resolveOrder"/>
<result column="total_tm" jdbcType="INTEGER" property="totalTM"/>
<result column="total_slot" jdbcType="INTEGER" property="totalSlot"/>
<result column="available_slot" jdbcType="INTEGER" property="availableSlot"/>
<result column="total_task" jdbcType="INTEGER" property="totalTask"/>
<result column="jm_memory" jdbcType="INTEGER" property="jmMemory"/>
<result column="tm_memory" jdbcType="INTEGER" property="tmMemory"/>
- <result column="option_state" jdbcType="INTEGER" property="optionStateEnum"/>
+ <result column="option_state" jdbcType="INTEGER" property="optionState"/>
<result column="alert_id" jdbcType="BIGINT" property="alertId"/>
<result column="description" jdbcType="VARCHAR" property="description"/>
<result column="create_time" jdbcType="DATE" property="createTime"/>
@@ -146,8 +146,8 @@
#{item}
</foreach>
</if>
- <if test="application.executionModeEnum != null and application.executionModeEnum != ''">
- and t.execution_mode = #{application.executionModeEnum}
+ <if test="application.executionMode != null and application.executionMode != ''">
+ and t.execution_mode = #{application.executionMode}
</if>
<if test="application.jobName != null and application.jobName != ''">
and t.job_name like '%${application.jobName}%'
@@ -193,8 +193,8 @@
<if test="application.tracking != null">
tracking=#{application.tracking},
</if>
- <if test="application.optionStateEnum != null">
- option_state=#{application.optionStateEnum},
+ <if test="application.optionState != null">
+ option_state=#{application.optionState},
</if>
<if test="application.startTime != null">
start_time=#{application.startTime},
@@ -304,7 +304,7 @@
select cluster_id, max(create_time) as ct
from t_flink_app
where cluster_id is not null
- and execution_mode = #{executionModeEnum}
+ and execution_mode = #{executionMode}
group by cluster_id
order by ct desc
) as ci
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkClusterMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkClusterMapper.xml
index e6f3f1cf8..769fe521c 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkClusterMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkClusterMapper.xml
@@ -25,7 +25,7 @@
<result column="cluster_name" jdbcType="VARCHAR" property="clusterName"/>
<result column="options" jdbcType="LONGVARCHAR" property="options"/>
<result column="yarn_queue" jdbcType="VARCHAR" property="yarnQueue"/>
- <result column="execution_mode" jdbcType="TINYINT" property="executionModeEnum"/>
+ <result column="execution_mode" jdbcType="TINYINT" property="executionMode"/>
<result column="version_id" jdbcType="BIGINT" property="versionId"/>
<result column="k8s_namespace" jdbcType="VARCHAR" property="k8sNamespace"/>
<result column="service_account" jdbcType="VARCHAR" property="serviceAccount"/>
@@ -36,9 +36,9 @@
<result column="k8s_rest_exposed_type" jdbcType="TINYINT" property="k8sRestExposedType"/>
<result column="k8s_hadoop_integration" jdbcType="BOOLEAN" property="k8sHadoopIntegration"/>
<result column="k8s_conf" jdbcType="VARCHAR" property="k8sConf"/>
- <result column="resolve_order" jdbcType="INTEGER" property="resolveOrderEnum"/>
+ <result column="resolve_order" jdbcType="INTEGER" property="resolveOrder"/>
<result column="exception" jdbcType="LONGVARCHAR" property="exception"/>
- <result column="cluster_state" jdbcType="TINYINT" property="clusterStateEnum"/>
+ <result column="cluster_state" jdbcType="TINYINT" property="clusterState"/>
<result column="create_time" jdbcType="DATE" property="createTime"/>
<result column="start_time" jdbcType="DATE" property="startTime"/>
<result column="end_time" jdbcType="DATE" property="endTime"/>
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ProjectMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ProjectMapper.xml
index 773737927..46f733e58 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ProjectMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ProjectMapper.xml
@@ -22,7 +22,7 @@
<id column="id" jdbcType="INTEGER" property="id"/>
<result column="team_id" jdbcType="BIGINT" property="teamId"/>
<result column="name" jdbcType="VARCHAR" property="name"/>
- <result column="git_credential" jdbcType="INTEGER" property="gitCredentialEnum"/>
+ <result column="git_credential" jdbcType="INTEGER" property="gitCredential"/>
<result column="url" jdbcType="VARCHAR" property="url"/>
<result column="repository" jdbcType="INTEGER" property="repository"/>
<result column="branches" jdbcType="VARCHAR" property="branches"/>
@@ -34,7 +34,7 @@
<result column="type" jdbcType="INTEGER" property="type"/>
<result column="description" jdbcType="VARCHAR" property="description"/>
<result column="last_build" jdbcType="TIMESTAMP" property="lastBuild"/>
- <result column="build_state" jdbcType="INTEGER" property="buildStateEnum"/>
+ <result column="build_state" jdbcType="INTEGER" property="buildState"/>
<result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
<result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime"/>
</resultMap>
@@ -85,8 +85,8 @@
<if test="project.name != null and project.name != ''">
and t.name like '%${project.name}%'
</if>
- <if test="project.buildStateEnum != null">
- and t.build_state = #{project.buildStateEnum}
+ <if test="project.buildState != null">
+ and t.build_state = #{project.buildState}
</if>
</where>
</select>
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/system/UserMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/system/UserMapper.xml
index a196cb1ce..aeec2eb4f 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/system/UserMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/system/UserMapper.xml
@@ -31,7 +31,7 @@
<result column="sex" jdbcType="CHAR" property="sex"/>
<result column="description" jdbcType="VARCHAR" property="description"/>
<result column="last_team_id" jdbcType="BIGINT" property="lastTeamId"/>
- <result column="login_type" jdbcType="INTEGER" property="loginTypeEnum"/>
+ <result column="login_type" jdbcType="INTEGER" property="loginType"/>
</resultMap>
<select id="findUserDetail" resultType="org.apache.streampark.console.system.entity.User" parameterType="org.apache.streampark.console.system.entity.User">
diff --git a/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/task/FlinkK8sObserverBroker.scala b/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/task/FlinkK8sObserverBroker.scala
index f9b54d467..c12b0f41e 100644
--- a/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/task/FlinkK8sObserverBroker.scala
+++ b/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/task/FlinkK8sObserverBroker.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.console.core.task
import org.apache.streampark.common.conf.K8sFlinkConfig
-import org.apache.streampark.common.enums.{ClusterStateEnum, ExecutionModeEnum}
+import org.apache.streampark.common.enums.{ClusterState, FlinkExecutionMode}
import org.apache.streampark.common.zio.{OptionTraversableOps, PrettyStringOps}
import org.apache.streampark.common.zio.ZIOContainerSubscription.{ConcurrentMapExtension, RefMapExtension}
import org.apache.streampark.common.zio.ZIOExt.{IterableZStreamConverter, OptionZIOOps, UIOOps, ZStreamOptionEffectOps}
@@ -69,10 +69,10 @@ class FlinkK8sObserverBroker @Autowired() (
)
private val alertClusterStateList = Array(
- ClusterStateEnum.FAILED,
- ClusterStateEnum.UNKNOWN,
- ClusterStateEnum.LOST,
- ClusterStateEnum.KILLED
+ ClusterState.FAILED,
+ ClusterState.UNKNOWN,
+ ClusterState.LOST,
+ ClusterState.KILLED
)
private lazy val allDaemonEffects: Array[UIO[Unit]] = Array(
@@ -113,7 +113,7 @@ class FlinkK8sObserverBroker @Autowired() (
val fromApplicationRecords: UIO[Unit] = it
.safeFindApplication(
new LambdaQueryWrapper[Application]
- .in(Application.SFunc.EXECUTION_MODE, ExecutionModeEnum.getKubernetesMode)
+ .in(Application.SFunc.EXECUTION_MODE, FlinkExecutionMode.getKubernetesMode)
)(10)
.map(apps => apps.map(app => applicationToTrackKey(app)).filterSome.toVector)
.tap(keys => logInfo(s"Restore Flink K8s track-keys from Application records:\n${keys.prettyStr}"))
@@ -122,7 +122,7 @@ class FlinkK8sObserverBroker @Autowired() (
val fromFlinkClusterRecords: UIO[Unit] = it
.safeFindFlinkClusterRecord(
new LambdaQueryWrapper[FlinkCluster]
- .eq(FlinkCluster.SFunc.EXECUTION_MODE, ExecutionModeEnum.KUBERNETES_NATIVE_SESSION.getMode)
+ .eq(FlinkCluster.SFunc.EXECUTION_MODE, FlinkExecutionMode.KUBERNETES_NATIVE_SESSION.getMode)
)(10)
.map(clusters => clusters.map(e => TrackKey.cluster(e.getId, e.getK8sNamespace, e.getClusterId)))
.tap(keys => logInfo(s"Restore Flink K8s track-keys from FlinkCluster records:\n${keys.prettyStr}"))
diff --git a/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/utils/FlinkK8sDataTypeConverter.scala b/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/utils/FlinkK8sDataTypeConverter.scala
index 5f3fdb5e6..631698d3e 100644
--- a/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/utils/FlinkK8sDataTypeConverter.scala
+++ b/streampark-console/streampark-console-service/src/main/scala/org/apache/streampark/console/core/utils/FlinkK8sDataTypeConverter.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.console.core.utils
import org.apache.streampark.common.conf.{InternalConfigHolder, K8sFlinkConfig}
-import org.apache.streampark.common.enums.{ClusterStateEnum, ExecutionModeEnum}
+import org.apache.streampark.common.enums.{ClusterState, FlinkExecutionMode}
import org.apache.streampark.console.core.entity.{Application, FlinkCluster}
import org.apache.streampark.console.core.enums.FlinkAppStateEnum
import org.apache.streampark.console.core.service.SettingService
@@ -77,15 +77,15 @@ object FlinkK8sDataTypeConverter {
Try(FlinkAppStateEnum.valueOf(jobState.toString)).getOrElse(FlinkAppStateEnum.OTHER)
}
- /** Convert [[DeployCRStatus]] to [[ClusterStateEnum]]. */
- def k8sDeployStateToClusterState(crState: DeployCRStatus): ClusterStateEnum = {
+ /** Convert [[DeployCRStatus]] to [[ClusterState]]. */
+ def k8sDeployStateToClusterState(crState: DeployCRStatus): ClusterState = {
crState.evalState match {
- case EvalState.DEPLOYING => ClusterStateEnum.STARTING
- case EvalState.READY => ClusterStateEnum.RUNNING
- case EvalState.SUSPENDED => ClusterStateEnum.CANCELED
- case EvalState.FAILED => ClusterStateEnum.FAILED
- case EvalState.DELETED => ClusterStateEnum.KILLED
- case _ => ClusterStateEnum.UNKNOWN
+ case EvalState.DEPLOYING => ClusterState.STARTING
+ case EvalState.READY => ClusterState.RUNNING
+ case EvalState.SUSPENDED => ClusterState.CANCELED
+ case EvalState.FAILED => ClusterState.FAILED
+ case EvalState.DELETED => ClusterState.KILLED
+ case _ => ClusterState.UNKNOWN
}
}
@@ -110,7 +110,7 @@ object FlinkK8sDataTypeConverter {
val isLegal = {
flinkCluster != null &&
flinkCluster.getId != null &&
- ExecutionModeEnum.isKubernetesSessionMode(flinkCluster.getExecutionMode) &&
+ FlinkExecutionMode.isKubernetesSessionMode(flinkCluster.getExecutionMode) &&
StringUtils.isNoneBlank(flinkCluster.getClusterId) &&
StringUtils.isNoneBlank(flinkCluster.getK8sNamespace)
}
@@ -120,19 +120,19 @@ object FlinkK8sDataTypeConverter {
/** Convert [[Application]] to [[TrackKey]]. */
def applicationToTrackKey(app: Application): Option[TrackKey] = {
- import org.apache.streampark.common.enums.ExecutionModeEnum._
+ import org.apache.streampark.common.enums.FlinkExecutionMode._
val isLegal = {
app != null &&
app.getId != null &&
- ExecutionModeEnum.isKubernetesSessionMode(app.getExecutionMode) &&
+ FlinkExecutionMode.isKubernetesSessionMode(app.getExecutionMode) &&
StringUtils.isNoneBlank(app.getClusterId) &&
StringUtils.isNoneBlank(app.getK8sNamespace)
}
if (isLegal) None
else
- app.getExecutionModeEnum match {
+ app.getFlinkExecutionMode match {
case KUBERNETES_NATIVE_APPLICATION => Some(TrackKey.appJob(app.getId, app.getK8sNamespace, app.getClusterId))
case KUBERNETES_NATIVE_SESSION =>
Option(app.getK8sName) match {
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/SpringUnitTestBase.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/SpringUnitTestBase.java
index 0ade0bf3e..5e40b3122 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/SpringUnitTestBase.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/SpringUnitTestBase.java
@@ -19,7 +19,7 @@ package org.apache.streampark.console;
import org.apache.streampark.common.conf.CommonConfig;
import org.apache.streampark.common.conf.ConfigConst;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.FlinkCluster;
import org.apache.streampark.console.core.entity.YarnQueue;
@@ -90,12 +90,12 @@ public abstract class SpringUnitTestBase {
cluster.setClusterName(name);
cluster.setYarnQueue(yarnQueue);
cluster.setVersionId(versionId);
- cluster.setExecutionMode(ExecutionModeEnum.YARN_SESSION.getMode());
+ cluster.setExecutionMode(FlinkExecutionMode.YARN_SESSION.getMode());
return cluster;
}
protected Application mockYarnModeJobApp(
- Long teamId, String name, String yarnQueue, ExecutionModeEnum executionModeEnum) {
+ Long teamId, String name, String yarnQueue, FlinkExecutionMode executionModeEnum) {
Application application = new Application();
application.setYarnQueue(yarnQueue);
application.setTeamId(teamId);
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceITest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceITest.java
index 4b5be37bf..1a86093ca 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceITest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceITest.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.DeflaterUtils;
import org.apache.streampark.console.SpringIntegrationTestBase;
import org.apache.streampark.console.core.entity.Application;
@@ -102,7 +102,7 @@ class ApplicationManageServiceITest extends SpringIntegrationTestBase {
FlinkCluster flinkCluster = new FlinkCluster();
flinkCluster.setId(1L);
flinkCluster.setAddress(cluster.getFlinkJobManagerUrl());
- flinkCluster.setExecutionMode(ExecutionModeEnum.REMOTE.getMode());
+ flinkCluster.setExecutionMode(FlinkExecutionMode.REMOTE.getMode());
flinkCluster.setClusterName("docker-Cluster-1.17.1");
flinkCluster.setVersionId(1L);
flinkCluster.setUserId(100000L);
@@ -114,7 +114,7 @@ class ApplicationManageServiceITest extends SpringIntegrationTestBase {
application.setFlinkClusterId(1L);
application.setSqlId(100000L);
application.setVersionId(1L);
- application.setExecutionMode(ExecutionModeEnum.REMOTE.getMode());
+ application.setExecutionMode(FlinkExecutionMode.REMOTE.getMode());
// Avoid exceptional error.
application.setFlinkSql(
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceTest.java
index eef17730e..38b25d304 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/ApplicationManageServiceTest.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.SpringUnitTestBase;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.YarnQueue;
@@ -111,7 +111,7 @@ class ApplicationManageServiceTest extends SpringUnitTestBase {
YarnQueue yarnQueue = mockYarnQueue(targetTeamId, queueLabel);
yarnQueueService.save(yarnQueue);
Application application =
- mockYarnModeJobApp(targetTeamId, "app1", queueLabel, ExecutionModeEnum.YARN_APPLICATION);
+ mockYarnModeJobApp(targetTeamId, "app1", queueLabel, FlinkExecutionMode.YARN_APPLICATION);
assertThat(applicationServiceImpl.validateQueueIfNeeded(application)).isTrue();
// Test application without available queue
@@ -127,9 +127,9 @@ class ApplicationManageServiceTest extends SpringUnitTestBase {
// Test update for both versions in yarn-app or per-job with same yarn queue
Application app1 =
- mockYarnModeJobApp(teamId2, appName, queueLabel1, ExecutionModeEnum.YARN_APPLICATION);
+ mockYarnModeJobApp(teamId2, appName, queueLabel1, FlinkExecutionMode.YARN_APPLICATION);
Application app2 =
- mockYarnModeJobApp(teamId2, appName, queueLabel1, ExecutionModeEnum.YARN_PER_JOB);
+ mockYarnModeJobApp(teamId2, appName, queueLabel1, FlinkExecutionMode.YARN_PER_JOB);
assertThat(applicationServiceImpl.validateQueueIfNeeded(app1, app2)).isTrue();
// Test available queue
@@ -141,7 +141,7 @@ class ApplicationManageServiceTest extends SpringUnitTestBase {
assertThat(applicationServiceImpl.validateQueueIfNeeded(app1, app2)).isTrue();
// Test non-existed queue
- app1.setExecutionMode(ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION.getMode());
+ app1.setExecutionMode(FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION.getMode());
app2.setYarnQueue(nonExistedQueue);
assertThat(applicationServiceImpl.validateQueueIfNeeded(app1, app2)).isFalse();
}
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkClusterServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkClusterServiceTest.java
index b72ca2e97..eb47386da 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkClusterServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkClusterServiceTest.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.SpringUnitTestBase;
import org.apache.streampark.console.core.entity.FlinkCluster;
import org.apache.streampark.console.core.entity.YarnQueue;
@@ -95,7 +95,7 @@ class FlinkClusterServiceTest extends SpringUnitTestBase {
assertThat(clusterServiceImpl.validateQueueIfNeeded(cluster1, cluster2)).isTrue();
// Test non-existed queue
- cluster1.setExecutionMode(ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION.getMode());
+ cluster1.setExecutionMode(FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION.getMode());
cluster2.setYarnQueue(nonExistedQueue);
assertThat(clusterServiceImpl.validateQueueIfNeeded(cluster1, cluster2)).isFalse();
}
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/SavePointServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/SavePointServiceTest.java
index 20d749a48..e56d76f6e 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/SavePointServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/SavePointServiceTest.java
@@ -17,9 +17,9 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ApplicationTypeEnum;
-import org.apache.streampark.common.enums.DevelopmentModeEnum;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.ApplicationType;
+import org.apache.streampark.common.enums.FlinkDevelopmentMode;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.common.util.DeflaterUtils;
import org.apache.streampark.console.SpringUnitTestBase;
import org.apache.streampark.console.core.entity.Application;
@@ -94,17 +94,17 @@ class SavePointServiceTest extends SpringUnitTestBase {
app.setId(appId);
// Test for non-(StreamPark job Or FlinkSQL job)
- app.setAppType(ApplicationTypeEnum.APACHE_FLINK.getType());
+ app.setAppType(ApplicationType.APACHE_FLINK.getType());
assertThat(savePointServiceImpl.getSavepointFromAppCfgIfStreamParkOrSQLJob(app)).isNull();
- app.setAppType(ApplicationTypeEnum.STREAMPARK_FLINK.getType());
- app.setJobType(DevelopmentModeEnum.CUSTOM_CODE.getMode());
+ app.setAppType(ApplicationType.STREAMPARK_FLINK.getType());
+ app.setJobType(FlinkDevelopmentMode.CUSTOM_CODE.getMode());
assertThat(savePointServiceImpl.getSavepointFromAppCfgIfStreamParkOrSQLJob(app)).isNull();
// Test for (StreamPark job Or FlinkSQL job) without application config.
- app.setAppType(ApplicationTypeEnum.STREAMPARK_FLINK.getType());
+ app.setAppType(ApplicationType.STREAMPARK_FLINK.getType());
assertThat(savePointServiceImpl.getSavepointFromAppCfgIfStreamParkOrSQLJob(app)).isNull();
- app.setAppType(ApplicationTypeEnum.STREAMPARK_FLINK.getType());
- app.setJobType(DevelopmentModeEnum.CUSTOM_CODE.getMode());
+ app.setAppType(ApplicationType.STREAMPARK_FLINK.getType());
+ app.setJobType(FlinkDevelopmentMode.CUSTOM_CODE.getMode());
assertThat(savePointServiceImpl.getSavepointFromAppCfgIfStreamParkOrSQLJob(app)).isNull();
// Test for (StreamPark job Or FlinkSQL job) with application config just disabled checkpoint.
@@ -149,7 +149,7 @@ class SavePointServiceTest extends SpringUnitTestBase {
application.setId(appId);
application.setTeamId(teamId);
application.setVersionId(idOfFlinkEnv);
- application.setExecutionMode(ExecutionModeEnum.YARN_APPLICATION.getMode());
+ application.setExecutionMode(FlinkExecutionMode.YARN_APPLICATION.getMode());
applicationManageService.save(application);
FlinkEnv flinkEnv = new FlinkEnv();
@@ -169,7 +169,7 @@ class SavePointServiceTest extends SpringUnitTestBase {
Long clusterId = 1L;
// Test for it without cluster.
- application.setExecutionMode(ExecutionModeEnum.REMOTE.getMode());
+ application.setExecutionMode(FlinkExecutionMode.REMOTE.getMode());
application.setFlinkClusterId(clusterId);
assertThatThrownBy(() -> savePointServiceImpl.getSavepointFromDeployLayer(application))
.isInstanceOf(NullPointerException.class);
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/UserServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/UserServiceTest.java
index 67619d4b9..4d4932de9 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/UserServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/UserServiceTest.java
@@ -52,7 +52,7 @@ class UserServiceTest extends SpringUnitTestBase {
user.setUsername("test");
user.setNickName("test");
user.setPassword("test");
- user.setUserTypeEnum(UserTypeEnum.USER);
+ user.setUserType(UserTypeEnum.USER);
user.setStatus(User.STATUS_VALID);
userService.createUser(user);
// lock user
@@ -74,8 +74,8 @@ class UserServiceTest extends SpringUnitTestBase {
Resource resource = new Resource();
resource.setResourceName("test");
- resource.setResourceTypeEnum(ResourceTypeEnum.FLINK_APP);
- resource.setEngineTypeEnum(EngineTypeEnum.FLINK);
+ resource.setResourceType(ResourceTypeEnum.FLINK_APP);
+ resource.setEngineType(EngineTypeEnum.FLINK);
resource.setTeamId(1L);
resource.setCreatorId(user.getUserId());
resourceService.save(resource);
@@ -95,14 +95,14 @@ class UserServiceTest extends SpringUnitTestBase {
user.setUsername("test");
user.setNickName("test");
user.setPassword("test");
- user.setUserTypeEnum(UserTypeEnum.USER);
+ user.setUserType(UserTypeEnum.USER);
user.setStatus(User.STATUS_VALID);
userService.save(user);
Resource resource = new Resource();
resource.setResourceName("test");
- resource.setResourceTypeEnum(ResourceTypeEnum.FLINK_APP);
- resource.setEngineTypeEnum(EngineTypeEnum.FLINK);
+ resource.setResourceType(ResourceTypeEnum.FLINK_APP);
+ resource.setEngineType(EngineTypeEnum.FLINK);
resource.setTeamId(1L);
resource.setCreatorId(user.getUserId());
resourceService.save(resource);
@@ -116,7 +116,7 @@ class UserServiceTest extends SpringUnitTestBase {
targetUser.setUsername("test0");
targetUser.setNickName("test0");
targetUser.setPassword("test0");
- targetUser.setUserTypeEnum(UserTypeEnum.USER);
+ targetUser.setUserType(UserTypeEnum.USER);
targetUser.setStatus(User.STATUS_VALID);
userService.save(targetUser);
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/YarnQueueServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/YarnQueueServiceTest.java
index 00bec0e25..2d4eb7083 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/YarnQueueServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/YarnQueueServiceTest.java
@@ -17,7 +17,7 @@
package org.apache.streampark.console.core.service;
-import org.apache.streampark.common.enums.ExecutionModeEnum;
+import org.apache.streampark.common.enums.FlinkExecutionMode;
import org.apache.streampark.console.SpringUnitTestBase;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.ApiAlertException;
@@ -261,27 +261,27 @@ class YarnQueueServiceTest extends SpringUnitTestBase {
yarnQueueServiceImpl.checkNotReferencedByApplications(targetTeamId, queueLabel, operation);
// Test for existed applications that don't belong to the same team, not in yarn mode.
- applicationManageService.save(mockYarnModeJobApp(2L, "app1", null, ExecutionModeEnum.REMOTE));
+ applicationManageService.save(mockYarnModeJobApp(2L, "app1", null, FlinkExecutionMode.REMOTE));
yarnQueueServiceImpl.checkNotReferencedByApplications(targetTeamId, queueLabel, operation);
// Test for existed applications that don't belong to the same team, in yarn mode
applicationManageService.save(
- mockYarnModeJobApp(2L, "app2", null, ExecutionModeEnum.YARN_APPLICATION));
+ mockYarnModeJobApp(2L, "app2", null, FlinkExecutionMode.YARN_APPLICATION));
yarnQueueServiceImpl.checkNotReferencedByApplications(targetTeamId, queueLabel, operation);
// Test for existed applications that belong to the same team, but not in yarn mode.
applicationManageService.save(
- mockYarnModeJobApp(targetTeamId, "app3", null, ExecutionModeEnum.REMOTE));
+ mockYarnModeJobApp(targetTeamId, "app3", null, FlinkExecutionMode.REMOTE));
yarnQueueServiceImpl.checkNotReferencedByApplications(targetTeamId, queueLabel, operation);
// Test for existed applications that belong to the same team, but without yarn queue value.
applicationManageService.save(
- mockYarnModeJobApp(targetTeamId, "app4", null, ExecutionModeEnum.YARN_PER_JOB));
+ mockYarnModeJobApp(targetTeamId, "app4", null, FlinkExecutionMode.YARN_PER_JOB));
yarnQueueServiceImpl.checkNotReferencedByApplications(targetTeamId, queueLabel, operation);
// Test for existed applications, some apps belong to the same team, but others don't belong to.
applicationManageService.save(
- mockYarnModeJobApp(targetTeamId, "app5", queueLabel, ExecutionModeEnum.YARN_PER_JOB));
+ mockYarnModeJobApp(targetTeamId, "app5", queueLabel, FlinkExecutionMode.YARN_PER_JOB));
assertThatThrownBy(
() ->
yarnQueueServiceImpl.checkNotReferencedByApplications(
diff --git a/streampark-console/streampark-console-webapp/src/views/base/login/LoginForm.vue b/streampark-console/streampark-console-webapp/src/views/base/login/LoginForm.vue
index 1e37eb5f1..a2cd069a3 100644
--- a/streampark-console/streampark-console-webapp/src/views/base/login/LoginForm.vue
+++ b/streampark-console/streampark-console-webapp/src/views/base/login/LoginForm.vue
@@ -121,7 +121,7 @@
const loading = ref(false);
const userId = ref('');
const modelVisible = ref(false);
- const loginTypeEnum = ref(LoginTypeEnum.PASSWORD);
+ const loginType = ref(LoginTypeEnum.PASSWORD);
const enableSSO = ref(false);
const enableLDAP = ref(false);
@@ -133,7 +133,7 @@
const loginText = computed(() => {
const localText = t('sys.login.loginButton');
const ldapText = t('sys.login.ldapTip');
- if (loginTypeEnum.value === LoginTypeEnum.PASSWORD) {
+ if (loginType.value === LoginTypeEnum.PASSWORD) {
return { buttonText: localText, linkText: t('sys.login.ldapTip') };
}
return { buttonText: ldapText, linkText: t('sys.login.passwordTip') };
@@ -158,7 +158,7 @@
{
password: loginFormValue.password,
username: loginFormValue.account,
- loginTypeEnum: LoginTypeEnum[loginTypeEnum.value],
+ loginType: LoginTypeEnum[loginType.value],
},
'none',
);
@@ -222,11 +222,11 @@
}
function changeLoginType() {
- if (loginTypeEnum.value === LoginTypeEnum.PASSWORD) {
- loginTypeEnum.value = LoginTypeEnum.LDAP;
+ if (loginType.value === LoginTypeEnum.PASSWORD) {
+ loginType.value = LoginTypeEnum.LDAP;
return;
}
- loginTypeEnum.value = LoginTypeEnum.PASSWORD;
+ loginType.value = LoginTypeEnum.PASSWORD;
}
onMounted(() => {
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
index bb41ca4d2..39421b216 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/Add.vue
@@ -96,7 +96,7 @@
/* Initialize the form */
async function handleInitForm() {
const defaultValue = {
- resolveOrderEnum: 0,
+ resolveOrder: 0,
k8sRestExposedType: 0,
};
options.forEach((item) => {
@@ -137,18 +137,18 @@
function handleCluster(values: Recordable) {
let flinkClusterId =
- values.executionModeEnum == ExecModeEnum.YARN_SESSION
+ values.executionMode == ExecModeEnum.YARN_SESSION
? values.yarnSessionClusterId
: values.flinkClusterId;
const cluster =
unref(flinkClusters).filter((c) => {
if (flinkClusterId) {
- return c.id == flinkClusterId && c.clusterStateEnum === ClusterStateEnum.RUNNING;
+ return c.id == flinkClusterId && c.clusterState === ClusterStateEnum.RUNNING;
}
})[0] || null;
if (cluster) {
Object.assign(values, { flinkClusterId: cluster.id });
- if (values.executionModeEnum == ExecModeEnum.KUBERNETES_SESSION) {
+ if (values.executionMode == ExecModeEnum.KUBERNETES_SESSION) {
Object.assign(values, { clusterId: cluster.clusterId });
}
}
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/Detail.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/Detail.vue
index 7db24b4f5..9b43c936c 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/Detail.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/Detail.vue
@@ -155,7 +155,7 @@
ExecModeEnum.YARN_PER_JOB,
ExecModeEnum.YARN_SESSION,
ExecModeEnum.YARN_APPLICATION,
- ].includes(res.executionModeEnum)
+ ].includes(res.executionMode)
) {
handleYarn();
}
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
index a68bd976e..88691cc99 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditFlink.vue
@@ -73,7 +73,7 @@
});
/* Form reset */
- function handleReset(executionModeEnum?: string) {
+ function handleReset(executionMode?: string) {
nextTick(async () => {
let selectAlertId: string | undefined;
if (app.alertId) {
@@ -88,8 +88,8 @@
jar: app.jar,
description: app.description,
dynamicProperties: app.dynamicProperties,
- resolveOrderEnum: app.resolveOrderEnum,
- executionModeEnum: app.executionModeEnum,
+ resolveOrder: app.resolveOrder,
+ executionMode: app.executionMode,
yarnQueue: app.yarnQueue,
restartSize: app.restartSize,
checkPointFailure: {
@@ -100,7 +100,7 @@
versionId: app.versionId || null,
k8sRestExposedType: app.k8sRestExposedType,
clusterId: app.clusterId,
- [app.executionModeEnum == ExecModeEnum.YARN_SESSION
+ [app.executionMode == ExecModeEnum.YARN_SESSION
? 'yarnSessionClusterId'
: 'flinkClusterId']: app.flinkClusterId,
flinkImage: app.flinkImage,
@@ -110,8 +110,8 @@
module: app.module,
...resetParams,
};
- if (!executionModeEnum) {
- Object.assign(defaultParams, { executionModeEnum: app.executionModeEnum });
+ if (!executionMode) {
+ Object.assign(defaultParams, { executionMode: app.executionMode });
}
setFieldsValue(defaultParams);
app.args && programArgRef.value?.setContent(app.args);
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
index bb7296ab9..2225c38c6 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/EditStreamPark.vue
@@ -96,7 +96,7 @@
const [registerReviewDrawer, { openDrawer: openReviewDrawer }] = useDrawer();
/* Form reset */
- function handleReset(executionModeEnum?: string) {
+ function handleReset(executionMode?: string) {
let selectAlertId = '';
if (app.alertId) {
selectAlertId = unref(alerts).filter((t) => t.id == app.alertId)[0]?.id;
@@ -109,7 +109,7 @@
args: app.args || '',
description: app.description,
dynamicProperties: app.dynamicProperties,
- resolveOrderEnum: app.resolveOrderEnum,
+ resolveOrder: app.resolveOrder,
versionId: app.versionId || null,
teamResource: handleTeamResource(app.teamResource),
k8sRestExposedType: app.k8sRestExposedType,
@@ -122,15 +122,15 @@
cpFailureAction: app.cpFailureAction,
},
clusterId: app.clusterId,
- [app.executionModeEnum == ExecModeEnum.YARN_SESSION
+ [app.executionMode == ExecModeEnum.YARN_SESSION
? 'yarnSessionClusterId'
: 'flinkClusterId']: app.flinkClusterId,
flinkImage: app.flinkImage,
k8sNamespace: app.k8sNamespace,
...resetParams,
};
- if (!executionModeEnum) {
- Object.assign(defaultParams, { executionModeEnum: app.executionModeEnum });
+ if (!executionMode) {
+ Object.assign(defaultParams, { executionMode: app.executionMode });
}
setFieldsValue(defaultParams);
app.args && programArgRef.value?.setContent(app.args);
@@ -249,10 +249,10 @@
/* Send submission interface */
async function handleUpdateApp(params: Recordable) {
- if (params.executionModeEnum == ExecModeEnum.KUBERNETES_SESSION) {
+ if (params.executionMode == ExecModeEnum.KUBERNETES_SESSION) {
const cluster =
unref(flinkClusters).filter((c) => {
- return c.id == params.flinkClusterId && c.clusterStateEnum === ClusterStateEnum.RUNNING;
+ return c.id == params.flinkClusterId && c.clusterState === ClusterStateEnum.RUNNING;
})[0] || null;
if (cluster) {
Object.assign(params, { clusterId: cluster.clusterId });
@@ -307,7 +307,7 @@
setFieldsValue({
jobType: res.jobType,
appType: res.appType,
- executionModeEnum: res.executionModeEnum,
+ executionMode: res.executionMode,
flinkSql: res.flinkSql ? decodeByBase64(res.flinkSql) : '',
dependency: '',
module: res.module,
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
index c2bb754fa..c75d24e9d 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/View.vue
@@ -89,7 +89,7 @@
availableSlot: x.availableSlot,
},
];
- if (x['optionStateEnum'] === OptionStateEnum.NONE) {
+ if (x['optionState'] === OptionStateEnum.NONE) {
if (optionApps.starting.get(x.id)) {
if (timestamp - optionApps.starting.get(x.id) > 2000 * 2) {
optionApps.starting.delete(x.id);
@@ -149,7 +149,7 @@
// Task is running, restarting, in savePoint
if (
[AppStateEnum.RESTARTING, AppStateEnum.RUNNING].includes(app.state) ||
- app['optionStateEnum'] === OptionStateEnum.SAVEPOINTING
+ app['optionState'] === OptionStateEnum.SAVEPOINTING
) {
// yarn-per-job|yarn-session|yarn-application
handleView(app, unref(yarn));
@@ -209,7 +209,7 @@
:class="{
'cursor-pointer':
[AppStateEnum.RESTARTING, AppStateEnum.RUNNING].includes(record.state) ||
- record['optionStateEnum'] === OptionStateEnum.SAVEPOINTING,
+ record['optionState'] === OptionStateEnum.SAVEPOINTING,
}"
@click="handleJobView(record)"
>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StartApplicationModal.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StartApplicationModal.vue
index 473ff6d27..32bb55830 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StartApplicationModal.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/StartApplicationModal.vue
@@ -93,7 +93,7 @@
required: true,
},
{
- field: 'restoreModeEnum',
+ field: 'restoreMode',
label: 'restore mode',
component: 'Select',
defaultValue: RestoreModeEnum.NO_CLAIM,
@@ -127,10 +127,10 @@
const formValue = (await validate()) as Recordable;
const savePointed = formValue.startSavePointed;
const savePointPath = savePointed ? formValue['startSavePoint'] : null;
- const restoreModeEnum = savePointed ? formValue['restoreModeEnum'] : null;
+ const restoreMode = savePointed ? formValue['restoreMode'] : null;
const { data } = await fetchStart({
id: receiveData.application.id,
- restoreModeEnum,
+ restoreMode,
savePointed,
savePoint: savePointPath,
allowNonRestored: formValue.allowNonRestoredState || false,
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/project/Add.vue b/streampark-console/streampark-console-webapp/src/views/flink/project/Add.vue
index babd8effb..41c6215d9 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/project/Add.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/project/Add.vue
@@ -40,7 +40,7 @@
try {
const res = await createProject({
name: values.name,
- gitCredentialEnum: values.gitCredentialEnum,
+ gitCredential: values.gitCredential,
url: values.url,
repository: values.repository,
type: values.type,
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/project/Edit.vue b/streampark-console/streampark-console-webapp/src/views/flink/project/Edit.vue
index 3c09e87ef..99748c59f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/project/Edit.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/project/Edit.vue
@@ -37,7 +37,7 @@
const route = useRoute();
const router = useRouter();
- const buildStateEnum = ref<Nullable<number | string> | undefined>(null);
+ const buildState = ref<Nullable<number | string> | undefined>(null);
if (!route.query.id) {
router.go(-1);
}
@@ -48,9 +48,9 @@
projectResource.branches !== values.branches ||
projectResource.pom !== values.pom
) {
- buildStateEnum.value = BuildStateEnum.NEED_REBUILD;
+ buildState.value = BuildStateEnum.NEED_REBUILD;
} else {
- buildStateEnum.value = projectResource.buildStateEnum;
+ buildState.value = projectResource.buildState;
}
}
/* Update project */
@@ -60,7 +60,7 @@
const { data } = await updateProject({
id: route.query.id,
name: values.name,
- gitCredentialEnum: values.gitCredentialEnum,
+ gitCredential: values.gitCredential,
url: values.url,
repository: values.repository,
type: values.type,
@@ -71,7 +71,7 @@
pom: values.pom,
buildArgs: values.buildArgs,
description: values.description,
- buildStateEnum: buildStateEnum.value,
+ buildState: buildState.value,
});
if (data.data) {
router.go(-1);
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/project/View.vue b/streampark-console/streampark-console-webapp/src/views/flink/project/View.vue
index 337771bc4..6c2ebc2e9 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/project/View.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/project/View.vue
@@ -18,7 +18,7 @@
<PageWrapper contentFullHeight contentBackground contentClass="px-20px">
<a-card class="header" :bordered="false">
<template #extra>
- <a-radio-group v-model:value="queryParams.buildStateEnum">
+ <a-radio-group v-model:value="queryParams.buildState">
<a-radio-button
v-for="item in buttonList"
@click="handleQuery(item.key)"
@@ -106,7 +106,7 @@
const [registerLogModal, { openModal: openLogModal }] = useModal();
const buttonList = reactive(statusList);
const loading = ref(false);
- const buildStateEnum = ref('');
+ const buildState = ref('');
const searchValue = ref('');
const pageInfo = reactive({
currentPage: 1,
@@ -114,8 +114,8 @@
total: 0,
});
- const queryParams = reactive<{ buildStateEnum: string; name?: string }>({
- buildStateEnum: '',
+ const queryParams = reactive<{ buildState: string; name?: string }>({
+ buildState: '',
});
let projectDataSource = ref<Array<ProjectRecord>>([]);
@@ -146,7 +146,7 @@
const handleQuery = function (val: string | undefined) {
pageInfo.currentPage = 1;
- queryParams.buildStateEnum = val!;
+ queryParams.buildState = val!;
queryParams.name = searchValue.value;
queryData();
};
@@ -189,7 +189,7 @@
t,
searchValue,
pageInfo,
- buildStateEnum,
+ buildState,
buttonList,
handleQuery,
queryParams,
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/project/components/ListItem.vue b/streampark-console/streampark-console-webapp/src/views/flink/project/components/ListItem.vue
index e56e42a77..0368c992f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/project/components/ListItem.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/project/components/ListItem.vue
@@ -58,9 +58,9 @@
<p>{{ item.lastBuild || '--' }}</p>
</li>
<li class="list-content_item build_state">
- <span>{{ t('flink.project.form.buildStateEnum') }}</span>
+ <span>{{ t('flink.project.form.buildState') }}</span>
<p>
- <a-tag :color="buildStateEnum.color" :class="tagClass">{{ buildStateEnum.label }}</a-tag>
+ <a-tag :color="buildState.color" :class="tagClass">{{ buildState.label }}</a-tag>
</p>
</li>
</ul>
@@ -139,12 +139,12 @@
const props = defineProps({
item: { type: Object as PropType<ProjectRecord>, required: true },
});
- const needBuild = computed(() => props.item.buildStateEnum == BuildStateEnum.NEED_REBUILD);
- const isBuilding = computed(() => props.item.buildStateEnum == BuildStateEnum.BUILDING);
- const buildStateEnum = computed(() => {
- return buildStateMap[props.item.buildStateEnum] || buildStateMap[BuildStateEnum.FAILED];
+ const needBuild = computed(() => props.item.buildState == BuildStateEnum.NEED_REBUILD);
+ const isBuilding = computed(() => props.item.buildState == BuildStateEnum.BUILDING);
+ const buildState = computed(() => {
+ return buildStateMap[props.item.buildState] || buildStateMap[BuildStateEnum.FAILED];
});
- const tagClass = computed(() => buildStateEnum.value.className || '');
+ const tagClass = computed(() => buildState.value.className || '');
const svgName = computed(() => {
return (
{
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/project/components/RepositoryGroup.tsx b/streampark-console/streampark-console-webapp/src/views/flink/project/components/RepositoryGroup.tsx
index c549a3bbb..5f317752f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/project/components/RepositoryGroup.tsx
+++ b/streampark-console/streampark-console-webapp/src/views/flink/project/components/RepositoryGroup.tsx
@@ -18,7 +18,7 @@ import { defineComponent, h } from 'vue';
import type { PropType } from 'vue';
import { Form, Input, Select, Tag } from 'ant-design-vue';
export interface RepositoryProps {
- gitCredentialEnum: string | number;
+ gitCredential: string | number;
url: string;
}
import { useI18n } from '/@/hooks/web/useI18n';
@@ -61,10 +61,10 @@ export default defineComponent({
<div>
<Input.Group compact class="!flex custom-input-group">
<Select
- name="gitCredentialEnum"
+ name="gitCredential"
style="width: 80px"
placeholder={t('flink.project.form.gitCredentialPlaceholder')}
- value={props.value?.gitCredentialEnum}
+ value={props.value?.gitCredential}
onChange={(e: any) => handleProtocolChange(e)}
options={options}
/>
@@ -82,10 +82,10 @@ export default defineComponent({
<Tag color="#2db7f5" class="tag-note">
{t('flink.app.noteInfo.note')}
</Tag>
- {props.value?.gitCredentialEnum === 1 && (
+ {props.value?.gitCredential === 1 && (
<span>{t('flink.project.operationTips.httpsCredential')}</span>
)}
- {props.value?.gitCredentialEnum === 2 && (
+ {props.value?.gitCredential === 2 && (
<span>{t('flink.project.operationTips.sshCredential')}</span>
)}
</span>
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertModal.vue b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertModal.vue
index 963bf557d..422f8df56 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertModal.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertModal.vue
@@ -22,7 +22,7 @@
<script setup lang="ts" name="AlertModal">
import { ref, defineComponent, h } from 'vue';
import { omit } from 'lodash-es';
- import { alertFormSchema, alertTypeEnums } from '../index.data';
+ import { alertFormSchema, alertTypes } from '../index.data';
import { BasicModal, useModalInner } from '/@/components/Modal';
import { BasicForm, useForm } from '/@/components/Form';
import { Form, Select, Input, Divider } from 'ant-design-vue';
@@ -38,7 +38,7 @@
const { t } = useI18n();
const emit = defineEmits(['reload', 'register']);
const alertId = ref<string | null>(null);
- const alertTypeEnum = ref<string[]>([]);
+ const alertType = ref<string[]>([]);
const { Swal } = useMessage();
const userStore = useUserStore();
@@ -99,10 +99,10 @@
const [registerModal, { changeOkLoading, closeModal }] = useModalInner((data) => {
resetFields();
alertId.value = '';
- alertTypeEnum.value = [];
+ alertType.value = [];
if (data && Object.keys(data).length > 0) {
alertId.value = data.alertId;
- alertTypeEnum.value = data.alertTypeEnum;
+ alertType.value = data.alertType;
setFieldsValue(omit(data, 'alertId'));
}
});
@@ -116,7 +116,7 @@
id: alertId.value,
alertName: formValue.alertName,
userId: userStore.getUserInfo?.userId,
- alertTypeEnum: eval(formValue.alertTypeEnum.join('+')),
+ alertType: eval(formValue.alertType.join('+')),
emailParams: { contacts: formValue.alertEmail },
dingTalkParams: {
token: formValue.dingtalkToken,
@@ -209,10 +209,10 @@
:placeholder="t('setting.alarm.faultAlertType')"
allowClear
mode="multiple"
- @change="(value: string[]) => (alertTypeEnum = value)"
+ @change="(value: string[]) => (alertType = value)"
>
<SelectOption
- v-for="(v, k) in alertTypeEnums"
+ v-for="(v, k) in alertTypes"
:key="`alertType_${k}`"
:disabled="v.disabled"
:value="k"
@@ -224,7 +224,7 @@
</template>
<template #alertEmail="{ model, field }">
<!-- Alert Email -->
- <template v-if="(alertTypeEnum || []).includes('1')">
+ <template v-if="(alertType || []).includes('1')">
<Divider>
<SvgIcon name="mail" size="20" />
{{ t('setting.alarm.email') }}
@@ -254,7 +254,7 @@
</template>
</template>
- <template #alertDingURL="{ model, field }" v-if="(alertTypeEnum || []).includes('2')">
+ <template #alertDingURL="{ model, field }" v-if="(alertType || []).includes('2')">
<Divider>
<SvgIcon name="dingtalk" size="20" />
{{ t('setting.alarm.dingTalk') }}
@@ -280,14 +280,14 @@
</template>
<!-- WeChat -->
- <template #weToken="{ model, field, schema }" v-if="(alertTypeEnum || []).includes('4')">
+ <template #weToken="{ model, field, schema }" v-if="(alertType || []).includes('4')">
<Divider> <SvgIcon name="wecom" size="20" /> {{ t('setting.alarm.weChat') }} </Divider>
<FormItem :label="schema.label" :name="field" :rules="schema.rules">
<InputTextArea v-model:value="model[field]" v-bind="schema.componentProps" />
</FormItem>
</template>
- <template #alertSms="{ model, field, schema }" v-if="(alertTypeEnum || []).includes('8')">
+ <template #alertSms="{ model, field, schema }" v-if="(alertType || []).includes('8')">
<Divider> <SvgIcon name="message" size="20" /> {{ t('setting.alarm.sms') }} </Divider>
<FormItem :label="schema.label" :name="field" :rules="schema.rules">
<Input v-model:value="model[field]" v-bind="schema.componentProps" />
@@ -295,7 +295,7 @@
</template>
<!-- lark -->
- <template #larkToken="{ model, field, schema }" v-if="(alertTypeEnum || []).includes('16')">
+ <template #larkToken="{ model, field, schema }" v-if="(alertType || []).includes('16')">
<Divider> <SvgIcon name="lark" size="20" /> {{ t('setting.alarm.lark') }} </Divider>
<FormItem :label="schema.label" :name="field" :rules="schema.rules">
<Input
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertTypeInfo.vue b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertTypeInfo.vue
index fb80346bb..c76f9cc38 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertTypeInfo.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/components/AlertTypeInfo.vue
@@ -19,12 +19,12 @@
import { Descriptions, Tag } from 'ant-design-vue';
import { computed, toRefs } from 'vue';
import { SvgIcon } from '/@/components/Icon';
- import { alertTypeEnums } from '../index.data';
+ import { alertTypes } from '../index.data';
import { useI18n } from '/@/hooks/web/useI18n';
const { t } = useI18n();
const props = defineProps({
- alertTypeEnum: {
+ alertType: {
type: String,
validator: (v: string) => ['1', '2', '4', '8', '16'].includes(v),
required: true,
@@ -34,7 +34,7 @@
required: true,
},
});
- const { alertTypeEnum, alertSource } = toRefs(props);
+ const { alertType, alertSource } = toRefs(props);
const DescriptionsItem = Descriptions.Item;
const emailInfo = computed(() => {
@@ -57,19 +57,19 @@
<template>
<div class="mt-10px border-dot text-16px cursor-pointer flex items-center">
<div class="flex items-center">
- <SvgIcon :name="alertTypeEnums[alertTypeEnum].icon" :size="20" class="!align-middle" />
+ <SvgIcon :name="alertTypes[alertType].icon" :size="20" class="!align-middle" />
<span class="pl-10px">
- {{ alertTypeEnums[alertTypeEnum].name }}
+ {{ alertTypes[alertType].name }}
</span>
</div>
</div>
<Descriptions size="small" :column="1" class="pl-15px mt-10px">
- <template v-if="alertTypeEnum === '1'">
+ <template v-if="alertType === '1'">
<DescriptionsItem :label="t('setting.alarm.alertEmail')">
<span class="text-blue-500">{{ emailInfo.contacts || '' }}</span>
</DescriptionsItem>
</template>
- <template v-else-if="alertTypeEnum === '2'">
+ <template v-else-if="alertType === '2'">
<DescriptionsItem :label="t('setting.alarm.dingTalkUser')">
{{ dingTalk.contacts || '' }}
</DescriptionsItem>
@@ -79,12 +79,12 @@
</Tag>
</DescriptionsItem>
</template>
- <template v-else-if="alertTypeEnum === '4'">
+ <template v-else-if="alertType === '4'">
<DescriptionsItem :label="t('setting.alarm.weChattoken')">
{{ desensitization(weChat.token || '') }}
</DescriptionsItem>
</template>
- <template v-else-if="alertTypeEnum === '16'">
+ <template v-else-if="alertType === '16'">
<DescriptionsItem :label="t('setting.alarm.larkIsAtAll')">
<Tag :color="lark.isAtAll ? 'green' : 'red'" class="!leading-20px">
{{ lark.isAtAll }}
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.data.ts b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.data.ts
index dcebfffb7..0becf9887 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.data.ts
+++ b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.data.ts
@@ -175,7 +175,7 @@ export const alertFormSchema: Array<FormSchema> = [
},
];
-export const alertTypeEnums = {
+export const alertTypes = {
'1': { name: t('setting.alarm.email'), value: 1, disabled: false, icon: 'mail' },
'2': { name: t('setting.alarm.dingTalk'), value: 2, disabled: false, icon: 'dingtalk' },
'4': { name: t('setting.alarm.weChat'), value: 4, disabled: false, icon: 'wecom' },
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.vue b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.vue
index 2660393cc..74f995316 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/Alarm/index.vue
@@ -50,7 +50,7 @@
/* Get alert configuration */
async function getAlertSetting() {
const res = await fetchAlertSetting();
- res.map((a) => (a.alertTypeTags = computeAlertType(a.alertTypeEnum)));
+ res.map((a) => (a.alertTypeTags = computeAlertType(a.alertType)));
alerts.value = res;
}
const alertTypeMap = {
@@ -117,7 +117,7 @@
openAlertModal(true, {
alertId: item.id,
alertName: item.alertName,
- alertTypeEnum: item.alertTypeTags,
+ alertType: item.alertTypeTags,
alertEmail: emailParams.contacts,
alertDingURL: dingTalkParams.alertDingURL,
dingtalkToken: dingTalkParams.token,
@@ -261,27 +261,27 @@
</template>
<AlertTypeInfo
- :alertTypeEnum="String(AlertTypeEnum.MAIL)"
+ :alertType="String(AlertTypeEnum.MAIL)"
:alertSource="item"
v-if="item.alertTypeTags.includes(String(AlertTypeEnum.MAIL))"
/>
<AlertTypeInfo
- :alertTypeEnum="String(AlertTypeEnum.WECOM)"
+ :alertType="String(AlertTypeEnum.WECOM)"
:alertSource="item"
v-if="item.alertTypeTags.includes(String(AlertTypeEnum.WECOM))"
/>
<AlertTypeInfo
- :alertTypeEnum="String(AlertTypeEnum.DINGTALK)"
+ :alertType="String(AlertTypeEnum.DINGTALK)"
:alertSource="item"
v-if="item.alertTypeTags.includes(String(AlertTypeEnum.DINGTALK))"
/>
<AlertTypeInfo
- :alertTypeEnum="String(AlertTypeEnum.MESSAGE)"
+ :alertType="String(AlertTypeEnum.MESSAGE)"
:alertSource="item"
v-if="item.alertTypeTags.includes(String(AlertTypeEnum.MESSAGE))"
/>
<AlertTypeInfo
- :alertTypeEnum="String(AlertTypeEnum.LARK)"
+ :alertType="String(AlertTypeEnum.LARK)"
:alertSource="item"
v-if="item.alertTypeTags.includes(String(AlertTypeEnum.LARK))"
/>
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/EditCluster.vue b/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/EditCluster.vue
index 817351228..9c2590226 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/EditCluster.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/EditCluster.vue
@@ -107,11 +107,11 @@
setFieldsValue({
clusterName: cluster.clusterName,
clusterId: cluster.clusterId,
- executionModeEnum: cluster.executionModeEnum,
+ executionMode: cluster.executionMode,
address: cluster.address,
description: cluster.description,
dynamicProperties: cluster.dynamicProperties,
- resolveOrderEnum: cluster.resolveOrderEnum,
+ resolveOrder: cluster.resolveOrder,
yarnQueue: cluster.yarnQueue,
alertId: selectAlertId,
versionId: cluster.versionId || null,
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/index.vue b/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/index.vue
index 452e54a95..4b330ca73 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/index.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/FlinkCluster/index.vue
@@ -59,7 +59,7 @@
const clusters = ref<FlinkCluster[]>([]);
const loading = ref(false);
function handleIsStart(item) {
- return item.clusterStateEnum === ClusterStateEnum.RUNNING;
+ return item.clusterState === ClusterStateEnum.RUNNING;
}
/* Go to edit cluster */
@@ -158,7 +158,7 @@
</ListItemMeta>
<div class="list-content" style="width: 10%">
<div class="list-content-item">
- <span>{{ t('setting.flinkCluster.form.executionModeEnum') }}</span>
+ <span>{{ t('setting.flinkCluster.form.executionMode') }}</span>
<p style="margin-top: 10px">
{{ item.executionModeEnum.toLowerCase() }}
</p>
@@ -168,8 +168,8 @@
class="list-content"
style="width: 40%"
v-if="
- item.executionModeEnum === ExecModeEnum.REMOTE ||
- item.executionModeEnum === ExecModeEnum.YARN_SESSION
+ item.executionMode === ExecModeEnum.REMOTE ||
+ item.executionMode === ExecModeEnum.YARN_SESSION
"
>
<div class="list-content-item">
@@ -185,14 +185,14 @@
class="list-content"
style="width: 10%"
v-if="
- item.executionModeEnum === ExecModeEnum.REMOTE ||
- item.executionModeEnum === ExecModeEnum.YARN_SESSION
+ item.executionMode === ExecModeEnum.REMOTE ||
+ item.executionMode === ExecModeEnum.YARN_SESSION
"
>
<div class="list-content-item">
<span>{{ t('setting.flinkCluster.form.runState') }}</span>
<p style="margin-top: 10px">
- <State :data="{ clusterStateEnum: item.clusterStateEnum }" />
+ <State :data="{ clusterState: item.clusterState }" />
</p>
</div>
</div>
@@ -212,7 +212,7 @@
<template v-if="handleIsStart(item)">
<Tooltip :title="t('setting.flinkCluster.stop')">
<a-button
- :disabled="item.executionModeEnum === ExecModeEnum.REMOTE"
+ :disabled="item.executionMode === ExecModeEnum.REMOTE"
v-auth="'cluster:create'"
@click="handleShutdownCluster(item)"
shape="circle"
@@ -227,7 +227,7 @@
<template v-else>
<Tooltip :title="t('setting.flinkCluster.start')">
<a-button
- :disabled="item.executionModeEnum === ExecModeEnum.REMOTE"
+ :disabled="item.executionMode === ExecModeEnum.REMOTE"
v-auth="'cluster:create'"
@click="handleDeployCluster(item)"
shape="circle"
diff --git a/streampark-console/streampark-console-webapp/src/views/system/user/User.vue b/streampark-console/streampark-console-webapp/src/views/system/user/User.vue
index 756c9b0b1..b04cac08a 100644
--- a/streampark-console/streampark-console-webapp/src/views/system/user/User.vue
+++ b/streampark-console/streampark-console-webapp/src/views/system/user/User.vue
@@ -107,7 +107,7 @@
tooltip: t('system.user.table.reset'),
ifShow: () =>
(record.username !== 'admin' || userName.value === 'admin') &&
- record.loginTypeEnum == LoginTypeEnum[LoginTypeEnum.PASSWORD],
+ record.loginType == LoginTypeEnum[LoginTypeEnum.PASSWORD],
popConfirm: {
title: t('system.user.table.resetTip'),
confirm: handleReset.bind(null, record),
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
index d9bf7f633..8487c3927 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/CancelRequest.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.bean
import org.apache.streampark.common.conf.{FlinkVersion, K8sFlinkConfig}
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import javax.annotation.Nullable
@@ -27,7 +27,7 @@ import java.util.{Map => JavaMap}
case class CancelRequest(
id: Long,
flinkVersion: FlinkVersion,
- executionMode: ExecutionModeEnum,
+ executionMode: FlinkExecutionMode,
@Nullable properties: JavaMap[String, Any],
clusterId: String,
jobId: String,
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
index e2ba9fc6c..13479ea11 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/DeployRequest.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.bean
import org.apache.streampark.common.conf.{FlinkVersion, Workspace}
-import org.apache.streampark.common.enums.{ExecutionModeEnum, FlinkK8sRestExposedTypeEnum}
+import org.apache.streampark.common.enums.{FlinkExecutionMode, FlinkK8sRestExposedType}
import org.apache.streampark.flink.util.FlinkUtils
import org.apache.commons.io.FileUtils
@@ -31,7 +31,7 @@ import java.util.{Map => JavaMap}
case class DeployRequest(
flinkVersion: FlinkVersion,
- executionMode: ExecutionModeEnum,
+ executionMode: FlinkExecutionMode,
properties: JavaMap[String, Any],
clusterId: String,
@Nullable k8sDeployParam: KubernetesDeployParam) {
@@ -69,5 +69,4 @@ case class KubernetesDeployParam(
kubeConf: String = "~/.kube/config",
serviceAccount: String = KubernetesConfigOptions.KUBERNETES_SERVICE_ACCOUNT.defaultValue(),
flinkImage: String = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),
- @Nullable flinkRestExposedType: FlinkK8sRestExposedTypeEnum =
- FlinkK8sRestExposedTypeEnum.CLUSTER_IP)
+ @Nullable flinkRestExposedType: FlinkK8sRestExposedType = FlinkK8sRestExposedType.CLUSTER_IP)
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/KubernetesSubmitParam.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/KubernetesSubmitParam.scala
index 8671d88bd..e3351200f 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/KubernetesSubmitParam.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/KubernetesSubmitParam.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.client.bean
-import org.apache.streampark.common.enums.FlinkK8sRestExposedTypeEnum
+import org.apache.streampark.common.enums.FlinkK8sRestExposedType
import org.apache.streampark.flink.kubernetes.v2.model.IngressDef
import javax.annotation.Nullable
@@ -51,7 +51,7 @@ case class KubernetesSubmitParam(
taskManagerEphemeralStorage: Option[String] = None,
taskManagerPodTemplate: Option[String] = None,
logConfiguration: JMap[String, String] = new util.HashMap[String, String](),
- flinkRestExposedType: Option[FlinkK8sRestExposedTypeEnum] = None,
+ flinkRestExposedType: Option[FlinkK8sRestExposedType] = None,
ingressDefinition: Option[IngressDef] = None
)
@@ -72,7 +72,7 @@ object KubernetesSubmitParam {
kubernetesName: String,
kubernetesNamespace: String,
baseImage: String,
- @Nullable flinkRestExposedType: FlinkK8sRestExposedTypeEnum,
+ @Nullable flinkRestExposedType: FlinkK8sRestExposedType,
@Nullable ingressDefinition: IngressDef): KubernetesSubmitParam =
KubernetesSubmitParam(
clusterId = clusterId,
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SavepointRequestTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SavepointRequestTrait.scala
index 3d2c5b542..3165250aa 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SavepointRequestTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SavepointRequestTrait.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.bean
import org.apache.streampark.common.conf.{FlinkVersion, K8sFlinkConfig}
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import javax.annotation.Nullable
@@ -28,7 +28,7 @@ trait SavepointRequestTrait {
val flinkVersion: FlinkVersion
- val executionMode: ExecutionModeEnum
+ val executionMode: FlinkExecutionMode
val clusterId: String
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
index 86c777a68..32d417527 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/ShutDownRequest.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.bean
import org.apache.streampark.common.conf.FlinkVersion
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import javax.annotation.Nullable
@@ -26,7 +26,7 @@ import java.util.{Map => JavaMap}
case class ShutDownRequest(
flinkVersion: FlinkVersion,
- executionMode: ExecutionModeEnum,
+ executionMode: FlinkExecutionMode,
@Nullable properties: JavaMap[String, Any],
clusterId: String,
@Nullable kubernetesDeployParam: KubernetesDeployParam)
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
index ef440a54c..18665186f 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
@@ -38,17 +38,17 @@ import scala.util.Try
case class SubmitRequest(
flinkVersion: FlinkVersion,
- executionMode: ExecutionModeEnum,
+ executionMode: FlinkExecutionMode,
properties: JavaMap[String, Any],
flinkYaml: String,
- developmentMode: DevelopmentModeEnum,
+ developmentMode: FlinkDevelopmentMode,
id: Long,
jobId: String,
appName: String,
appConf: String,
- applicationType: ApplicationTypeEnum,
+ applicationType: ApplicationType,
savePoint: String,
- restoreMode: RestoreModeEnum,
+ restoreMode: FlinkRestoreMode,
args: String,
@Nullable buildResult: BuildResult,
@Nullable k8sSubmitParam: KubernetesSubmitParam,
@@ -59,8 +59,8 @@ case class SubmitRequest(
lazy val appOption: Map[String, String] = getParameterMap(KEY_FLINK_OPTION_PREFIX)
lazy val appMain: String = this.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL => ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS
- case DevelopmentModeEnum.PYFLINK => ConfigConst.PYTHON_DRIVER_CLASS_NAME
+ case FlinkDevelopmentMode.FLINK_SQL => ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS
+ case FlinkDevelopmentMode.PYFLINK => ConfigConst.PYTHON_DRIVER_CLASS_NAME
case _ => appProperties(KEY_FLINK_APPLICATION_MAIN_CLASS)
}
@@ -82,7 +82,7 @@ case class SubmitRequest(
lazy val userJarFile: File = {
executionMode match {
- case ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION => null
+ case FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION => null
case _ =>
checkBuildResult()
new File(buildResult.asInstanceOf[ShadedBuildResponse].shadedJarPath)
@@ -169,7 +169,7 @@ case class SubmitRequest(
@throws[Exception]
def checkBuildResult(): Unit = {
executionMode match {
- case ExecutionModeEnum.KUBERNETES_NATIVE_SESSION =>
+ case FlinkExecutionMode.KUBERNETES_NATIVE_SESSION =>
if (buildResult == null) {
throw new Exception(
s"[flink-submit] current job: ${this.effectiveAppName} was not yet built, buildResult is empty" +
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/TriggerSavepointRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/TriggerSavepointRequest.scala
index 4c4acb9f7..065da0bea 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/TriggerSavepointRequest.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/TriggerSavepointRequest.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.bean
import org.apache.streampark.common.conf.{FlinkVersion, K8sFlinkConfig}
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import javax.annotation.Nullable
@@ -28,7 +28,7 @@ import java.util.{Map => JavaMap}
case class TriggerSavepointRequest(
id: Long,
flinkVersion: FlinkVersion,
- executionMode: ExecutionModeEnum,
+ executionMode: FlinkExecutionMode,
@Nullable properties: JavaMap[String, Any],
clusterId: String,
jobId: String,
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientEndpoint.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientEndpoint.scala
index 58a0da06b..3cd633968 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientEndpoint.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/FlinkClientEndpoint.scala
@@ -18,15 +18,15 @@
package org.apache.streampark.flink.client
import org.apache.streampark.common.conf.K8sFlinkConfig
-import org.apache.streampark.common.enums.ExecutionModeEnum
-import org.apache.streampark.common.enums.ExecutionModeEnum._
+import org.apache.streampark.common.enums.FlinkExecutionMode
+import org.apache.streampark.common.enums.FlinkExecutionMode._
import org.apache.streampark.flink.client.`trait`.FlinkClientTrait
import org.apache.streampark.flink.client.bean._
import org.apache.streampark.flink.client.impl._
object FlinkClientEndpoint {
- private[this] val clients: Map[ExecutionModeEnum, FlinkClientTrait] = Map(
+ private[this] val clients: Map[FlinkExecutionMode, FlinkClientTrait] = Map(
LOCAL -> LocalClient,
REMOTE -> RemoteClient,
YARN_APPLICATION -> YarnApplicationClient,
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala
index d08be791c..f53a3c74d 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeApplicationClient.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.client.impl
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import org.apache.streampark.common.util.Utils
import org.apache.streampark.flink.client.`trait`.KubernetesNativeClientTrait
import org.apache.streampark.flink.client.bean._
@@ -94,14 +94,16 @@ object KubernetesNativeApplicationClient extends KubernetesNativeClientTrait {
flinkConfig: Configuration): CancelResponse = {
flinkConfig.safeSet(
DeploymentOptions.TARGET,
- ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION.getName)
+ FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
super.doCancel(cancelRequest, flinkConfig)
}
override def doTriggerSavepoint(
triggerSavepointRequest: TriggerSavepointRequest,
flinkConf: Configuration): SavepointResponse = {
- flinkConf.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
+ flinkConf.safeSet(
+ DeploymentOptions.TARGET,
+ FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION.getName)
super.doTriggerSavepoint(triggerSavepointRequest, flinkConf)
}
}
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.scala
index 23ebb93d2..0ae093f0d 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/KubernetesNativeSessionClient.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.client.impl
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
import org.apache.streampark.common.util.{Logger, Utils}
import org.apache.streampark.flink.client.`trait`.KubernetesNativeClientTrait
import org.apache.streampark.flink.client.bean._
@@ -124,7 +124,7 @@ object KubernetesNativeSessionClient extends KubernetesNativeClientTrait with Lo
flinkConfig: Configuration): CancelResponse = {
flinkConfig.safeSet(
DeploymentOptions.TARGET,
- ExecutionModeEnum.KUBERNETES_NATIVE_SESSION.getName)
+ FlinkExecutionMode.KUBERNETES_NATIVE_SESSION.getName)
super.doCancel(cancelRequest, flinkConfig)
}
@@ -254,7 +254,9 @@ object KubernetesNativeSessionClient extends KubernetesNativeClientTrait with Lo
override def doTriggerSavepoint(
triggerSavepointRequest: TriggerSavepointRequest,
flinkConfig: Configuration): SavepointResponse = {
- flinkConfig.safeSet(DeploymentOptions.TARGET, ExecutionMode.KUBERNETES_NATIVE_SESSION.getName)
+ flinkConfig.safeSet(
+ DeploymentOptions.TARGET,
+ FlinkExecutionMode.KUBERNETES_NATIVE_SESSION.getName)
super.doTriggerSavepoint(triggerSavepointRequest, flinkConfig)
}
}
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
index 52441fd82..f39fa341a 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/impl/YarnApplicationClient.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.client.impl
import org.apache.streampark.common.conf.{ConfigConst, Workspace}
-import org.apache.streampark.common.enums.DevelopmentModeEnum
+import org.apache.streampark.common.enums.FlinkDevelopmentMode
import org.apache.streampark.common.fs.FsOperator
import org.apache.streampark.common.util.{FileUtils, HdfsUtils, Utils}
import org.apache.streampark.flink.client.`trait`.YarnClientTrait
@@ -69,7 +69,7 @@ object YarnApplicationClient extends YarnClientTrait {
submitRequest.hdfsWorkspace.appPlugins
)
submitRequest.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL =>
+ case FlinkDevelopmentMode.FLINK_SQL =>
array += s"${workspace.APP_SHIMS}/flink-${submitRequest.flinkVersion.majorVersion}"
val jobLib = s"${workspace.APP_WORKSPACE}/${submitRequest.id}/lib"
if (HdfsUtils.exists(jobLib)) {
@@ -95,7 +95,7 @@ object YarnApplicationClient extends YarnClientTrait {
// yarn application Type
.safeSet(YarnConfigOptions.APPLICATION_TYPE, submitRequest.applicationType.getName)
- if (submitRequest.developmentMode == DevelopmentModeEnum.PYFLINK) {
+ if (submitRequest.developmentMode == FlinkDevelopmentMode.PYFLINK) {
val pyVenv: String = workspace.APP_PYTHON_VENV
if (!FsOperator.hdfs.exists(pyVenv)) {
throw new RuntimeException(s"$pyVenv File does not exist")
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
index ee706f9d1..092488c47 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -19,7 +19,7 @@ package org.apache.streampark.flink.client.`trait`
import org.apache.streampark.common.conf.{ConfigConst, Workspace}
import org.apache.streampark.common.conf.ConfigConst._
-import org.apache.streampark.common.enums.{ApplicationTypeEnum, DevelopmentModeEnum, ExecutionModeEnum, RestoreModeEnum}
+import org.apache.streampark.common.enums.{ApplicationType, FlinkDevelopmentMode, FlinkExecutionMode, FlinkRestoreMode}
import org.apache.streampark.common.fs.FsOperator
import org.apache.streampark.common.util.{DeflaterUtils, FileUtils, Logger, SystemPropertyUtils}
import org.apache.streampark.flink.client.bean._
@@ -93,7 +93,7 @@ trait FlinkClientTrait extends Logger {
val (commandLine, flinkConfig) = getCommandLineAndFlinkConfig(submitRequest)
submitRequest.developmentMode match {
- case DevelopmentModeEnum.PYFLINK =>
+ case FlinkDevelopmentMode.PYFLINK =>
val flinkOptPath: String = System.getenv(ConfigConstants.ENV_FLINK_OPT_DIR)
if (StringUtils.isBlank(flinkOptPath)) {
logWarn(s"Get environment variable ${ConfigConstants.ENV_FLINK_OPT_DIR} fail")
@@ -140,9 +140,9 @@ trait FlinkClientTrait extends Logger {
submitRequest.allowNonRestoredState)
if (
submitRequest.flinkVersion.checkVersion(
- RestoreModeEnum.SINCE_FLINK_VERSION) && submitRequest.restoreMode != null
+ FlinkRestoreMode.SINCE_FLINK_VERSION) && submitRequest.restoreMode != null
) {
- flinkConfig.setString(RestoreModeEnum.RESTORE_MODE, submitRequest.restoreMode.getName);
+ flinkConfig.setString(FlinkRestoreMode.RESTORE_MODE, submitRequest.restoreMode.getName);
}
}
@@ -236,7 +236,7 @@ trait FlinkClientTrait extends Logger {
private[client] def getJobGraph(
submitRequest: SubmitRequest,
flinkConfig: Configuration): (PackagedProgram, JobGraph) = {
- if (submitRequest.developmentMode == DevelopmentModeEnum.PYFLINK) {
+ if (submitRequest.developmentMode == FlinkDevelopmentMode.PYFLINK) {
val pythonVenv: String = Workspace.local.APP_PYTHON_VENV
if (!FsOperator.lfs.exists(pythonVenv)) {
throw new RuntimeException(s"$pythonVenv File does not exist")
@@ -487,14 +487,14 @@ trait FlinkClientTrait extends Logger {
}
}
- if (submitRequest.applicationType == ApplicationTypeEnum.STREAMPARK_FLINK) {
+ if (submitRequest.applicationType == ApplicationType.STREAMPARK_FLINK) {
programArgs += PARAM_KEY_FLINK_CONF += submitRequest.flinkYaml
programArgs += PARAM_KEY_APP_NAME += DeflaterUtils.zipString(submitRequest.effectiveAppName)
programArgs += PARAM_KEY_FLINK_PARALLELISM += getParallelism(submitRequest).toString
submitRequest.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL =>
+ case FlinkDevelopmentMode.FLINK_SQL =>
programArgs += PARAM_KEY_FLINK_SQL += submitRequest.flinkSQL
if (submitRequest.appConf != null) {
programArgs += PARAM_KEY_APP_CONF += submitRequest.appConf
@@ -514,8 +514,8 @@ trait FlinkClientTrait extends Logger {
}
if (
- submitRequest.developmentMode == DevelopmentModeEnum.PYFLINK
- && submitRequest.executionMode != ExecutionModeEnum.YARN_APPLICATION
+ submitRequest.developmentMode == FlinkDevelopmentMode.PYFLINK
+ && submitRequest.executionMode != FlinkExecutionMode.YARN_APPLICATION
) {
// python file
programArgs.add("-py")
@@ -601,7 +601,7 @@ trait FlinkClientTrait extends Logger {
.key(CheckpointingOptions.SAVEPOINT_DIRECTORY.key())
.stringType()
.defaultValue {
- if (request.executionMode == ExecutionModeEnum.YARN_APPLICATION) {
+ if (request.executionMode == FlinkExecutionMode.YARN_APPLICATION) {
Workspace.remote.APP_SAVEPOINTS
} else null
}
diff --git a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeClientTrait.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeClientTrait.scala
index 1ce9a8304..712fd4bc8 100644
--- a/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeClientTrait.scala
+++ b/streampark-flink/streampark-flink-client/streampark-flink-client-core/src/main/scala/org/apache/streampark/flink/client/trait/KubernetesNativeClientTrait.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.client.`trait`
-import org.apache.streampark.common.enums.{ExecutionModeEnum, FlinkK8sRestExposedTypeEnum}
+import org.apache.streampark.common.enums.{FlinkExecutionMode, FlinkK8sRestExposedType}
import org.apache.streampark.flink.client.bean._
import org.apache.streampark.flink.kubernetes.PodTemplateTool
import org.apache.streampark.flink.packer.pipeline.DockerImageBuildResponse
@@ -48,7 +48,7 @@ trait KubernetesNativeClientTrait extends FlinkClientTrait {
covertToServiceExposedType(submitRequest.k8sSubmitParam.flinkRestExposedType.get))
if (submitRequest.buildResult != null) {
- if (submitRequest.executionMode == ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION) {
+ if (submitRequest.executionMode == FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION) {
val buildResult = submitRequest.buildResult.asInstanceOf[DockerImageBuildResponse]
buildResult.podTemplatePaths.foreach(
p => {
@@ -170,12 +170,11 @@ trait KubernetesNativeClientTrait extends FlinkClientTrait {
KubernetesConfigOptions.CLUSTER_ID)}, " +
s"namespace=${conf.get(KubernetesConfigOptions.NAMESPACE)}"
- private def covertToServiceExposedType(
- exposedType: FlinkK8sRestExposedTypeEnum): ServiceExposedType =
+ private def covertToServiceExposedType(exposedType: FlinkK8sRestExposedType): ServiceExposedType =
exposedType match {
- case FlinkK8sRestExposedTypeEnum.CLUSTER_IP => ServiceExposedType.ClusterIP
- case FlinkK8sRestExposedTypeEnum.LOAD_BALANCER => ServiceExposedType.LoadBalancer
- case FlinkK8sRestExposedTypeEnum.NODE_PORT => ServiceExposedType.NodePort
+ case FlinkK8sRestExposedType.CLUSTER_IP => ServiceExposedType.ClusterIP
+ case FlinkK8sRestExposedType.LOAD_BALANCER => ServiceExposedType.LoadBalancer
+ case FlinkK8sRestExposedType.NODE_PORT => ServiceExposedType.NodePort
case _ => ServiceExposedType.LoadBalancer
}
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkK8sExecuteModeEnum.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkK8sExecuteModeEnum.scala
index 8d52a1434..1de3f43ac 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkK8sExecuteModeEnum.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkK8sExecuteModeEnum.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.kubernetes.enums
-import org.apache.streampark.common.enums.ExecutionModeEnum
+import org.apache.streampark.common.enums.FlinkExecutionMode
/** execution mode of flink on kubernetes */
object FlinkK8sExecuteModeEnum extends Enumeration {
@@ -25,18 +25,18 @@ object FlinkK8sExecuteModeEnum extends Enumeration {
val SESSION: FlinkK8sExecuteModeEnum.Value = Value("kubernetes-session")
val APPLICATION: FlinkK8sExecuteModeEnum.Value = Value("kubernetes-application")
- def of(mode: ExecutionModeEnum): Value = {
+ def of(mode: FlinkExecutionMode): Value = {
mode match {
- case ExecutionModeEnum.KUBERNETES_NATIVE_SESSION => SESSION
- case ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION => APPLICATION
+ case FlinkExecutionMode.KUBERNETES_NATIVE_SESSION => SESSION
+ case FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION => APPLICATION
case _ => throw new IllegalStateException(s"Illegal K8sExecuteMode, ${mode.name}")
}
}
- def toExecutionMode(mode: FlinkK8sExecuteModeEnum.Value): ExecutionModeEnum = {
+ def toExecutionMode(mode: FlinkK8sExecuteModeEnum.Value): FlinkExecutionMode = {
mode match {
- case SESSION => ExecutionModeEnum.KUBERNETES_NATIVE_SESSION
- case APPLICATION => ExecutionModeEnum.KUBERNETES_NATIVE_APPLICATION
+ case SESSION => FlinkExecutionMode.KUBERNETES_NATIVE_SESSION
+ case APPLICATION => FlinkExecutionMode.KUBERNETES_NATIVE_APPLICATION
}
}
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
index 1dbfeab32..abca419b2 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/BuildRequest.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.packer.pipeline
import org.apache.streampark.common.conf.{FlinkVersion, Workspace}
-import org.apache.streampark.common.enums.{DevelopmentModeEnum, ExecutionModeEnum}
+import org.apache.streampark.common.enums.{FlinkDevelopmentMode, FlinkExecutionMode}
import org.apache.streampark.flink.kubernetes.model.K8sPodTemplates
import org.apache.streampark.flink.packer.docker.DockerConf
import org.apache.streampark.flink.packer.maven.DependencyInfo
@@ -38,9 +38,9 @@ sealed trait FlinkBuildParam extends BuildParam {
def workspace: String
- def executionMode: ExecutionModeEnum
+ def executionMode: FlinkExecutionMode
- def developmentMode: DevelopmentModeEnum
+ def developmentMode: FlinkDevelopmentMode
def flinkVersion: FlinkVersion
@@ -51,7 +51,7 @@ sealed trait FlinkBuildParam extends BuildParam {
lazy val providedLibs: DependencyInfo = {
val providedLibs =
ArrayBuffer(localWorkspace.APP_JARS, localWorkspace.APP_PLUGINS, customFlinkUserJar)
- if (developmentMode == DevelopmentModeEnum.FLINK_SQL) {
+ if (developmentMode == FlinkDevelopmentMode.FLINK_SQL) {
providedLibs += s"${localWorkspace.APP_SHIMS}/flink-${flinkVersion.majorVersion}"
}
dependencyInfo.merge(providedLibs.toSet)
@@ -76,8 +76,8 @@ case class FlinkK8sSessionBuildRequest(
workspace: String,
mainClass: String,
customFlinkUserJar: String,
- executionMode: ExecutionModeEnum,
- developmentMode: DevelopmentModeEnum,
+ executionMode: FlinkExecutionMode,
+ developmentMode: FlinkDevelopmentMode,
flinkVersion: FlinkVersion,
dependencyInfo: DependencyInfo,
clusterId: String,
@@ -89,8 +89,8 @@ case class FlinkK8sApplicationBuildRequest(
workspace: String,
mainClass: String,
customFlinkUserJar: String,
- executionMode: ExecutionModeEnum,
- developmentMode: DevelopmentModeEnum,
+ executionMode: FlinkExecutionMode,
+ developmentMode: FlinkDevelopmentMode,
flinkVersion: FlinkVersion,
dependencyInfo: DependencyInfo,
clusterId: String,
@@ -108,8 +108,8 @@ case class FlinkRemotePerJobBuildRequest(
mainClass: String,
customFlinkUserJar: String,
skipBuild: Boolean,
- executionMode: ExecutionModeEnum,
- developmentMode: DevelopmentModeEnum,
+ executionMode: FlinkExecutionMode,
+ developmentMode: FlinkDevelopmentMode,
flinkVersion: FlinkVersion,
dependencyInfo: DependencyInfo)
extends FlinkBuildParam
@@ -119,6 +119,6 @@ case class FlinkYarnApplicationBuildRequest(
mainClass: String,
localWorkspace: String,
yarnProvidedPath: String,
- developmentMode: DevelopmentModeEnum,
+ developmentMode: FlinkDevelopmentMode,
dependencyInfo: DependencyInfo)
extends BuildParam
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipeline.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipeline.scala
index f94af2347..6caba3e35 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipeline.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipeline.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.packer.pipeline.impl
-import org.apache.streampark.common.enums.DevelopmentModeEnum
+import org.apache.streampark.common.enums.FlinkDevelopmentMode
import org.apache.streampark.common.fs.LfsOperator
import org.apache.streampark.common.util.ThreadUtils
import org.apache.streampark.flink.kubernetes.PodTemplateTool
@@ -91,8 +91,8 @@ class FlinkK8sApplicationBuildPipeline(request: FlinkK8sApplicationBuildRequest)
execStep(3) {
val shadedJarOutputPath = request.getShadedJarPath(buildWorkspace)
val extJarLibs = request.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL => request.dependencyInfo.extJarLibs
- case DevelopmentModeEnum.CUSTOM_CODE => Set[String]()
+ case FlinkDevelopmentMode.FLINK_SQL => request.dependencyInfo.extJarLibs
+ case FlinkDevelopmentMode.CUSTOM_CODE => Set[String]()
}
val shadedJar =
MavenTool.buildFatJar(request.mainClass, request.providedLibs, shadedJarOutputPath)
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipelineV2.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipelineV2.scala
index ea490bbb6..5e086e517 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipelineV2.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkK8sApplicationBuildPipelineV2.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.packer.pipeline.impl
-import org.apache.streampark.common.enums.DevelopmentModeEnum
+import org.apache.streampark.common.enums.FlinkDevelopmentMode
import org.apache.streampark.common.fs.LfsOperator
import org.apache.streampark.flink.packer.maven.MavenTool
import org.apache.streampark.flink.packer.pipeline._
@@ -51,8 +51,8 @@ class FlinkK8sApplicationBuildPipelineV2(request: FlinkK8sApplicationBuildReques
execStep(2) {
val shadedJarOutputPath = request.getShadedJarPath(buildWorkspace)
val extJarLibs = request.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL => request.dependencyInfo.extJarLibs
- case DevelopmentModeEnum.CUSTOM_CODE => Set.empty[String]
+ case FlinkDevelopmentMode.FLINK_SQL => request.dependencyInfo.extJarLibs
+ case FlinkDevelopmentMode.CUSTOM_CODE => Set.empty[String]
case _ => Set.empty[String]
}
val shadedJar =
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
index 199e0b3db..fbd741070 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkRemoteBuildPipeline.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.packer.pipeline.impl
-import org.apache.streampark.common.enums.DevelopmentModeEnum
+import org.apache.streampark.common.enums.FlinkDevelopmentMode
import org.apache.streampark.common.fs.{FsOperator, LfsOperator}
import org.apache.streampark.flink.packer.maven.MavenTool
import org.apache.streampark.flink.packer.pipeline._
@@ -50,7 +50,7 @@ class FlinkRemoteBuildPipeline(request: FlinkRemotePerJobBuildRequest) extends B
val shadedJar =
execStep(2) {
request.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL =>
+ case FlinkDevelopmentMode.FLINK_SQL =>
val output = MavenTool.buildFatJar(
request.mainClass,
request.providedLibs,
@@ -64,7 +64,7 @@ class FlinkRemoteBuildPipeline(request: FlinkRemotePerJobBuildRequest) extends B
val mavenJars =
execStep(3) {
request.developmentMode match {
- case DevelopmentModeEnum.PYFLINK =>
+ case FlinkDevelopmentMode.PYFLINK =>
val mavenArts = MavenTool.resolveArtifacts(request.dependencyInfo.mavenArts.asJava)
mavenArts.map(_.getAbsolutePath) ++ request.dependencyInfo.extJarLibs
case _ => List[String]()
@@ -73,7 +73,7 @@ class FlinkRemoteBuildPipeline(request: FlinkRemotePerJobBuildRequest) extends B
execStep(4) {
request.developmentMode match {
- case DevelopmentModeEnum.PYFLINK =>
+ case FlinkDevelopmentMode.PYFLINK =>
mavenJars.foreach(
jar => {
val lfs: FsOperator = FsOperator.lfs
diff --git a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
index 36023cb8e..e804c834d 100644
--- a/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
+++ b/streampark-flink/streampark-flink-packer/src/main/scala/org/apache/streampark/flink/packer/pipeline/impl/FlinkYarnApplicationBuildPipeline.scala
@@ -18,7 +18,7 @@
package org.apache.streampark.flink.packer.pipeline.impl
import org.apache.streampark.common.conf.Workspace
-import org.apache.streampark.common.enums.DevelopmentModeEnum
+import org.apache.streampark.common.enums.FlinkDevelopmentMode
import org.apache.streampark.common.fs.{FsOperator, HdfsOperator, LfsOperator}
import org.apache.streampark.common.util.ImplicitsUtils._
import org.apache.streampark.flink.packer.maven.MavenTool
@@ -47,7 +47,7 @@ class FlinkYarnApplicationBuildPipeline(request: FlinkYarnApplicationBuildReques
override protected def buildProcess(): SimpleBuildResponse = {
execStep(1) {
request.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL | DevelopmentModeEnum.PYFLINK =>
+ case FlinkDevelopmentMode.FLINK_SQL | FlinkDevelopmentMode.PYFLINK =>
LfsOperator.mkCleanDirs(request.localWorkspace)
HdfsOperator.mkCleanDirs(request.yarnProvidedPath)
case _ =>
@@ -58,7 +58,7 @@ class FlinkYarnApplicationBuildPipeline(request: FlinkYarnApplicationBuildReques
val mavenJars =
execStep(2) {
request.developmentMode match {
- case DevelopmentModeEnum.FLINK_SQL | DevelopmentModeEnum.PYFLINK =>
+ case FlinkDevelopmentMode.FLINK_SQL | FlinkDevelopmentMode.PYFLINK =>
val mavenArts = MavenTool.resolveArtifacts(request.dependencyInfo.mavenArts)
mavenArts.map(_.getAbsolutePath) ++ request.dependencyInfo.extJarLibs
case _ => List[String]()
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkSqlValidator.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkSqlValidator.scala
index 87d7ec91d..26c31f928 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkSqlValidator.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkSqlValidator.scala
@@ -16,7 +16,7 @@
*/
package org.apache.streampark.flink.core
-import org.apache.streampark.common.enums.FlinkSqlValidationFailedTypeEnum
+import org.apache.streampark.common.enums.FlinkSqlValidationFailedType
import org.apache.streampark.common.util.{ExceptionUtils, Logger, Utils}
import org.apache.streampark.flink.core.SqlCommand._
@@ -103,7 +103,7 @@ object FlinkSqlValidator extends Logger {
val errorLine = call.lineStart + line.toInt - 1
return FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.SYNTAX_ERROR,
+ failedType = FlinkSqlValidationFailedType.SYNTAX_ERROR,
lineStart = call.lineStart,
lineEnd = call.lineEnd,
errorLine = errorLine,
@@ -114,7 +114,7 @@ object FlinkSqlValidator extends Logger {
} else {
return FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.SYNTAX_ERROR,
+ failedType = FlinkSqlValidationFailedType.SYNTAX_ERROR,
lineStart = call.lineStart,
lineEnd = call.lineEnd,
sql = call.originSql,
@@ -131,7 +131,7 @@ object FlinkSqlValidator extends Logger {
} else {
FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.SYNTAX_ERROR,
+ failedType = FlinkSqlValidationFailedType.SYNTAX_ERROR,
lineStart = sqlCommands.head.lineStart,
lineEnd = sqlCommands.last.lineEnd,
exception = "No 'INSERT' statement to trigger the execution of the Flink job."
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
index 4ecba14a1..87a54bd07 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
@@ -17,8 +17,8 @@
package org.apache.streampark.flink.core
import org.apache.streampark.common.conf.ConfigConst._
-import org.apache.streampark.common.enums.ApiTypeEnum
-import org.apache.streampark.common.enums.ApiTypeEnum.ApiType
+import org.apache.streampark.common.enums.ApiType
+import org.apache.streampark.common.enums.ApiType.ApiType
import org.apache.streampark.common.util._
import org.apache.streampark.flink.core.conf.FlinkConfiguration
@@ -36,13 +36,13 @@ private[flink] object FlinkStreamingInitializer {
def initialize(args: Array[String], config: (StreamExecutionEnvironment, ParameterTool) => Unit)
: (ParameterTool, StreamExecutionEnvironment) = {
- val flinkInitializer = new FlinkStreamingInitializer(args, ApiTypeEnum.SCALA)
+ val flinkInitializer = new FlinkStreamingInitializer(args, ApiType.SCALA)
flinkInitializer.streamEnvConfFunc = config
(flinkInitializer.configuration.parameter, flinkInitializer.streamEnv)
}
def initialize(args: StreamEnvConfig): (ParameterTool, StreamExecutionEnvironment) = {
- val flinkInitializer = new FlinkStreamingInitializer(args.args, ApiTypeEnum.JAVA)
+ val flinkInitializer = new FlinkStreamingInitializer(args.args, ApiType.JAVA)
flinkInitializer.javaStreamEnvConfFunc = args.conf
(flinkInitializer.configuration.parameter, flinkInitializer.streamEnv)
}
@@ -66,9 +66,9 @@ private[flink] class FlinkStreamingInitializer(args: Array[String], apiType: Api
JavaStreamEnv.getExecutionEnvironment(configuration.envConfig))
apiType match {
- case ApiTypeEnum.JAVA if javaStreamEnvConfFunc != null =>
+ case ApiType.JAVA if javaStreamEnvConfFunc != null =>
javaStreamEnvConfFunc.configuration(env.getJavaEnv, configuration.parameter)
- case ApiTypeEnum.SCALA if streamEnvConfFunc != null =>
+ case ApiType.SCALA if streamEnvConfFunc != null =>
streamEnvConfFunc(env, configuration.parameter)
case _ =>
}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
index d45d42336..2169b8ea5 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
@@ -17,8 +17,8 @@
package org.apache.streampark.flink.core
import org.apache.streampark.common.conf.ConfigConst._
-import org.apache.streampark.common.enums.{ApiTypeEnum, PlannerTypeEnum}
-import org.apache.streampark.common.enums.ApiTypeEnum.ApiType
+import org.apache.streampark.common.enums.{ApiType, PlannerType}
+import org.apache.streampark.common.enums.ApiType.ApiType
import org.apache.streampark.common.util.{DeflaterUtils, PropertiesUtils}
import org.apache.streampark.flink.core.EnhancerImplicit._
import org.apache.streampark.flink.core.conf.FlinkConfiguration
@@ -40,13 +40,13 @@ private[flink] object FlinkTableInitializer {
def initialize(
args: Array[String],
config: (TableConfig, ParameterTool) => Unit): (ParameterTool, TableEnvironment) = {
- val flinkInitializer = new FlinkTableInitializer(args, ApiTypeEnum.SCALA)
+ val flinkInitializer = new FlinkTableInitializer(args, ApiType.SCALA)
flinkInitializer.tableConfFunc = config
(flinkInitializer.configuration.parameter, flinkInitializer.tableEnv)
}
def initialize(args: TableEnvConfig): (ParameterTool, TableEnvironment) = {
- val flinkInitializer = new FlinkTableInitializer(args.args, ApiTypeEnum.JAVA)
+ val flinkInitializer = new FlinkTableInitializer(args.args, ApiType.JAVA)
flinkInitializer.javaTableEnvConfFunc = args.conf
(flinkInitializer.configuration.parameter, flinkInitializer.tableEnv)
}
@@ -57,7 +57,7 @@ private[flink] object FlinkTableInitializer {
configTable: (TableConfig, ParameterTool) => Unit)
: (ParameterTool, StreamExecutionEnvironment, StreamTableEnvironment) = {
- val flinkInitializer = new FlinkTableInitializer(args, ApiTypeEnum.SCALA)
+ val flinkInitializer = new FlinkTableInitializer(args, ApiType.SCALA)
flinkInitializer.streamEnvConfFunc = configStream
flinkInitializer.tableConfFunc = configTable
(
@@ -68,7 +68,7 @@ private[flink] object FlinkTableInitializer {
def initialize(args: StreamTableEnvConfig)
: (ParameterTool, StreamExecutionEnvironment, StreamTableEnvironment) = {
- val flinkInitializer = new FlinkTableInitializer(args.args, ApiTypeEnum.JAVA)
+ val flinkInitializer = new FlinkTableInitializer(args.args, ApiType.JAVA)
flinkInitializer.javaStreamEnvConfFunc = args.streamConfig
flinkInitializer.javaTableEnvConfFunc = args.tableConfig
(
@@ -86,9 +86,9 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
val builder = EnvironmentSettings.newInstance()
- Try(PlannerTypeEnum.withName(parameter.get(KEY_FLINK_TABLE_PLANNER)))
- .getOrElse(PlannerTypeEnum.BLINK) match {
- case PlannerTypeEnum.BLINK =>
+ Try(PlannerType.withName(parameter.get(KEY_FLINK_TABLE_PLANNER)))
+ .getOrElse(PlannerType.BLINK) match {
+ case PlannerType.BLINK =>
val useBlinkPlanner =
Try(builder.getClass.getDeclaredMethod("useBlinkPlanner")).getOrElse(null)
if (useBlinkPlanner == null) {
@@ -98,7 +98,7 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
useBlinkPlanner.invoke(builder)
logInfo("blinkPlanner will be use.")
}
- case PlannerTypeEnum.OLD =>
+ case PlannerType.OLD =>
val useOldPlanner = Try(builder.getClass.getDeclaredMethod("useOldPlanner")).getOrElse(null)
if (useOldPlanner == null) {
logWarn("useOldPlanner deprecated")
@@ -107,7 +107,7 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
useOldPlanner.invoke(builder)
logInfo("useOldPlanner will be use.")
}
- case PlannerTypeEnum.ANY =>
+ case PlannerType.ANY =>
val useAnyPlanner = Try(builder.getClass.getDeclaredMethod("useAnyPlanner")).getOrElse(null)
if (useAnyPlanner == null) {
logWarn("useAnyPlanner deprecated")
@@ -142,9 +142,9 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
envSettings.inBatchMode()
val tableEnv = TableEnvironment.create(envSettings.build()).setAppName
apiType match {
- case ApiTypeEnum.JAVA if javaTableEnvConfFunc != null =>
+ case ApiType.JAVA if javaTableEnvConfFunc != null =>
javaTableEnvConfFunc.configuration(tableEnv.getConfig, parameter)
- case ApiTypeEnum.SCALA if tableConfFunc != null =>
+ case ApiType.SCALA if tableConfFunc != null =>
tableConfFunc(tableEnv.getConfig, parameter)
case _ =>
}
@@ -164,9 +164,9 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
}
val streamTableEnv = StreamTableEnvironment.create(streamEnv, setting).setAppName
apiType match {
- case ApiTypeEnum.JAVA if javaTableEnvConfFunc != null =>
+ case ApiType.JAVA if javaTableEnvConfFunc != null =>
javaTableEnvConfFunc.configuration(streamTableEnv.getConfig, parameter)
- case ApiTypeEnum.SCALA if tableConfFunc != null =>
+ case ApiType.SCALA if tableConfFunc != null =>
tableConfFunc(streamTableEnv.getConfig, parameter)
case _ =>
}
diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/SqlCommandParser.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/SqlCommandParser.scala
index 4cd86aac8..0d21aab5c 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/SqlCommandParser.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/SqlCommandParser.scala
@@ -17,7 +17,7 @@
package org.apache.streampark.flink.core
import org.apache.streampark.common.conf.ConfigConst.PARAM_PREFIX
-import org.apache.streampark.common.enums.FlinkSqlValidationFailedTypeEnum
+import org.apache.streampark.common.enums.FlinkSqlValidationFailedType
import org.apache.streampark.common.util.Logger
import enumeratum.EnumEntry
@@ -46,7 +46,7 @@ object SqlCommandParser extends Logger {
validationCallback(
FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.VERIFY_FAILED,
+ failedType = FlinkSqlValidationFailedType.VERIFY_FAILED,
exception = sqlEmptyError))
null
} else {
@@ -62,7 +62,7 @@ object SqlCommandParser extends Logger {
validationCallback(
FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.UNSUPPORTED_SQL,
+ failedType = FlinkSqlValidationFailedType.UNSUPPORTED_SQL,
lineStart = segment.start,
lineEnd = segment.end,
exception = s"unsupported sql",
@@ -80,7 +80,7 @@ object SqlCommandParser extends Logger {
validationCallback(
FlinkSqlValidationResult(
success = false,
- failedType = FlinkSqlValidationFailedTypeEnum.VERIFY_FAILED,
+ failedType = FlinkSqlValidationFailedType.VERIFY_FAILED,
exception = "flink sql syntax error, no executable sql"))
null
} else {
@@ -407,7 +407,7 @@ case class SqlCommandCall(
case class FlinkSqlValidationResult(
success: JavaBool = true,
- failedType: FlinkSqlValidationFailedTypeEnum = null,
+ failedType: FlinkSqlValidationFailedType = null,
lineStart: Int = 0,
lineEnd: Int = 0,
errorLine: Int = 0,