From 0f909fca12fefeb323bef5cf4436230749d38b02 Mon Sep 17 00:00:00 2001 From: zhukai Date: Thu, 26 Dec 2019 14:15:58 +0800 Subject: [PATCH] Add FlinkArgsUtilsTest which is the UT of FlinkArgsUtils (#1557) 1.Add FlinkArgsUtilsTest 2.Optimize FlinkArgsUtils and fix bug. use StringUtils.isNotEmpty(queue) instead of StringUtils.isNotEmpty(others) 3.Add the UT path in pom --- .../server/utils/FlinkArgsUtils.java | 75 +++++----- .../server/utils/FlinkArgsUtilsTest.java | 131 ++++++++++++++++++ pom.xml | 1 + 3 files changed, 172 insertions(+), 35 deletions(-) create mode 100644 dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java index b6fcb2b40..ab34ddfc2 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtils.java @@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.server.utils; import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.enums.ProgramType; +import org.apache.dolphinscheduler.common.process.ResourceInfo; import org.apache.dolphinscheduler.common.task.flink.FlinkParameters; import org.apache.commons.lang.StringUtils; import org.slf4j.LoggerFactory; @@ -44,9 +45,11 @@ public class FlinkArgsUtils { */ public static List buildArgs(FlinkParameters param) { List args = new ArrayList<>(); + String deployMode = "cluster"; - if (StringUtils.isNotEmpty(param.getDeployMode())) { - deployMode = param.getDeployMode(); + String tmpDeployMode = param.getDeployMode(); + if (StringUtils.isNotEmpty(tmpDeployMode)) { + deployMode = tmpDeployMode; } if (!"local".equals(deployMode)) { @@ -54,68 +57,70 @@ public class FlinkArgsUtils { args.add(Constants.FLINK_YARN_CLUSTER); //yarn-cluster - - if (param.getSlot() != 0) { + int slot = param.getSlot(); + if (slot != 0) { args.add(Constants.FLINK_YARN_SLOT); - args.add(String.format("%d", param.getSlot())); //-ys + args.add(String.format("%d", slot)); //-ys } - if (StringUtils.isNotEmpty(param.getAppName())) { //-ynm + String appName = param.getAppName(); + if (StringUtils.isNotEmpty(appName)) { //-ynm args.add(Constants.FLINK_APP_NAME); - args.add(param.getAppName()); + args.add(appName); } - if (param.getTaskManager() != 0) { //-yn + int taskManager = param.getTaskManager(); + if (taskManager != 0) { //-yn args.add(Constants.FLINK_TASK_MANAGE); - args.add(String.format("%d", param.getTaskManager())); + args.add(String.format("%d", taskManager)); } - if (StringUtils.isNotEmpty(param.getJobManagerMemory())) { + String jobManagerMemory = param.getJobManagerMemory(); + if (StringUtils.isNotEmpty(jobManagerMemory)) { args.add(Constants.FLINK_JOB_MANAGE_MEM); - args.add(param.getJobManagerMemory()); //-yjm + args.add(jobManagerMemory); //-yjm } - if (StringUtils.isNotEmpty(param.getTaskManagerMemory())) { // -ytm + String taskManagerMemory = param.getTaskManagerMemory(); + if (StringUtils.isNotEmpty(taskManagerMemory)) { // -ytm args.add(Constants.FLINK_TASK_MANAGE_MEM); - args.add(param.getTaskManagerMemory()); + args.add(taskManagerMemory); } args.add(Constants.FLINK_detach); //-d - } - if (param.getProgramType() != null) { - if (param.getProgramType() != ProgramType.PYTHON) { - if (StringUtils.isNotEmpty(param.getMainClass())) { - args.add(Constants.FLINK_MAIN_CLASS); //-c - args.add(param.getMainClass()); //main class - } - } + ProgramType programType = param.getProgramType(); + String mainClass = param.getMainClass(); + if (programType != null && programType != ProgramType.PYTHON && StringUtils.isNotEmpty(mainClass)) { + args.add(Constants.FLINK_MAIN_CLASS); //-c + args.add(param.getMainClass()); //main class } - if (param.getMainJar() != null) { - args.add(param.getMainJar().getRes()); + ResourceInfo mainJar = param.getMainJar(); + if (mainJar != null) { + args.add(mainJar.getRes()); } - if (StringUtils.isNotEmpty(param.getMainArgs())) { - args.add(param.getMainArgs()); + String mainArgs = param.getMainArgs(); + if (StringUtils.isNotEmpty(mainArgs)) { + args.add(mainArgs); } // --files --conf --libjar ... - if (StringUtils.isNotEmpty(param.getOthers())) { - String others = param.getOthers(); - if (!others.contains("--qu")) { - if (StringUtils.isNotEmpty(param.getQueue()) && !deployMode.equals("local")) { - args.add(Constants.FLINK_QUEUE); - args.add(param.getQueue()); - } + String others = param.getOthers(); + String queue = param.getQueue(); + if (StringUtils.isNotEmpty(others)) { + + if (!others.contains(Constants.FLINK_QUEUE) && StringUtils.isNotEmpty(queue) && !deployMode.equals("local")) { + args.add(Constants.FLINK_QUEUE); + args.add(param.getQueue()); } - args.add(param.getOthers()); - } else if (StringUtils.isNotEmpty(param.getQueue()) && !deployMode.equals("local")) { + args.add(others); + } else if (StringUtils.isNotEmpty(queue) && !deployMode.equals("local")) { args.add(Constants.FLINK_QUEUE); args.add(param.getQueue()); - } return args; diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java new file mode 100644 index 000000000..710d2c250 --- /dev/null +++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/FlinkArgsUtilsTest.java @@ -0,0 +1,131 @@ +/* + * 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.dolphinscheduler.server.utils; + +import org.apache.dolphinscheduler.common.enums.ProgramType; +import org.apache.dolphinscheduler.common.process.ResourceInfo; +import org.apache.dolphinscheduler.common.task.flink.FlinkParameters; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +/** + * Test FlinkArgsUtils + */ +public class FlinkArgsUtilsTest { + + private static final Logger logger = LoggerFactory.getLogger(FlinkArgsUtilsTest.class); + + public String mode = "cluster"; + public int slot = 2; + public String appName = "testFlink"; + public int taskManager = 4; + public String taskManagerMemory = "2G"; + public String jobManagerMemory = "4G"; + public ProgramType programType = ProgramType.JAVA; + public String mainClass = "com.test"; + public ResourceInfo mainJar = null; + public String mainArgs = "testArgs"; + public String queue = "queue1"; + public String others = "--input file:///home"; + + + @Before + public void setUp() throws Exception { + + ResourceInfo main = new ResourceInfo(); + main.setRes("testflink-1.0.0-SNAPSHOT.jar"); + mainJar = main; + } + + /** + * Test buildArgs + */ + @Test + public void testBuildArgs() { + + //Define params + FlinkParameters param = new FlinkParameters(); + param.setDeployMode(mode); + param.setMainClass(mainClass); + param.setAppName(appName); + param.setSlot(slot); + param.setTaskManager(taskManager); + param.setJobManagerMemory(jobManagerMemory); + param.setTaskManagerMemory(taskManagerMemory); + param.setMainJar(mainJar); + param.setProgramType(programType); + param.setMainArgs(mainArgs); + param.setQueue(queue); + param.setOthers(others); + + //Invoke buildArgs + List result = FlinkArgsUtils.buildArgs(param); + for (String s : result) { + logger.info(s); + } + + //Expected values and order + assertEquals(result.size(),20); + + assertEquals(result.get(0),"-m"); + assertEquals(result.get(1),"yarn-cluster"); + + assertEquals(result.get(2),"-ys"); + assertSame(Integer.valueOf(result.get(3)),slot); + + assertEquals(result.get(4),"-ynm"); + assertEquals(result.get(5),appName); + + assertEquals(result.get(6),"-yn"); + assertSame(Integer.valueOf(result.get(7)),taskManager); + + assertEquals(result.get(8),"-yjm"); + assertEquals(result.get(9),jobManagerMemory); + + assertEquals(result.get(10),"-ytm"); + assertEquals(result.get(11),taskManagerMemory); + + assertEquals(result.get(12),"-d"); + + assertEquals(result.get(13),"-c"); + assertEquals(result.get(14),mainClass); + + assertEquals(result.get(15),mainJar.getRes()); + assertEquals(result.get(16),mainArgs); + + assertEquals(result.get(17),"--qu"); + assertEquals(result.get(18),queue); + + assertEquals(result.get(19),others); + + //Others param without --qu + FlinkParameters param1 = new FlinkParameters(); + param1.setQueue(queue); + param1.setDeployMode(mode); + result = FlinkArgsUtils.buildArgs(param1); + assertEquals(result.size(),5); + + } +} \ No newline at end of file diff --git a/pom.xml b/pom.xml index e0f4e9303..5b80ee1e3 100644 --- a/pom.xml +++ b/pom.xml @@ -655,6 +655,7 @@ **/alert/utils/JSONUtilsTest.java **/alert/utils/PropertyUtilsTest.java **/server/utils/SparkArgsUtilsTest.java + **/server/utils/FlinkArgsUtilsTest.java -- GitLab