fix the #14729 problem (#14902)

LiuCanWu · 刘阳 · SbloodyS · web-flow · commit e748c2eb9a8e · 2023-09-21T18:46:14.000+08:00
Co-authored-by: 刘阳 &lt;liuy2590@chinaunicom.cn&gt;
Co-authored-by: xiangzihao &lt;460888207@qq.com&gt;
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java
@@ -273,6 +273,8 @@ private static List<String> buildRunCommandLineForOthers(TaskExecutionContext ta
             args.add(others);
         }
 
+        // determine yarn queue
+        determinedYarnQueue(args, flinkParameters, deployMode, flinkVersion);
         ProgramType programType = flinkParameters.getProgramType();
         String mainClass = flinkParameters.getMainClass();
         if (programType != null && programType != ProgramType.PYTHON && StringUtils.isNotEmpty(mainClass)) {
@@ -295,8 +297,6 @@ private static List<String> buildRunCommandLineForOthers(TaskExecutionContext ta
             args.add(ParameterUtils.convertParameterPlaceholders(mainArgs, ParameterUtils.convert(paramsMap)));
         }
 
-        // determine yarn queue
-        determinedYarnQueue(args, flinkParameters, deployMode, flinkVersion);
         return args;
     }
 
@@ -310,8 +310,10 @@ private static void determinedYarnQueue(List<String> args, FlinkParameters flink
                 } else {
                     doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);
                 }
+                break;
             case APPLICATION:
                 doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS);
+                break;
         }
     }
 
@@ -323,9 +325,11 @@ private static void doAddQueue(List<String> args, FlinkParameters flinkParameter
                 switch (option) {
                     case FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS:
                         args.add(String.format(FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS + "=%s", yarnQueue));
+                        break;
                     case FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE:
                         args.add(FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);
                         args.add(yarnQueue);
+                        break;
                 }
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -273,6 +273,8 @@ private static List<String> buildRunCommandLineForOthers(TaskExecutionContext ta`
`273`	`273`	`args.add(others);`
`274`	`274`	`}`
`275`	`275`
	`276`	`+ // determine yarn queue`
	`277`	`+ determinedYarnQueue(args, flinkParameters, deployMode, flinkVersion);`
`276`	`278`	`ProgramType programType = flinkParameters.getProgramType();`
`277`	`279`	`String mainClass = flinkParameters.getMainClass();`
`278`	`280`	`if (programType != null && programType != ProgramType.PYTHON && StringUtils.isNotEmpty(mainClass)) {`
`@@ -295,8 +297,6 @@ private static List<String> buildRunCommandLineForOthers(TaskExecutionContext ta`
`295`	`297`	`args.add(ParameterUtils.convertParameterPlaceholders(mainArgs, ParameterUtils.convert(paramsMap)));`
`296`	`298`	`}`
`297`	`299`
`298`		`- // determine yarn queue`
`299`		`- determinedYarnQueue(args, flinkParameters, deployMode, flinkVersion);`
`300`	`300`	`return args;`
`301`	`301`	`}`
`302`	`302`
`@@ -310,8 +310,10 @@ private static void determinedYarnQueue(List<String> args, FlinkParameters flink`
`310`	`310`	`} else {`
`311`	`311`	`doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);`
`312`	`312`	`}`
	`313`	`+ break;`
`313`	`314`	`case APPLICATION:`
`314`	`315`	`doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS);`
	`316`	`+ break;`
`315`	`317`	`}`
`316`	`318`	`}`
`317`	`319`
`@@ -323,9 +325,11 @@ private static void doAddQueue(List<String> args, FlinkParameters flinkParameter`
`323`	`325`	`switch (option) {`
`324`	`326`	`case FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS:`
`325`	`327`	`args.add(String.format(FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS + "=%s", yarnQueue));`
	`328`	`+ break;`
`326`	`329`	`case FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE:`
`327`	`330`	`args.add(FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);`
`328`	`331`	`args.add(yarnQueue);`
	`332`	`+ break;`
`329`	`333`	`}`
`330`	`334`	`}`
`331`	`335`	`}`