diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtils.java index 6099a0d49d..2b40b079c6 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtils.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtils.java @@ -59,9 +59,8 @@ public class TaskParametersUtils { switch (EnumUtils.getEnum(TaskType.class, taskType)) { case SUB_PROCESS: return JSONUtils.parseObject(parameter, SubProcessParameters.class); - case WATERDROP: - return JSONUtils.parseObject(parameter, ShellParameters.class); case SHELL: + case WATERDROP: return JSONUtils.parseObject(parameter, ShellParameters.class); case PROCEDURE: return JSONUtils.parseObject(parameter, ProcedureParameters.class); diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/TaskManager.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/TaskManager.java index f98d451350..34eea9dade 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/TaskManager.java +++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/TaskManager.java @@ -14,8 +14,8 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.dolphinscheduler.server.worker.task; +package org.apache.dolphinscheduler.server.worker.task; import org.apache.dolphinscheduler.common.enums.TaskType; import org.apache.dolphinscheduler.common.utils.EnumUtils; @@ -30,6 +30,7 @@ import org.apache.dolphinscheduler.server.worker.task.shell.ShellTask; import org.apache.dolphinscheduler.server.worker.task.spark.SparkTask; import org.apache.dolphinscheduler.server.worker.task.sql.SqlTask; import org.apache.dolphinscheduler.server.worker.task.sqoop.SqoopTask; + import org.slf4j.Logger; /** @@ -37,42 +38,39 @@ import org.slf4j.Logger; */ public class TaskManager { - /** - * create new task - * @param taskExecutionContext taskExecutionContext - * @param logger logger - * @return AbstractTask - * @throws IllegalArgumentException illegal argument exception - */ - public static AbstractTask newTask(TaskExecutionContext taskExecutionContext, - Logger logger) - throws IllegalArgumentException { - switch (EnumUtils.getEnum(TaskType.class,taskExecutionContext.getTaskType())) { - case SHELL: - return new ShellTask(taskExecutionContext, logger); - case WATERDROP: - return new ShellTask(taskExecutionContext, logger); - case PROCEDURE: - return new ProcedureTask(taskExecutionContext, logger); - case SQL: - return new SqlTask(taskExecutionContext, logger); - case MR: - return new MapReduceTask(taskExecutionContext, logger); - case SPARK: - return new SparkTask(taskExecutionContext, logger); - case FLINK: - return new FlinkTask(taskExecutionContext, logger); - case PYTHON: - return new PythonTask(taskExecutionContext, logger); - case HTTP: - return new HttpTask(taskExecutionContext, logger); - case DATAX: - return new DataxTask(taskExecutionContext, logger); - case SQOOP: - return new SqoopTask(taskExecutionContext, logger); - default: - logger.error("unsupport task type: {}", taskExecutionContext.getTaskType()); - throw new IllegalArgumentException("not support task type"); + /** + * create new task + * @param taskExecutionContext taskExecutionContext + * @param logger logger + * @return AbstractTask + * @throws IllegalArgumentException illegal argument exception + */ + public static AbstractTask newTask(TaskExecutionContext taskExecutionContext, Logger logger) throws IllegalArgumentException { + switch (EnumUtils.getEnum(TaskType.class,taskExecutionContext.getTaskType())) { + case SHELL: + case WATERDROP: + return new ShellTask(taskExecutionContext, logger); + case PROCEDURE: + return new ProcedureTask(taskExecutionContext, logger); + case SQL: + return new SqlTask(taskExecutionContext, logger); + case MR: + return new MapReduceTask(taskExecutionContext, logger); + case SPARK: + return new SparkTask(taskExecutionContext, logger); + case FLINK: + return new FlinkTask(taskExecutionContext, logger); + case PYTHON: + return new PythonTask(taskExecutionContext, logger); + case HTTP: + return new HttpTask(taskExecutionContext, logger); + case DATAX: + return new DataxTask(taskExecutionContext, logger); + case SQOOP: + return new SqoopTask(taskExecutionContext, logger); + default: + logger.error("unsupport task type: {}", taskExecutionContext.getTaskType()); + throw new IllegalArgumentException("not support task type"); + } } - } } diff --git a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskManagerTest.java b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskManagerTest.java new file mode 100644 index 0000000000..058270e1e1 --- /dev/null +++ b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskManagerTest.java @@ -0,0 +1,103 @@ +/* + * 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.worker.task; + +import org.apache.dolphinscheduler.common.utils.LoggerUtils; +import org.apache.dolphinscheduler.server.entity.TaskExecutionContext; +import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl; +import org.apache.dolphinscheduler.service.bean.SpringApplicationContext; + +import java.util.Date; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@RunWith(PowerMockRunner.class) +@PrepareForTest({SpringApplicationContext.class}) +public class TaskManagerTest { + + private TaskExecutionContext taskExecutionContext; + + private Logger taskLogger; + + private TaskExecutionContextCacheManagerImpl taskExecutionContextCacheManager; + + @Before + public void before() { + // init task execution context, logger + taskExecutionContext = new TaskExecutionContext(); + taskExecutionContext.setProcessId(12345); + taskExecutionContext.setProcessDefineId(1); + taskExecutionContext.setProcessInstanceId(1); + taskExecutionContext.setTaskInstanceId(1); + taskExecutionContext.setTaskType(""); + taskExecutionContext.setFirstSubmitTime(new Date()); + taskExecutionContext.setDelayTime(0); + taskExecutionContext.setLogPath("/tmp/test.log"); + taskExecutionContext.setHost("localhost"); + taskExecutionContext.setExecutePath("/tmp/dolphinscheduler/exec/process/1/2/3/4"); + + taskLogger = LoggerFactory.getLogger(LoggerUtils.buildTaskId( + LoggerUtils.TASK_LOGGER_INFO_PREFIX, + taskExecutionContext.getProcessDefineId(), + taskExecutionContext.getProcessInstanceId(), + taskExecutionContext.getTaskInstanceId() + )); + + taskExecutionContextCacheManager = new TaskExecutionContextCacheManagerImpl(); + taskExecutionContextCacheManager.cacheTaskExecutionContext(taskExecutionContext); + + PowerMockito.mockStatic(SpringApplicationContext.class); + PowerMockito.when(SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class)) + .thenReturn(taskExecutionContextCacheManager); + } + + @Test + public void testNewTask() { + + taskExecutionContext.setTaskType("SHELL"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("WATERDROP"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("HTTP"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("MR"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("SPARK"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("FLINK"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("PYTHON"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("DATAX"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + taskExecutionContext.setTaskType("SQOOP"); + Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + //taskExecutionContext.setTaskType(null); + //Assert.assertNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + //taskExecutionContext.setTaskType("XXX"); + //Assert.assertNotNull(TaskManager.newTask(taskExecutionContext,taskLogger)); + } +} diff --git a/pom.xml b/pom.xml index 65cbe62bbb..207518c936 100644 --- a/pom.xml +++ b/pom.xml @@ -847,6 +847,7 @@ **/server/worker/task/sqoop/SqoopTaskTest.java + **/server/worker/task/TaskManagerTest.java **/server/worker/EnvFileTest.java **/server/worker/runner/TaskExecuteThreadTest.java **/service/quartz/cron/CronUtilsTest.java