You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/03/22 00:59:32 UTC
[incubator-streampark] branch dev updated: [fix] fix security issue and polish code (#2486)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new b43d357f1 [fix] fix security issue and polish code (#2486)
b43d357f1 is described below
commit b43d357f13d82f199785a71156739425890391e6
Author: gongzhongqiang <76...@qq.com>
AuthorDate: Wed Mar 22 08:59:23 2023 +0800
[fix] fix security issue and polish code (#2486)
Co-authored-by: gongzhongqiang <go...@gigacloudtech.com>
---
.../streampark/common/enums/ClusterState.java | 2 +-
.../streampark/common/enums/DevelopmentMode.java | 4 +--
.../common/enums/FlinkK8sRestExposedType.java | 6 ++--
.../org/apache/streampark/common/util/Utils.scala | 1 -
.../console/core/controller/IndexController.java | 3 +-
.../console/core/entity/Application.java | 4 +--
.../alert/impl/EmailAlertNotifyServiceImpl.java | 1 +
.../core/service/impl/AppBuildPipeServiceImpl.java | 6 ++--
.../service/impl/ApplicationBackUpServiceImpl.java | 32 ++++------------------
.../core/service/impl/FlinkClusterServiceImpl.java | 4 +--
.../console/core/task/CheckpointProcessor.java | 3 ++
.../flink/client/bean/DeployRequest.scala | 2 +-
.../flink/client/bean/SubmitRequest.scala | 2 +-
.../flink/client/impl/YarnApplicationClient.scala | 2 +-
.../flink/client/trait/FlinkClientTrait.scala | 2 +-
.../client/trait/KubernetesNativeClientTrait.scala | 6 ++--
.../flink/packer/pipeline/BuildRequest.scala | 2 +-
.../impl/FlinkK8sApplicationBuildPipeline.scala | 4 +--
.../impl/FlinkYarnApplicationBuildPipeline.scala | 4 +--
19 files changed, 37 insertions(+), 53 deletions(-)
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
index 3c13a1927..6b0ea9c4c 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/ClusterState.java
@@ -26,7 +26,7 @@ public enum ClusterState implements Serializable {
/** cluster started */
STARTED(1),
/** cluster stopped */
- STOPED(2),
+ STOPPED(2),
/** cluster lost */
LOST(3);
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentMode.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentMode.java
index 0c74f6b4d..5649d5f95 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentMode.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/DevelopmentMode.java
@@ -22,10 +22,10 @@ import java.io.Serializable;
public enum DevelopmentMode implements Serializable {
/** custom code */
- CUSTOMCODE("Custom Code", 1),
+ CUSTOM_CODE("Custom Code", 1),
/** Flink SQL */
- FLINKSQL("Flink SQL", 2);
+ FLINK_SQL("Flink SQL", 2);
private final String mode;
diff --git a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
index 888cc6c9b..29426e768 100644
--- a/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
+++ b/streampark-common/src/main/java/org/apache/streampark/common/enums/FlinkK8sRestExposedType.java
@@ -23,13 +23,13 @@ import java.io.Serializable;
public enum FlinkK8sRestExposedType implements Serializable {
/** LoadBalancer */
- LoadBalancer("LoadBalancer", 0),
+ LOAD_BALANCER("LoadBalancer", 0),
/** ClusterIP */
- ClusterIP("ClusterIP", 1),
+ CLUSTER_IP("ClusterIP", 1),
/** NodePort */
- NodePort("NodePort", 2);
+ NODE_PORT("NodePort", 2);
private final String name;
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
index a372b7daa..b89eec0a8 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
@@ -191,7 +191,6 @@ object Utils {
case c if c == classOf[Double] => v.toDouble.asInstanceOf[T]
case c if c == classOf[Short] => v.toShort.asInstanceOf[T]
case c if c == classOf[Boolean] => v.toBoolean.asInstanceOf[T]
- case c if c == classOf[String] => v.asInstanceOf[T]
case c if c == classOf[JavaByte] => v.toByte.asInstanceOf[T]
case c if c == classOf[JavaInt] => JavaInt.valueOf(v).asInstanceOf[T]
case c if c == classOf[JavaLong] => JavaLong.valueOf(v).asInstanceOf[T]
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/IndexController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/IndexController.java
index 3bf3ca408..950a4d130 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/IndexController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/IndexController.java
@@ -18,6 +18,7 @@
package org.apache.streampark.console.core.controller;
import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.servlet.ModelAndView;
@@ -25,7 +26,7 @@ import org.springframework.web.servlet.ModelAndView;
@Controller
public class IndexController {
- @RequestMapping("/")
+ @GetMapping("/")
public ModelAndView index() {
return new ModelAndView("/index.html");
}
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 00a77714d..975ad7832 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
@@ -445,12 +445,12 @@ public class Application implements Serializable {
@JsonIgnore
public boolean isFlinkSqlJob() {
- return DevelopmentMode.FLINKSQL.getValue().equals(this.getJobType());
+ return DevelopmentMode.FLINK_SQL.getValue().equals(this.getJobType());
}
@JsonIgnore
public boolean isCustomCodeJob() {
- return DevelopmentMode.CUSTOMCODE.getValue().equals(this.getJobType());
+ return DevelopmentMode.CUSTOM_CODE.getValue().equals(this.getJobType());
}
@JsonIgnore
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/EmailAlertNotifyServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/EmailAlertNotifyServiceImpl.java
index 3c1d974ba..38ff6d71a 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/EmailAlertNotifyServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/EmailAlertNotifyServiceImpl.java
@@ -84,6 +84,7 @@ public class EmailAlertNotifyServiceImpl implements AlertNotifyService {
htmlEmail.setAuthentication(senderEmail.getUserName(), senderEmail.getPassword());
htmlEmail.setFrom(senderEmail.getFrom());
if (senderEmail.isSsl()) {
+ htmlEmail.setSSLCheckServerIdentity(true);
htmlEmail.setSSLOnConnect(true);
htmlEmail.setSslSmtpPort(senderEmail.getSmtpPort().toString());
} else {
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 7a978e1b0..e24f94c76 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
@@ -328,7 +328,7 @@ public class AppBuildPipeServiceImpl
case YARN_APPLICATION:
String yarnProvidedPath = app.getAppLib();
String localWorkspace = app.getLocalAppHome().concat("/lib");
- if (app.getDevelopmentMode().equals(DevelopmentMode.CUSTOMCODE)
+ if (app.getDevelopmentMode().equals(DevelopmentMode.CUSTOM_CODE)
&& app.getApplicationType().equals(ApplicationType.APACHE_FLINK)) {
yarnProvidedPath = app.getAppHome();
localWorkspace = app.getLocalAppHome();
@@ -407,7 +407,7 @@ public class AppBuildPipeServiceImpl
/** copy from {@link ApplicationServiceImpl#start(Application, boolean)} */
private String retrieveFlinkUserJar(FlinkEnv flinkEnv, Application app) {
switch (app.getDevelopmentMode()) {
- case CUSTOMCODE:
+ case CUSTOM_CODE:
switch (app.getApplicationType()) {
case STREAMPARK_FLINK:
return String.format("%s/%s", app.getAppLib(), app.getModule().concat(".jar"));
@@ -418,7 +418,7 @@ public class AppBuildPipeServiceImpl
"[StreamPark] unsupported ApplicationType of custom code: "
+ app.getApplicationType());
}
- case FLINKSQL:
+ case FLINK_SQL:
String sqlDistJar = commonService.getSqlClientJar(flinkEnv);
if (app.getExecutionModeEnum() == ExecutionMode.YARN_APPLICATION) {
String clientPath = Workspace.remote().APP_CLIENT();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
index d2a48594d..5622ee388 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
@@ -18,7 +18,6 @@
package org.apache.streampark.console.core.service.impl;
import org.apache.streampark.common.fs.FsOperator;
-import org.apache.streampark.common.util.ThreadUtils;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.base.exception.InternalException;
@@ -47,11 +46,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
@Slf4j
@Service
@Transactional(propagation = Propagation.SUPPORTS, readOnly = true, rollbackFor = Exception.class)
@@ -67,16 +61,6 @@ public class ApplicationBackUpServiceImpl
@Autowired private FlinkSqlService flinkSqlService;
- private final ExecutorService executorService =
- new ThreadPoolExecutor(
- Runtime.getRuntime().availableProcessors() * 5,
- Runtime.getRuntime().availableProcessors() * 10,
- 60L,
- TimeUnit.SECONDS,
- new LinkedBlockingQueue<>(1024),
- ThreadUtils.threadFactory("streampark-rollback-executor"),
- new ThreadPoolExecutor.AbortPolicy());
-
@Override
public IPage<ApplicationBackUp> page(ApplicationBackUp backUp, RestRequest request) {
Page<ApplicationBackUp> page = new MybatisPager<ApplicationBackUp>().getDefaultPage(request);
@@ -116,7 +100,7 @@ public class ApplicationBackUpServiceImpl
// if running, set Latest
if (application.isRunning()) {
- // rollback to buckup config
+ // rollback to back up config
configService.setLatestOrEffective(true, backParam.getId(), backParam.getAppId());
} else {
effectiveService.saveOrUpdate(backParam.getAppId(), EffectiveType.CONFIG, backParam.getId());
@@ -135,15 +119,11 @@ public class ApplicationBackUpServiceImpl
fsOperator.copyDir(backParam.getPath(), application.getAppHome());
// update restart status
- try {
- applicationService.update(
- new UpdateWrapper<Application>()
- .lambda()
- .eq(Application::getId, application.getId())
- .set(Application::getRelease, ReleaseState.NEED_RESTART.get()));
- } catch (Exception e) {
- throw e;
- }
+ applicationService.update(
+ new UpdateWrapper<Application>()
+ .lambda()
+ .eq(Application::getId, application.getId())
+ .set(Application::getRelease, ReleaseState.NEED_RESTART.get()));
}
@Override
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 229ec40f0..1a3b29a5f 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
@@ -196,7 +196,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
}
} catch (Exception e) {
log.error(e.getMessage(), e);
- flinkCluster.setClusterState(ClusterState.STOPED.getValue());
+ flinkCluster.setClusterState(ClusterState.STOPPED.getValue());
flinkCluster.setException(e.toString());
updateById(flinkCluster);
throw new ApiDetailException(e);
@@ -299,7 +299,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
ShutDownResponse shutDownResponse = future.get(60, TimeUnit.SECONDS);
if (shutDownResponse != null) {
flinkCluster.setAddress(null);
- flinkCluster.setClusterState(ClusterState.STOPED.getValue());
+ flinkCluster.setClusterState(ClusterState.STOPPED.getValue());
updateById(flinkCluster);
} else {
throw new ApiAlertException("get shutdown response failed");
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/CheckpointProcessor.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/CheckpointProcessor.java
index 6cd8149f7..a1cb65fe6 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/CheckpointProcessor.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/CheckpointProcessor.java
@@ -119,6 +119,9 @@ public class CheckpointProcessor {
throw new RuntimeException(e);
}
break;
+ default:
+ // do nothing
+ break;
}
} else {
counter.increment();
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 bcfd9c1c8..bd2629e6d 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
@@ -64,4 +64,4 @@ case class KubernetesDeployParam(clusterId: String,
kubeConf: String = "~/.kube/config",
serviceAccount: String = KubernetesConfigOptions.KUBERNETES_SERVICE_ACCOUNT.defaultValue(),
flinkImage: String = KubernetesConfigOptions.CONTAINER_IMAGE.defaultValue(),
- @Nullable flinkRestExposedType: FlinkK8sRestExposedType = FlinkK8sRestExposedType.ClusterIP)
+ @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/SubmitRequest.scala b/streampark-flink/streampark-flink-client/streampark-flink-client-api/src/main/scala/org/apache/streampark/flink/client/bean/SubmitRequest.scala
index f21968628..e928447ce 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
@@ -62,7 +62,7 @@ case class SubmitRequest(flinkVersion: FlinkVersion,
lazy val appOption: Map[String, String] = getParameterMap(KEY_FLINK_OPTION_PREFIX)
lazy val appMain: String = this.developmentMode match {
- case DevelopmentMode.FLINKSQL => ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS
+ case DevelopmentMode.FLINK_SQL => ConfigConst.STREAMPARK_FLINKSQL_CLIENT_CLASS
case _ => appProperties(KEY_FLINK_APPLICATION_MAIN_CLASS)
}
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 e583c90ec..d2729833d 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
@@ -66,7 +66,7 @@ object YarnApplicationClient extends YarnClientTrait {
submitRequest.hdfsWorkspace.appJars,
submitRequest.hdfsWorkspace.appPlugins)
submitRequest.developmentMode match {
- case DevelopmentMode.FLINKSQL =>
+ case DevelopmentMode.FLINK_SQL =>
val version = submitRequest.flinkVersion.version.split("\\.").map(_.trim.toInt)
version match {
case Array(1, 12, _) => array += s"${workspace.APP_SHIMS}/flink-1.12"
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 364959b1c..0a4441f26 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
@@ -430,7 +430,7 @@ trait FlinkClientTrait extends Logger {
programArgs += PARAM_KEY_FLINK_PARALLELISM
programArgs += getParallelism(submitRequest).toString
submitRequest.developmentMode match {
- case DevelopmentMode.FLINKSQL =>
+ case DevelopmentMode.FLINK_SQL =>
programArgs += PARAM_KEY_FLINK_SQL
programArgs += submitRequest.flinkSQL
if (submitRequest.appConf != 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 04ce7be91..a439aeef7 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
@@ -150,9 +150,9 @@ trait KubernetesNativeClientTrait extends FlinkClientTrait {
s"namespace=${conf.get(KubernetesConfigOptions.NAMESPACE)}"
private def covertToServiceExposedType(exposedType: FlinkK8sRestExposedType): ServiceExposedType = exposedType match {
- case FlinkK8sRestExposedType.ClusterIP => ServiceExposedType.ClusterIP
- case FlinkK8sRestExposedType.LoadBalancer => ServiceExposedType.LoadBalancer
- case FlinkK8sRestExposedType.NodePort => 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-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 acde98fd0..6a512e7f6 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
@@ -55,7 +55,7 @@ sealed trait FlinkBuildParam extends BuildParam {
localWorkspace.APP_JARS,
localWorkspace.APP_PLUGINS,
customFlinkUserJar)
- if (developmentMode == DevelopmentMode.FLINKSQL) {
+ if (developmentMode == DevelopmentMode.FLINK_SQL) {
providedLibs += {
val version = flinkVersion.version.split("\\.").map(_.trim.toInt)
version match {
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 cb005ca60..9df52fd6c 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
@@ -87,8 +87,8 @@ class FlinkK8sApplicationBuildPipeline(request: FlinkK8sApplicationBuildRequest)
execStep(3) {
val shadedJarOutputPath = request.getShadedJarPath(buildWorkspace)
val extJarLibs = request.developmentMode match {
- case DevelopmentMode.FLINKSQL => request.dependencyInfo.extJarLibs
- case DevelopmentMode.CUSTOMCODE => Set[String]()
+ case DevelopmentMode.FLINK_SQL => request.dependencyInfo.extJarLibs
+ case DevelopmentMode.CUSTOM_CODE => Set[String]()
}
val shadedJar = MavenTool.buildFatJar(request.mainClass, request.providedLibs, shadedJarOutputPath)
logInfo(s"output shaded flink job jar: ${shadedJar.getAbsolutePath}")
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 12bc36024..c5df6ea0a 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
@@ -49,7 +49,7 @@ class FlinkYarnApplicationBuildPipeline(request: FlinkYarnApplicationBuildReques
override protected def buildProcess(): SimpleBuildResponse = {
execStep(1) {
request.developmentMode match {
- case DevelopmentMode.FLINKSQL =>
+ case DevelopmentMode.FLINK_SQL =>
LfsOperator.mkCleanDirs(request.localWorkspace)
HdfsOperator.mkCleanDirs(request.yarnProvidedPath)
case _ =>
@@ -60,7 +60,7 @@ class FlinkYarnApplicationBuildPipeline(request: FlinkYarnApplicationBuildReques
val mavenJars =
execStep(2) {
request.developmentMode match {
- case DevelopmentMode.FLINKSQL =>
+ case DevelopmentMode.FLINK_SQL =>
val mavenArts = MavenTool.resolveArtifacts(request.dependencyInfo.mavenArts)
mavenArts.map(_.getAbsolutePath) ++ request.dependencyInfo.extJarLibs
case _ => Set[String]()