You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/02/16 05:08:30 UTC
[incubator-streampark] branch dev updated: [Issue-2299] Support the yarn queue label quick-setting for yarn application mode. (#2317)
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 02da4c3ef [Issue-2299] Support the yarn queue label quick-setting for yarn application mode. (#2317)
02da4c3ef is described below
commit 02da4c3ef37329d9faba5a66b0ce359c1d90b9f1
Author: Pan Yuepeng <fl...@126.com>
AuthorDate: Thu Feb 16 13:08:25 2023 +0800
[Issue-2299] Support the yarn queue label quick-setting for yarn application mode. (#2317)
* [Issue-2299] Introduce YarnQueueLabelExpression util class
* [Issue-2299] Adapt the FE pages input placeholders
* [Issue-2299] Adjust the backend logic to process label quick-setting
* [Issue-2299] Support yarn session cluster queue label quick-setting
---
.../streampark/common/conf/ConfigConst.scala | 2 +
.../console/core/entity/Application.java | 12 +-
.../console/core/entity/FlinkCluster.java | 5 +-
.../core/service/impl/ApplicationServiceImpl.java | 46 +++++--
.../core/service/impl/FlinkClusterServiceImpl.java | 4 +
.../core/utils/YarnQueueLabelExpression.java | 133 +++++++++++++++++++++
.../core/utils/YarnQueueLabelExpressionTest.java | 66 ++++++++++
.../src/locales/lang/en/flink/app.ts | 1 +
.../src/locales/lang/en/flink/setting.ts | 1 -
.../src/locales/lang/zh-CN/flink/app.ts | 1 +
.../src/locales/lang/zh-CN/flink/setting.ts | 1 -
.../flink/app/hooks/useCreateAndEditSchema.ts | 5 +-
.../src/views/flink/app/hooks/useFlinkRender.tsx | 22 ++++
.../src/views/flink/app/hooks/useFlinkSchema.ts | 5 +-
.../views/flink/setting/hooks/useClusterSetting.ts | 5 +-
15 files changed, 279 insertions(+), 30 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
index d02dc6d31..ab9d325eb 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/conf/ConfigConst.scala
@@ -124,6 +124,8 @@ object ConfigConst {
val KEY_YARN_APP_QUEUE = "yarn.application.queue"
+ val KEY_YARN_APP_NODE_LABEL = "yarn.application.node-label"
+
val KEY_K8S_IMAGE_PULL_POLICY = "kubernetes.container.image.pull-policy"
// ---table---
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 5c891b5c9..0c014eda3 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
@@ -17,7 +17,6 @@
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.ApplicationType;
@@ -36,6 +35,7 @@ import org.apache.streampark.console.core.enums.FlinkAppState;
import org.apache.streampark.console.core.enums.LaunchState;
import org.apache.streampark.console.core.enums.ResourceFrom;
import org.apache.streampark.console.core.metrics.flink.JobsOverview;
+import org.apache.streampark.console.core.utils.YarnQueueLabelExpression;
import org.apache.streampark.flink.kubernetes.model.K8sPodTemplates;
import org.apache.streampark.flink.packer.maven.Artifact;
import org.apache.streampark.flink.packer.maven.DependencyInfo;
@@ -597,16 +597,18 @@ public class Application implements Serializable {
public void updateHotParams(Application appParam) {
ExecutionMode executionModeEnum = appParam.getExecutionModeEnum();
Map<String, String> hotParams = new HashMap<>(0);
- if (ExecutionMode.YARN_APPLICATION.equals(executionModeEnum)) {
- if (StringUtils.isNotEmpty(appParam.getYarnQueue())) {
- hotParams.put(ConfigConst.KEY_YARN_APP_QUEUE(), appParam.getYarnQueue());
- }
+ if (needFillYarnQueueLabel(executionModeEnum)) {
+ hotParams.putAll(YarnQueueLabelExpression.getQueueLabelMap(appParam.getYarnQueue()));
}
if (!hotParams.isEmpty()) {
this.setHotParams(JacksonUtils.write(hotParams));
}
}
+ private boolean needFillYarnQueueLabel(ExecutionMode mode) {
+ return ExecutionMode.YARN_PER_JOB.equals(mode) || ExecutionMode.YARN_APPLICATION.equals(mode);
+ }
+
@Data
public static class Dependency {
private List<Pom> pom = Collections.emptyList();
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 f9098a0e2..4e8a38070 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/FlinkCluster.java
@@ -27,6 +27,7 @@ import org.apache.streampark.common.util.YarnUtils;
import org.apache.streampark.console.base.util.CommonUtils;
import org.apache.streampark.console.base.util.JacksonUtils;
import org.apache.streampark.console.core.metrics.flink.Overview;
+import org.apache.streampark.console.core.utils.YarnQueueLabelExpression;
import org.apache.streampark.flink.client.FlinkClient;
import com.baomidou.mybatisplus.annotation.FieldStrategy;
@@ -125,9 +126,7 @@ public class FlinkCluster implements Serializable {
Map<String, Object> map = JacksonUtils.read(this.options, Map.class);
if (ExecutionMode.YARN_SESSION.equals(getExecutionModeEnum())) {
map.put(ConfigConst.KEY_YARN_APP_NAME(), this.clusterName);
- if (StringUtils.isNotEmpty(this.yarnQueue)) {
- map.put(ConfigConst.KEY_YARN_APP_QUEUE(), this.yarnQueue);
- }
+ map.putAll(YarnQueueLabelExpression.getQueueLabelMap(yarnQueue));
}
map.entrySet().removeIf(entry -> entry.getValue() == null);
return map;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index d51002325..d394148ec 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -76,6 +76,7 @@ import org.apache.streampark.console.core.service.SavePointService;
import org.apache.streampark.console.core.service.SettingService;
import org.apache.streampark.console.core.service.VariableService;
import org.apache.streampark.console.core.task.FlinkRESTAPIWatcher;
+import org.apache.streampark.console.core.utils.YarnQueueLabelExpression;
import org.apache.streampark.flink.client.FlinkClient;
import org.apache.streampark.flink.client.bean.CancelRequest;
import org.apache.streampark.flink.client.bean.CancelResponse;
@@ -131,6 +132,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -145,6 +147,7 @@ import java.util.stream.Collectors;
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;
+import static org.apache.streampark.console.core.utils.YarnQueueLabelExpression.checkQueueLabelIfNeed;
@Slf4j
@Service
@@ -698,6 +701,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
appParam.setOptionState(OptionState.NONE.getValue());
appParam.setCreateTime(new Date());
appParam.setDefaultModeIngress(settingService.getIngressModeDefault());
+ checkQueueLabelIfNeed(appParam.getExecutionMode(), appParam.getYarnQueue());
appParam.doSetHotParams();
if (appParam.isUploadJob()) {
String jarPath =
@@ -813,6 +817,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Transactional(rollbackFor = {Exception.class})
public boolean update(Application appParam) {
try {
+ checkQueueLabelIfNeed(appParam.getExecutionMode(), appParam.getYarnQueue());
Application application = getById(appParam.getId());
application.setLaunch(LaunchState.NEED_LAUNCH.get());
if (application.isUploadJob()) {
@@ -1115,17 +1120,28 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
}
- if (!application.getHotParamsMap().isEmpty()) {
- if (ExecutionMode.YARN_APPLICATION.equals(application.getExecutionModeEnum())) {
- if (application.getHotParamsMap().containsKey(ConfigConst.KEY_YARN_APP_QUEUE())) {
- application.setYarnQueue(
- application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_QUEUE()).toString());
- }
- }
- }
+ setYarnQueue(application);
+
return application;
}
+ private void setYarnQueue(Application application) {
+ if (!(ExecutionMode.YARN_APPLICATION == application.getExecutionModeEnum()
+ || ExecutionMode.YARN_PER_JOB == application.getExecutionModeEnum())) {
+ return;
+ }
+
+ Map<String, Object> hotParamsMap = application.getHotParamsMap();
+ if (!hotParamsMap.isEmpty() && hotParamsMap.containsKey(ConfigConst.KEY_YARN_APP_QUEUE())) {
+ String yarnQueue = hotParamsMap.get(ConfigConst.KEY_YARN_APP_QUEUE()).toString();
+ String labelExpr =
+ Optional.ofNullable(hotParamsMap.get(ConfigConst.KEY_YARN_APP_NODE_LABEL()))
+ .map(Object::toString)
+ .orElse(null);
+ application.setYarnQueue(YarnQueueLabelExpression.of(yarnQueue, labelExpr).toString());
+ }
+ }
+
@Override
public String getMain(Application application) {
File jarFile;
@@ -1620,11 +1636,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
properties.put(RestOptions.ADDRESS.key(), activeAddress.getHost());
properties.put(RestOptions.PORT.key(), activeAddress.getPort());
} else if (ExecutionMode.isYarnMode(application.getExecutionModeEnum())) {
- String yarnQueue =
- (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_QUEUE());
- if (yarnQueue != null) {
- properties.put(ConfigConst.KEY_YARN_APP_QUEUE(), yarnQueue);
- }
if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
Utils.required(
@@ -1634,6 +1645,15 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
+ "the cluster has been deleted. Please contact the Admin.",
application.getFlinkClusterId()));
properties.put(ConfigConst.KEY_YARN_APP_ID(), cluster.getClusterId());
+ } else {
+ String yarnQueue =
+ (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_QUEUE());
+ String yarnLabelExpr =
+ (String) application.getHotParamsMap().get(ConfigConst.KEY_YARN_APP_NODE_LABEL());
+ Optional.ofNullable(yarnQueue)
+ .ifPresent(yq -> properties.put(ConfigConst.KEY_YARN_APP_QUEUE(), yq));
+ Optional.ofNullable(yarnLabelExpr)
+ .ifPresent(yLabel -> properties.put(ConfigConst.KEY_YARN_APP_NODE_LABEL(), yLabel));
}
} else if (ExecutionMode.isKubernetesMode(application.getExecutionModeEnum())) {
properties.put(ConfigConst.KEY_K8S_IMAGE_PULL_POLICY(), "Always");
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 9a586d8dc..5442ac580 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
@@ -55,6 +55,8 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import static org.apache.streampark.console.core.utils.YarnQueueLabelExpression.checkQueueLabelIfNeed;
+
@Slf4j
@Service
@Transactional(propagation = Propagation.SUPPORTS, readOnly = true, rollbackFor = Exception.class)
@@ -124,6 +126,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
@Override
public Boolean create(FlinkCluster flinkCluster) {
flinkCluster.setUserId(commonService.getUserId());
+ checkQueueLabelIfNeed(flinkCluster.getExecutionMode(), flinkCluster.getYarnQueue());
flinkCluster.setCreateTime(new Date());
if (ExecutionMode.REMOTE.equals(flinkCluster.getExecutionModeEnum())) {
flinkCluster.setClusterState(ClusterState.STARTED.getValue());
@@ -195,6 +198,7 @@ public class FlinkClusterServiceImpl extends ServiceImpl<FlinkClusterMapper, Fli
@Override
public void update(FlinkCluster cluster) {
+ checkQueueLabelIfNeed(cluster.getExecutionMode(), cluster.getYarnQueue());
FlinkCluster flinkCluster = getById(cluster.getId());
flinkCluster.setClusterName(cluster.getClusterName());
flinkCluster.setDescription(cluster.getDescription());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java
new file mode 100644
index 000000000..370005cb4
--- /dev/null
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.utils;
+
+import org.apache.streampark.common.conf.ConfigConst;
+import org.apache.streampark.common.enums.ExecutionMode;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.regex.Pattern;
+
+/** Util class for parsing and checking Yarn queue & Label */
+public class YarnQueueLabelExpression {
+
+ private static final String AT = "@";
+
+ private static final String REGEX = "[a-zA-Z0-9_\\-]+";
+
+ @VisibleForTesting public static final String ERR_HINTS = "Yarn queue label format is invalid.";
+
+ private static final Pattern QUEUE_LABEL_PATTERN =
+ Pattern.compile(String.format("^(%s)(.%s)*(%s(%s)(,%s)*)?$", REGEX, REGEX, AT, REGEX, REGEX));
+
+ private static final String QUEUE_LABEL_FORMAT = "%s" + AT + "%s";
+
+ private final String queue;
+ private @Nullable final String labelExpression;
+
+ private YarnQueueLabelExpression(String queue, String labelExpression) {
+ this.labelExpression = StringUtils.isEmpty(labelExpression) ? null : labelExpression;
+ this.queue = queue;
+ }
+
+ public Optional<String> getLabelExpression() {
+ return Optional.ofNullable(labelExpression);
+ }
+
+ public @Nonnull String getQueue() {
+ return queue;
+ }
+
+ @Override
+ public String toString() {
+ return StringUtils.isEmpty(labelExpression)
+ ? queue
+ : String.format(QUEUE_LABEL_FORMAT, queue, labelExpression);
+ }
+
+ private static boolean isValid(String queueLabel, boolean ignoreEmpty) {
+ if (StringUtils.isEmpty(queueLabel)) {
+ return ignoreEmpty;
+ }
+ return QUEUE_LABEL_PATTERN.matcher(queueLabel).matches();
+ }
+
+ // Visible for test.
+ public static boolean isValid(String queueLabel) {
+ return isValid(queueLabel, false);
+ }
+
+ /**
+ * Only check the validation of queue-labelExpression when using yarn application or yarn-session
+ * mode.
+ *
+ * @param executionMode execution mode.
+ * @param queueLabel queueLabel expression.
+ */
+ public static void checkQueueLabelIfNeed(int executionMode, String queueLabel) {
+ if (ExecutionMode.isYarnMode(executionMode)) {
+ if (!YarnQueueLabelExpression.isValid(queueLabel, true)) {
+ throw new IllegalArgumentException(ERR_HINTS);
+ }
+ }
+ }
+
+ // Visible for test.
+ public static YarnQueueLabelExpression of(@Nonnull String queueLabelExpr) {
+ if (isValid(queueLabelExpr, false)) {
+ String[] strs = queueLabelExpr.split(AT);
+ if (strs.length == 2) {
+ return new YarnQueueLabelExpression(strs[0], strs[1]);
+ }
+ return new YarnQueueLabelExpression(strs[0], null);
+ }
+ throw new IllegalArgumentException(ERR_HINTS);
+ }
+
+ public static YarnQueueLabelExpression of(
+ @Nonnull String queue, @Nullable String labelExpression) {
+ YarnQueueLabelExpression queueLabelExpression =
+ new YarnQueueLabelExpression(queue, labelExpression);
+ if (isValid(queueLabelExpression.toString(), false)) {
+ return queueLabelExpression;
+ }
+ throw new IllegalArgumentException(ERR_HINTS);
+ }
+
+ public static Map<String, String> getQueueLabelMap(String queueLabelExp) {
+ if (StringUtils.isEmpty(queueLabelExp)) {
+ return Collections.emptyMap();
+ }
+ YarnQueueLabelExpression yarnQueueLabelExpression = of(queueLabelExp);
+ Map<String, String> map = new HashMap<>(2);
+ yarnQueueLabelExpression
+ .getLabelExpression()
+ .ifPresent(labelExp -> map.put(ConfigConst.KEY_YARN_APP_NODE_LABEL(), labelExp));
+ map.put(ConfigConst.KEY_YARN_APP_QUEUE(), yarnQueueLabelExpression.queue);
+ return map;
+ }
+}
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpressionTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpressionTest.java
new file mode 100644
index 000000000..8068c688b
--- /dev/null
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpressionTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.core.utils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.streampark.console.core.utils.YarnQueueLabelExpression.ERR_HINTS;
+import static org.apache.streampark.console.core.utils.YarnQueueLabelExpression.isValid;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test class for {@link YarnQueueLabelExpression}. */
+class YarnQueueLabelExpressionTest {
+
+ @Test
+ void testIsValid() {
+ // Test 'default' queue
+ assertThat(isValid("default")).isTrue();
+ // Test default queue default label
+ assertThat(isValid("default@default")).isTrue();
+ // Test queue1, with label1,label2
+ assertThat(isValid("queue1@label1,label2")).isTrue();
+ // test 'default ' queue
+ assertThat(isValid("default ")).isFalse();
+ // test 'root.default' queue
+ assertThat(isValid("root.default")).isTrue();
+ // test 'root.default@' queue
+ assertThat(isValid("root.default@")).isFalse();
+ // test 'root.default@label1'
+ assertThat(isValid("root.default@label1")).isTrue();
+ // test 'root.default@label1,label2'
+ assertThat(isValid("root.default@label1,label2")).isTrue();
+ // test 'root.default@label1,' queue
+ assertThat(isValid("root.default@label1,")).isFalse();
+ // test 'default@'
+ assertThat(isValid("default@")).isFalse();
+ // test 'default.@label1,label2'
+ assertThat(isValid("default.@label1,label2")).isFalse();
+ // test 'queue1@label2, '
+ assertThat(isValid("queue1@label2, ")).isFalse();
+ }
+
+ @Test
+ void testOf() {
+ assertThat(YarnQueueLabelExpression.of("a").getLabelExpression()).isEmpty();
+ assertThat(YarnQueueLabelExpression.of("a").getQueue()).isEqualTo("a");
+ assertThatThrownBy(() -> YarnQueueLabelExpression.of("a@"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(ERR_HINTS);
+ }
+}
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
index 2a8de5270..0bc92f92e 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
@@ -242,5 +242,6 @@ export default {
'Explicitly configuring both "total process memory" and "total Flink memory" is not recommended. It may lead to deployment failures due to potential memory configuration conflicts. Configuring other memory components also requires caution as it can produce further configuration conflicts, The easiest way is to set "total process memory"',
dynamicProperties:
'It works the same as -D$property=$value in CLI mode, Allows specifying multiple generic configuration options. The available options can be found from',
+ yarnQueue: 'It\'s used to quickly specify not only "yarn.application.name" but also "yarn.application.node-label". For example, "queue1" represents the value of "yarn.application.name" is "queue1", but "queue1@label1,label2" represents that the value of "yarn.application.name" is "queue1" and the value of "yarn.application.node-label" is "label1,label2". The queue and label are joined by "@".',
},
};
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/setting.ts b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/setting.ts
index e27686b80..3bc7ac3b1 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/setting.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/setting.ts
@@ -148,7 +148,6 @@ export default {
clusterName: 'Please enter cluster name',
executionMode: 'Please enter cluster name',
versionId: 'please select Flink Version',
- yarnQueue: 'Please enter yarn queue',
addressRemoteMode: 'Please enter jobManager URL',
addressNoRemoteMode: 'Please enter cluster address, e.g: http://host:port',
yarnSessionClusterId: 'Please enter Yarn Session cluster',
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
index f43768089..f45916752 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
@@ -235,5 +235,6 @@ export default {
'不建议同时配置 "total process memory" 和 "total Flink memory"。 由于潜在的内存配置冲突，它可能会导致部署失败。 配置其他内存组件也需要谨慎，因为它会产生进一步的配置冲突，最简单的方法是设置"total process memory"',
dynamicProperties:
'动态Properties: 与flink run命令行模式下的 -D$property=$value 的作用相同，允许指定多个参数。 更多可以设置的参数请查阅',
+ yarnQueue: '此输入不仅支持快速设置 "yarn.application.name" 还支持设置 "yarn.application.node-label"。例如，输入 "queue1" 表示 "yarn.application.name" 的值为 "queue1"，而输入 "queue1@label1,label2" 则表示 "yarn.application.name" 设置为 "queue1" 且 "yarn.application.node-label" 设置为 "label1,label2"。Queue 和 label 之间使用 "@" 分隔。',
},
};
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/setting.ts b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/setting.ts
index 88b01fe41..8edc086b9 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/setting.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/setting.ts
@@ -146,7 +146,6 @@ export default {
clusterName: '请输入集群名称',
executionMode: '请选择执行模式',
versionId: '请选择Flink版本',
- yarnQueue: '请选择Yarn队列',
addressRemoteMode: '请输入Flink 集群JobManager URL访问地址',
addressNoRemoteMode: '请输入集群地址，例如：http://host:port',
yarnSessionClusterId: '请输入Yarn Session模式集群ID',
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
index ab731b402..0dc9502c0 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useCreateAndEditSchema.ts
@@ -26,6 +26,7 @@ import {
renderIsSetConfig,
renderOptionsItems,
renderTotalMemory,
+ renderYarnQueue
} from './useFlinkRender';
import { fetchCheckName } from '/@/api/flink/app/app';
@@ -466,8 +467,8 @@ export const useCreateAndEditSchema = (
field: 'yarnQueue',
label: t('flink.app.yarnQueue'),
component: 'Input',
- componentProps: { placeholder: t('flink.app.addAppTips.yarnQueuePlaceholder') },
- ifShow: ({ values }) => values.executionMode == ExecModeEnum.YARN_APPLICATION,
+ ifShow: ({ values }) => values.executionMode == ExecModeEnum.YARN_APPLICATION || values.executionMode == ExecModeEnum.YARN_PER_JOB,
+ render: renderYarnQueue
},
{
field: 'podTemplate',
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
index 761ca5578..c953abcff 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx
@@ -231,6 +231,28 @@ export const renderOptionsItems = (
});
};
+/* render Yarn Queue */
+export const renderYarnQueue = ({ model, field }: RenderCallbackParams) => {
+ return (
+ <div>
+ <Input
+ name="yarnQueue"
+ placeholder={t('flink.app.addAppTips.yarnQueuePlaceholder')}
+ value={model[field]}
+ onInput={(e: ChangeEvent) => (model[field] = e?.target?.value)}
+ />
+ <p class="conf-desc mt-10px">
+ <span class="note-info">
+ <Tag color="#2db7f5" class="tag-note">
+ {t('flink.app.noteInfo.note')}
+ </Tag>
+ {t('flink.app.noteInfo.yarnQueue')}
+ </span>
+ </p>
+ </div>
+ );
+};
+
/* render memory option */
export const renderDynamicProperties = ({ model, field }: RenderCallbackParams) => {
return (
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkSchema.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkSchema.ts
index d1a25c691..2ba99ce36 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkSchema.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkSchema.ts
@@ -39,6 +39,7 @@ import {
renderOptionsItems,
getAlertSvgIcon,
renderIsSetConfig,
+ renderYarnQueue
} from './useFlinkRender';
import Icon from '/@/components/Icon';
import { Alert } from 'ant-design-vue';
@@ -339,8 +340,8 @@ export const useFlinkSchema = (editModel?: string) => {
field: 'yarnQueue',
label: t('flink.app.yarnQueue'),
component: 'Input',
- componentProps: { placeholder: t('flink.app.addAppTips.yarnQueuePlaceholder') },
- ifShow: ({ values }) => values.executionMode == ExecModeEnum.YARN_APPLICATION,
+ ifShow: ({ values }) => values.executionMode == ExecModeEnum.YARN_APPLICATION || values.executionMode == ExecModeEnum.YARN_PER_JOB,
+ render: renderYarnQueue
},
{
field: 'podTemplate',
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/setting/hooks/useClusterSetting.ts b/streampark-console/streampark-console-webapp/src/views/flink/setting/hooks/useClusterSetting.ts
index 841e59d66..a4822200f 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/setting/hooks/useClusterSetting.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/setting/hooks/useClusterSetting.ts
@@ -24,6 +24,7 @@ import {
renderInputDropdown,
renderOptionsItems,
renderTotalMemory,
+ renderYarnQueue,
} from '../../app/hooks/useFlinkRender';
import { fetchCheckHadoop } from '/@/api/flink/setting';
import { fetchFlinkEnv } from '/@/api/flink/setting/flinkEnv';
@@ -153,9 +154,7 @@ export const useClusterSetting = () => {
label: t('flink.setting.cluster.form.yarnQueue'),
component: 'Input',
ifShow: ({ values }) => values.executionMode == ExecModeEnum.YARN_SESSION,
- componentProps: {
- placeholder: t('flink.setting.cluster.placeholder.yarnQueue'),
- },
+ render: renderYarnQueue
},
{
field: 'clusterId',