Radeity commented on code in PR #14237:
URL:
https://github.com/apache/dolphinscheduler/pull/14237#discussion_r1210262143
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java:
##########
@@ -197,16 +196,19 @@ private static List<String>
buildRunCommandLineForOthers(TaskExecutionContext ta
args.add(FlinkConstants.FLINK_RUN); // run
args.add(FlinkConstants.FLINK_EXECUTION_TARGET); // -t
args.add(FlinkConstants.FLINK_YARN_PER_JOB); //
yarn-per-job
+
Review Comment:
Please remove redundant blank line.
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java:
##########
@@ -197,16 +196,19 @@ private static List<String>
buildRunCommandLineForOthers(TaskExecutionContext ta
args.add(FlinkConstants.FLINK_RUN); // run
args.add(FlinkConstants.FLINK_EXECUTION_TARGET); // -t
args.add(FlinkConstants.FLINK_YARN_PER_JOB); //
yarn-per-job
+
} else {
args.add(FlinkConstants.FLINK_RUN); // run
args.add(FlinkConstants.FLINK_RUN_MODE); // -m
args.add(FlinkConstants.FLINK_YARN_CLUSTER); //
yarn-cluster
+
Review Comment:
Ditto.
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java:
##########
@@ -164,12 +164,9 @@ public static List<String>
buildInitOptionsForSql(FlinkParameters flinkParameter
}
// yarn.application.queue
- String others = flinkParameters.getOthers();
- if (StringUtils.isEmpty(others) ||
!others.contains(FlinkConstants.FLINK_QUEUE)) {
- String queue = flinkParameters.getQueue();
- if (StringUtils.isNotEmpty(queue)) {
-
initOptions.add(String.format(FlinkConstants.FLINK_FORMAT_YARN_APPLICATION_QUEUE,
queue));
- }
+ String queue = flinkParameters.getQueue();
+ if (StringUtils.isNotEmpty(queue)) {
Review Comment:
We will check whether user defines this arg by themselves in `others`, so
you just have to add one condition in the original way:
`!others.contains(FlinkConstants. FLINK_QUEUE_FOR_TARGETS)`, btw, may I ask why
you name them `FLINK_QUEUE_FOR_MODE` and `FLINK_QUEUE_FOR_TARGETS `?
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java:
##########
@@ -197,16 +196,19 @@ private static List<String>
buildRunCommandLineForOthers(TaskExecutionContext ta
args.add(FlinkConstants.FLINK_RUN); // run
args.add(FlinkConstants.FLINK_EXECUTION_TARGET); // -t
args.add(FlinkConstants.FLINK_YARN_PER_JOB); //
yarn-per-job
+
} else {
args.add(FlinkConstants.FLINK_RUN); // run
args.add(FlinkConstants.FLINK_RUN_MODE); // -m
args.add(FlinkConstants.FLINK_YARN_CLUSTER); //
yarn-cluster
+
}
break;
case APPLICATION:
args.add(FlinkConstants.FLINK_RUN_APPLICATION); //
run-application
args.add(FlinkConstants.FLINK_EXECUTION_TARGET); // -t
args.add(FlinkConstants.FLINK_YARN_APPLICATION); //
yarn-application
+
Review Comment:
Ditto.
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java:
##########
@@ -306,6 +299,41 @@ private static List<String>
buildRunCommandLineForOthers(TaskExecutionContext ta
args.add(ParameterUtils.convertParameterPlaceholders(mainArgs,
ParamUtils.convert(paramsMap)));
}
+ // determine yarn queue
+ determinedYarnQueue(args, flinkParameters, deployMode, flinkVersion);
return args;
}
+
+ private static void determinedYarnQueue(List<String> args, FlinkParameters
flinkParameters,
+ FlinkDeployMode deployMode, String
flinkVersion) {
+ String others = flinkParameters.getOthers();
+ switch (deployMode) {
+ case CLUSTER:
+ if (FLINK_VERSION_AFTER_OR_EQUALS_1_12.equals(flinkVersion)
+ ||
FLINK_VERSION_AFTER_OR_EQUALS_1_13.equals(flinkVersion)) {
+ if (StringUtils.isEmpty(others) ||
!others.contains(FlinkConstants.FLINK_QUEUE_FOR_TARGETS)) {
+ String queue = flinkParameters.getQueue();
+ if (StringUtils.isNotEmpty(queue)) { //
-Dyarn.application.queue=%s
+
args.add(String.format(FlinkConstants.FLINK_QUEUE_FOR_TARGETS, queue));
+ }
+ }
+ } else {
+ if (StringUtils.isEmpty(others) ||
!others.contains(FlinkConstants.FLINK_QUEUE_FOR_MODE)) {
+ String queue = flinkParameters.getQueue();
+ if (StringUtils.isNotEmpty(queue)) { // -yqu
+ args.add(FlinkConstants.FLINK_QUEUE_FOR_MODE);
+ args.add(queue);
+ }
+ }
+ }
+ case APPLICATION:
+ if (StringUtils.isEmpty(others) ||
!others.contains(FlinkConstants.FLINK_QUEUE_FOR_TARGETS)) {
+ String queue = flinkParameters.getQueue();
+ if (StringUtils.isNotEmpty(queue)) { //
-Dyarn.application.queue=%s
+
args.add(String.format(FlinkConstants.FLINK_QUEUE_FOR_TARGETS, queue));
+ }
+ }
Review Comment:
There are some duplicated code. Can we clean up the logic and try to avoid
introducing this new method?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
