Browse Source

[Bug-13010] [Task] The Flink SQL task page selects the pre-job deployment mode, but the task executed by the worker is the Flink local mode (#13011)

3.2.0-release
Kerwin 2 years ago committed by GitHub
parent
commit
ae670e6e1c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 8
      dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java
  2. 4
      dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java
  3. 1
      dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkConstants.java
  4. 8
      dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java
  5. 12
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink.ts

8
dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java

@ -115,18 +115,18 @@ public class FlinkArgsUtilsTest {
} }
@Test @Test
public void testInitOptionsInClusterMode() throws Exception { public void testInitOptionsInLocalMode() throws Exception {
List<String> initOptions = List<String> initOptions =
FlinkArgsUtils.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.CLUSTER)); FlinkArgsUtils.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.LOCAL));
Assertions.assertEquals(2, initOptions.size()); Assertions.assertEquals(2, initOptions.size());
Assertions.assertTrue(initOptions.contains("set execution.target=local")); Assertions.assertTrue(initOptions.contains("set execution.target=local"));
Assertions.assertTrue(initOptions.contains("set parallelism.default=4")); Assertions.assertTrue(initOptions.contains("set parallelism.default=4"));
} }
@Test @Test
public void testInitOptionsInApplicationMode() throws Exception { public void testInitOptionsInClusterMode() throws Exception {
List<String> initOptions = FlinkArgsUtils List<String> initOptions = FlinkArgsUtils
.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.APPLICATION)); .buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.CLUSTER));
Assertions.assertEquals(6, initOptions.size()); Assertions.assertEquals(6, initOptions.size());
Assertions.assertTrue(initOptions.contains("set execution.target=yarn-per-job")); Assertions.assertTrue(initOptions.contains("set execution.target=yarn-per-job"));
Assertions.assertTrue(initOptions.contains("set taskmanager.numberOfTaskSlots=4")); Assertions.assertTrue(initOptions.contains("set taskmanager.numberOfTaskSlots=4"));

4
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java

@ -126,9 +126,9 @@ public class FlinkArgsUtils {
/** /**
* Currently flink sql on yarn only supports yarn-per-job mode * Currently flink sql on yarn only supports yarn-per-job mode
*/ */
if (FlinkDeployMode.CLUSTER == deployMode) { if (FlinkDeployMode.LOCAL == deployMode) {
// execution.target // execution.target
initOptions.add(String.format(FlinkConstants.FLINK_FORMAT_EXECUTION_TARGET, "local")); initOptions.add(String.format(FlinkConstants.FLINK_FORMAT_EXECUTION_TARGET, FlinkConstants.FLINK_LOCAL));
} else { } else {
// execution.target // execution.target
initOptions.add( initOptions.add(

1
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkConstants.java

@ -43,6 +43,7 @@ public class FlinkConstants {
public static final String FLINK_YARN_CLUSTER = "yarn-cluster"; public static final String FLINK_YARN_CLUSTER = "yarn-cluster";
public static final String FLINK_YARN_APPLICATION = "yarn-application"; public static final String FLINK_YARN_APPLICATION = "yarn-application";
public static final String FLINK_YARN_PER_JOB = "yarn-per-job"; public static final String FLINK_YARN_PER_JOB = "yarn-per-job";
public static final String FLINK_LOCAL = "local";
public static final String FLINK_RUN_MODE = "-m"; public static final String FLINK_RUN_MODE = "-m";
public static final String FLINK_EXECUTION_TARGET = "-t"; public static final String FLINK_EXECUTION_TARGET = "-t";
public static final String FLINK_YARN_SLOT = "-ys"; public static final String FLINK_YARN_SLOT = "-ys";

8
dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/test/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtilsTest.java

@ -115,18 +115,18 @@ public class FlinkArgsUtilsTest {
} }
@Test @Test
public void testInitOptionsInClusterMode() throws Exception { public void testInitOptionsInLocalMode() throws Exception {
List<String> initOptions = List<String> initOptions =
FlinkArgsUtils.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.CLUSTER)); FlinkArgsUtils.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.LOCAL));
Assertions.assertEquals(2, initOptions.size()); Assertions.assertEquals(2, initOptions.size());
Assertions.assertTrue(initOptions.contains("set execution.target=local")); Assertions.assertTrue(initOptions.contains("set execution.target=local"));
Assertions.assertTrue(initOptions.contains("set parallelism.default=4")); Assertions.assertTrue(initOptions.contains("set parallelism.default=4"));
} }
@Test @Test
public void testInitOptionsInApplicationMode() throws Exception { public void testInitOptionsInClusterMode() throws Exception {
List<String> initOptions = FlinkArgsUtils List<String> initOptions = FlinkArgsUtils
.buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.APPLICATION)); .buildInitOptionsForSql(buildTestFlinkParametersWithDeployMode(FlinkDeployMode.CLUSTER));
Assertions.assertEquals(6, initOptions.size()); Assertions.assertEquals(6, initOptions.size());
Assertions.assertTrue(initOptions.contains("set execution.target=yarn-per-job")); Assertions.assertTrue(initOptions.contains("set execution.target=yarn-per-job"));
Assertions.assertTrue(initOptions.contains("set taskmanager.numberOfTaskSlots=4")); Assertions.assertTrue(initOptions.contains("set taskmanager.numberOfTaskSlots=4"));

12
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink.ts

@ -44,6 +44,18 @@ export function useFlink(model: { [field: string]: any }): IJsonItem[] {
const appNameSpan = computed(() => (model.deployMode !== 'local' ? 24 : 0)) const appNameSpan = computed(() => (model.deployMode !== 'local' ? 24 : 0))
const deployModeOptions = computed(() => { const deployModeOptions = computed(() => {
if (model.programType === 'SQL') {
return [
{
label: 'per-job/cluster',
value: 'cluster'
},
{
label: 'local',
value: 'local'
}
]
}
if (model.flinkVersion === '<1.10') { if (model.flinkVersion === '<1.10') {
return [ return [
{ {

Loading…
Cancel
Save