You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@streampark.apache.org by "RocMarshal (via GitHub)" <gi...@apache.org> on 2023/04/09 15:55:03 UTC
[GitHub] [incubator-streampark] RocMarshal commented on a diff in pull request #2448: [Feature][Issue-2324/Issue-2420] Support selection on yarn queue when create/edit yarn per-job/application-mode application & yarn-session cluster.
RocMarshal commented on code in PR #2448:
URL: https://github.com/apache/incubator-streampark/pull/2448#discussion_r1161296854
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java:
##########
@@ -1724,4 +1731,52 @@ private String getSavePointed(Application appParam) {
}
return null;
}
+
+ /** Check queue label validation when create the application if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(Application appParam) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ appParam.getExecutionModeEnum(), appParam.getYarnQueue());
+ if (!isYarnPerJobAppModeNotDefaultQueue(appParam)) {
+ return;
+ }
+ boolean exist =
+ yarnQueueService.existByTeamIdQueueLabel(appParam.getTeamId(), appParam.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_APP_QUEUE_HINT, appParam.getYarnQueue(), appParam.getTeamId()));
+ }
+
+ /** Check queue label validation when update the application if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(Application oldApp, Application newApp) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ newApp.getExecutionModeEnum(), newApp.getYarnQueue());
+ if (!isYarnPerJobAppModeNotDefaultQueue(newApp)) {
+ return;
+ }
+
+ oldApp.setYarnQueueByHotParams();
+ if (ExecutionMode.isYarnPerJobOrAppMode(newApp.getExecutionModeEnum())
+ && StringUtils.equals(oldApp.getYarnQueue(), newApp.getYarnQueue())) {
+ return;
+ }
+ boolean exist =
+ yarnQueueService.existByTeamIdQueueLabel(newApp.getTeamId(), newApp.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_APP_QUEUE_HINT, newApp.getYarnQueue(), newApp.getTeamId()));
+ }
+
+ /**
+ * Judge the execution mode whether is the Yarn PerJob or Application mode with not default or
+ * empty queue label.
+ *
+ * @param application application entity.
+ * @return If the executionMode is (Yarn PerJob or application mode) and the queue label is not
+ * (empty or default), return true, false else.
+ */
+ @VisibleForTesting
+ public boolean isYarnPerJobAppModeNotDefaultQueue(Application application) {
Review Comment:
done.
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java:
##########
@@ -356,4 +361,49 @@ public void delete(FlinkCluster cluster) {
}
removeById(id);
}
+
+ /** Check queue label validation when create the cluster if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(FlinkCluster clusterInfo) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ clusterInfo.getExecutionModeEnum(), clusterInfo.getYarnQueue());
+ if (!isYarnSessionModeAndNotDefaultQueue(clusterInfo)) {
+ return;
+ }
+ boolean exist = yarnQueueService.existByQueueLabel(clusterInfo.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_CLUSTER_QUEUE_HINT, clusterInfo.getYarnQueue()));
+ }
+
+ /** Check queue label validation when update the cluster if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(FlinkCluster oldCluster, FlinkCluster newCluster) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ newCluster.getExecutionModeEnum(), newCluster.getYarnQueue());
+ if (!isYarnSessionModeAndNotDefaultQueue(newCluster)) {
+ return;
+ }
+
+ if (ExecutionMode.isYarnSessionMode(newCluster.getExecutionModeEnum())
+ && StringUtils.equals(oldCluster.getYarnQueue(), newCluster.getYarnQueue())) {
+ return;
+ }
+ boolean exist = yarnQueueService.existByQueueLabel(newCluster.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_CLUSTER_QUEUE_HINT, newCluster.getYarnQueue()));
+ }
+
+ /**
+ * Judge the execution mode whether is the Yarn session mode with not default or empty queue
+ * label.
+ *
+ * @param cluster cluster.
+ * @return If the executionMode is yarn session mode and the queue label is not (empty or
+ * default), return true, false else.
+ */
+ @VisibleForTesting
+ public boolean isYarnSessionModeAndNotDefaultQueue(FlinkCluster cluster) {
Review Comment:
Done.
##########
streampark-common/src/main/java/org/apache/streampark/common/enums/ExecutionMode.java:
##########
@@ -78,6 +78,10 @@ public static boolean isYarnMode(ExecutionMode mode) {
return YARN_PER_JOB.equals(mode) || YARN_APPLICATION.equals(mode) || YARN_SESSION.equals(mode);
}
+ public static boolean isYarnPerJobOrAppMode(ExecutionMode mode) {
Review Comment:
two caller. I'll add a todo tag.
We would inline it back when the perjob mode is dropped.
##########
streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useFlinkRender.tsx:
##########
@@ -235,18 +238,24 @@ export const renderOptionsItems = (
export const renderYarnQueue = ({ model, field }: RenderCallbackParams) => {
return (
<div>
- <Input
+ <ApiSelect
name="yarnQueue"
- placeholder={t('flink.app.addAppTips.yarnQueuePlaceholder')}
+ placeholder={t('setting.yarnQueue.placeholder.yarnQueueLabelExpression')}
+ api={fetchYarnQueueList}
+ params={{ page: 1, pageSize: 9999, teamId: getUserTeamId() }}
Review Comment:
done.
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkClusterServiceImpl.java:
##########
@@ -356,4 +361,49 @@ public void delete(FlinkCluster cluster) {
}
removeById(id);
}
+
+ /** Check queue label validation when create the cluster if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(FlinkCluster clusterInfo) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ clusterInfo.getExecutionModeEnum(), clusterInfo.getYarnQueue());
+ if (!isYarnSessionModeAndNotDefaultQueue(clusterInfo)) {
+ return;
+ }
+ boolean exist = yarnQueueService.existByQueueLabel(clusterInfo.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_CLUSTER_QUEUE_HINT, clusterInfo.getYarnQueue()));
+ }
+
+ /** Check queue label validation when update the cluster if needed. */
+ @VisibleForTesting
+ public void checkQueueValidationIfNeeded(FlinkCluster oldCluster, FlinkCluster newCluster) {
+ yarnQueueService.checkQueueLabelFormatIfNeeded(
+ newCluster.getExecutionModeEnum(), newCluster.getYarnQueue());
+ if (!isYarnSessionModeAndNotDefaultQueue(newCluster)) {
+ return;
+ }
+
+ if (ExecutionMode.isYarnSessionMode(newCluster.getExecutionModeEnum())
+ && StringUtils.equals(oldCluster.getYarnQueue(), newCluster.getYarnQueue())) {
+ return;
+ }
+ boolean exist = yarnQueueService.existByQueueLabel(newCluster.getYarnQueue());
+ ApiAlertException.throwIfFalse(
+ exist, String.format(ERROR_CLUSTER_QUEUE_HINT, newCluster.getYarnQueue()));
+ }
+
+ /**
+ * Judge the execution mode whether is the Yarn session mode with not default or empty queue
+ * label.
+ *
+ * @param cluster cluster.
+ * @return If the executionMode is yarn session mode and the queue label is not (empty or
+ * default), return true, false else.
+ */
+ @VisibleForTesting
+ public boolean isYarnSessionModeAndNotDefaultQueue(FlinkCluster cluster) {
+ return ExecutionMode.isYarnSessionMode(cluster.getExecutionModeEnum())
+ && !yarnQueueService.isEmptyOrDefaultQueue(cluster.getYarnQueue());
+ }
Review Comment:
done.
##########
streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/YarnQueueServiceImpl.java:
##########
@@ -166,6 +168,40 @@ public void deleteYarnQueue(YarnQueue yarnQueue) {
removeById(yarnQueue.getId());
}
+ /**
+ * Only check the validation of queue-labelExpression when using yarn application or yarn-session
+ * mode or yarn-perjob mode.
+ *
+ * @param executionMode execution mode.
+ * @param queueLabel queueLabel expression.
+ */
+ @Override
+ public void checkQueueLabelFormatIfNeeded(ExecutionMode executionMode, String queueLabel) {
+ if (ExecutionMode.isYarnMode(executionMode)) {
+ ApiAlertException.throwIfFalse(isValid(queueLabel, true), ERR_FORMAT_HINTS);
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@streampark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org