seg_yinzy
/
DolphinScheduler
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Rename tis to pigeon from backend layer (#6437) 

* add mock getTaskLogName() for UT test pass

* make task-tis rename to task-pigeon

* rename module name 'tis' to pigeon

* rename module name 'tis' to pigeon,make sanity check pass

* remove execute console.log()
2.0-prepare
百岁 2021-10-01 14:54:04 +08:00 committed by GitHub
parent 6dd0e050ce
commit d284e44d52
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 101 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/TaskType.java
View File

@ -37,6 +37,7 @@ public enum TaskType {
* 11 CONDITIONS
* 12 SQOOP
* 13 WATERDROP
* 15 PIGEON
*/
SHELL(0, "SHELL"),
SQL(1, "SQL"),
@ -53,8 +54,7 @@ public enum TaskType {
SQOOP(12, "SQOOP"),
WATERDROP(13, "WATERDROP"),
SWITCH(14, "SWITCH"),
TIS(15, "TIS"),
;
PIGEON(15, "PIGEON");
TaskType(int code, String desc) {
this.code = code;

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/tis/TISCommonParameters.java → dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/tis/PigeonCommonParameters.java
View File

@ -30,9 +30,9 @@ import org.slf4j.LoggerFactory;
/**
* TIS parameter
*/
public class TISCommonParameters extends AbstractParameters {
public class PigeonCommonParameters extends AbstractParameters {
private static final Logger logger = LoggerFactory.getLogger(TISCommonParameters.class);
private static final Logger logger = LoggerFactory.getLogger(PigeonCommonParameters.class);
/**
* TIS target job name
*/

6
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtils.java
View File

@ -32,7 +32,7 @@ import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import org.apache.dolphinscheduler.common.task.sqoop.SqoopParameters;
import org.apache.dolphinscheduler.common.task.subprocess.SubProcessParameters;
import org.apache.dolphinscheduler.common.task.switchtask.SwitchParameters;
import org.apache.dolphinscheduler.common.task.tis.TISCommonParameters;
import org.apache.dolphinscheduler.common.task.tis.PigeonCommonParameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -86,8 +86,8 @@ public class TaskParametersUtils {
return JSONUtils.parseObject(parameter, SqoopParameters.class);
case "SWITCH":
return JSONUtils.parseObject(parameter, SwitchParameters.class);
case "TIS":
return JSONUtils.parseObject(parameter, TISCommonParameters.class);
case "PIGEON":
return JSONUtils.parseObject(parameter, PigeonCommonParameters.class);
default:
logger.error("not support task type: {}", taskType);
return null;

2
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/TaskParametersUtilsTest.java
View File

@ -41,6 +41,6 @@ public class TaskParametersUtilsTest {
Assert.assertNotNull(TaskParametersUtils.getParameters(TaskType.DEPENDENT.getDesc(), "{}"));
Assert.assertNotNull(TaskParametersUtils.getParameters(TaskType.FLINK.getDesc(), "{}"));
Assert.assertNotNull(TaskParametersUtils.getParameters(TaskType.HTTP.getDesc(), "{}"));
Assert.assertNotNull(TaskParametersUtils.getParameters(TaskType.TIS.getDesc(), "{}"));
Assert.assertNotNull(TaskParametersUtils.getParameters(TaskType.PIGEON.getDesc(), "{}"));
}
}

4
dolphinscheduler-dist/src/main/provisio/dolphinscheduler.xml
View File

@ -106,8 +106,8 @@
<unpack/>
</artifact>
</artifactSet>
<artifactSet to="lib/plugin/task/tis">
<artifact id="${project.groupId}:dolphinscheduler-task-tis:zip:${project.version}">
<artifactSet to="lib/plugin/task/pigeon">
<artifact id="${project.groupId}:dolphinscheduler-task-pigeon:zip:${project.version}">
<unpack/>
</artifact>
</artifactSet>

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumer.java
View File

@ -137,7 +137,7 @@ public class TaskPriorityQueueConsumer extends Thread {
result = dispatcher.dispatch(executionContext);
}
} catch (ExecuteException e) {
logger.error("dispatch error: {}", e.getMessage());
logger.error("dispatch error: {}", e.getMessage(),e);
}
return result;
}

4
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/pom.xml → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/pom.xml
View File

@ -26,7 +26,7 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>dolphinscheduler-task-tis</artifactId>
<artifactId>dolphinscheduler-task-pigeon</artifactId>
<packaging>dolphinscheduler-plugin</packaging>
<dependencies>
@ -107,7 +107,7 @@
</dependencies>
<build>
<finalName>dolphinscheduler-task-tis-${project.version}</finalName>
<finalName>dolphinscheduler-task-pigeon-${project.version}</finalName>
</build>
</project>

0
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/readme.md → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/readme.md
View File

26
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISConfig.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonConfig.java
View File

@ -15,15 +15,15 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.spi.utils.StringUtils;
import java.util.ResourceBundle;
public class TISConfig {
public class PigeonConfig {
private static TISConfig cfg;
private static PigeonConfig cfg;
private final String jobTriggerUrl;
private final String jobTriggerPostBody;
@ -33,15 +33,15 @@ public class TISConfig {
private final String jobLogsFetchUrl;
private final String jobCancelPostBody;
public static synchronized TISConfig getInstance() {
public static synchronized PigeonConfig getInstance() {
if (cfg == null) {
cfg = new TISConfig();
cfg = new PigeonConfig();
}
return cfg;
}
private TISConfig() {
ResourceBundle bundle = ResourceBundle.getBundle(TISConfig.class.getPackage().getName().replace(".", "/") + "/config");
private PigeonConfig() {
ResourceBundle bundle = ResourceBundle.getBundle(PigeonConfig.class.getPackage().getName().replace(".", "/") + "/config");
this.jobTriggerUrl = bundle.getString("job.trigger.url");
this.jobStatusUrl = bundle.getString("job.status.url");
this.jobTriggerPostBody = bundle.getString("job.trigger.post.body");
@ -55,7 +55,7 @@ public class TISConfig {
}
public String getJobTriggerUrl(String tisHost) {
checkTisHost(tisHost);
checkHost(tisHost);
return String.format(this.jobTriggerUrl, tisHost);
}
@ -67,17 +67,17 @@ public class TISConfig {
return String.format(jobStatusPostBody, taskId);
}
public String getJobLogsFetchUrl(String tisHost, String jobName, int taskId) {
checkTisHost(tisHost);
return String.format(jobLogsFetchUrl, tisHost, jobName, taskId);
public String getJobLogsFetchUrl(String host, String jobName, int taskId) {
checkHost(host);
return String.format(jobLogsFetchUrl, host, jobName, taskId);
}
public String getJobStatusUrl(String tisHost) {
checkTisHost(tisHost);
checkHost(tisHost);
return String.format(this.jobStatusUrl, tisHost);
}
private static void checkTisHost(String tisHost) {
private static void checkHost(String tisHost) {
if (StringUtils.isBlank(tisHost)) {
throw new IllegalArgumentException("param tisHost can not be null");
}

8
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParameters.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonParameters.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.spi.task.AbstractParameters;
import org.apache.dolphinscheduler.spi.task.ResourceInfo;
@ -30,11 +30,11 @@ import org.slf4j.LoggerFactory;
/**
* TIS parameter
*/
public class TISParameters extends AbstractParameters {
public class PigeonParameters extends AbstractParameters {
private static final Logger logger = LoggerFactory.getLogger(TISParameters.class);
private static final Logger logger = LoggerFactory.getLogger(PigeonParameters.class);
/**
* TIS target job name
* Pigeon target job name
*/
private String targetJobName;

6
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISParamsConstants.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonParamsConstants.java
View File

@ -15,13 +15,13 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
public class TISParamsConstants {
public class PigeonParamsConstants {
public static String NAME_TARGET_JOB_NAME = "targetJobName";
public static String TARGET_JOB_NAME = NAME_TARGET_JOB_NAME;
private TISParamsConstants() {
private PigeonParamsConstants() {
}
}

66
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTask.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTask.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTaskExecutor;
import org.apache.dolphinscheduler.spi.task.AbstractParameters;
@ -48,51 +48,51 @@ import org.java_websocket.handshake.ServerHandshake;
/**
* TIS DataX Task
**/
public class TISTask extends AbstractTaskExecutor {
public class PigeonTask extends AbstractTaskExecutor {
public static final String KEY_POOL_VAR_TIS_HOST = "tisHost";
public static final String KEY_POOL_VAR_PIGEON_HOST = "p_host";
private final TaskRequest taskExecutionContext;
private TISParameters tisParameters;
private PigeonParameters parameters;
private BizResult triggerResult;
private final TISConfig tisConfig;
private final PigeonConfig config;
public TISTask(TaskRequest taskExecutionContext) {
public PigeonTask(TaskRequest taskExecutionContext) {
super(taskExecutionContext);
this.taskExecutionContext = taskExecutionContext;
this.tisConfig = TISConfig.getInstance();
this.config = PigeonConfig.getInstance();
}
@Override
public void init() {
super.init();
logger.info("tis task params {}", taskExecutionContext.getTaskParams());
tisParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), TISParameters.class);
if (!tisParameters.checkParameters()) {
logger.info("PIGEON task params {}", taskExecutionContext.getTaskParams());
parameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), PigeonParameters.class);
if (!parameters.checkParameters()) {
throw new RuntimeException("datax task params is not valid");
}
}
@Override
public void handle() throws Exception {
// Trigger TIS DataX pipeline
logger.info("start execute TIS task");
// Trigger PIGEON DataX pipeline
logger.info("start execute PIGEON task");
long startTime = System.currentTimeMillis();
String targetJobName = this.tisParameters.getTargetJobName();
String tisHost = getTisHost();
String targetJobName = this.parameters.getTargetJobName();
String host = getHost();
try {
final String triggerUrl = getTriggerUrl();
final String getStatusUrl = tisConfig.getJobStatusUrl(tisHost);
final String getStatusUrl = config.getJobStatusUrl(host);
HttpPost post = new HttpPost(triggerUrl);
post.addHeader("appname", targetJobName);
addFormUrlencoded(post);
StringEntity entity = new StringEntity(tisConfig.getJobTriggerPostBody(), StandardCharsets.UTF_8);
StringEntity entity = new StringEntity(config.getJobTriggerPostBody(), StandardCharsets.UTF_8);
post.setEntity(entity);
ExecResult execState = null;
int taskId;
WebSocketClient webSocket = null;
try (CloseableHttpClient client = HttpClients.createDefault();
// trigger to start TIS dataX task
// trigger to start PIGEON dataX task
CloseableHttpResponse response = client.execute(post)) {
triggerResult = processResponse(triggerUrl, response, BizResult.class);
if (!triggerResult.isSuccess()) {
@ -101,11 +101,11 @@ public class TISTask extends AbstractTaskExecutor {
if (CollectionUtils.isNotEmpty(errormsg)) {
errs.append(",errs:").append(errormsg.stream().collect(Collectors.joining(",")));
}
throw new Exception("trigger TIS job faild taskName:" + targetJobName + errs.toString());
throw new Exception("trigger PIGEON job faild taskName:" + targetJobName + errs.toString());
}
taskId = triggerResult.getBizresult().getTaskid();
webSocket = receiveRealtimeLog(tisHost, targetJobName, taskId);
webSocket = receiveRealtimeLog(host, targetJobName, taskId);
setAppIds(String.valueOf(taskId));
@ -141,11 +141,11 @@ public class TISTask extends AbstractTaskExecutor {
}
long costTime = System.currentTimeMillis() - startTime;
logger.info("TIS task: {},taskId:{} costTime : {} milliseconds, statusCode : {}",
logger.info("PIGEON task: {},taskId:{} costTime : {} milliseconds, statusCode : {}",
targetJobName, taskId, costTime, (execState == ExecResult.SUCCESS) ? "'success'" : "'failure'");
setExitStatusCode((execState == ExecResult.SUCCESS) ? TaskConstants.EXIT_CODE_SUCCESS : TaskConstants.EXIT_CODE_FAILURE);
} catch (Exception e) {
logger.error("execute TIS dataX faild,TIS task name:" + targetJobName, e);
logger.error("execute PIGEON dataX faild,PIGEON task name:" + targetJobName, e);
setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
@ -165,7 +165,7 @@ public class TISTask extends AbstractTaskExecutor {
logger.info("start to cancelApplication taskId:{}", triggerResult.getTaskId());
final String triggerUrl = getTriggerUrl();
StringEntity entity = new StringEntity(tisConfig.getJobCancelPostBody(triggerResult.getTaskId()), StandardCharsets.UTF_8);
StringEntity entity = new StringEntity(config.getJobCancelPostBody(triggerResult.getTaskId()), StandardCharsets.UTF_8);
CancelResult cancelResult = null;
HttpPost post = new HttpPost(triggerUrl);
@ -181,26 +181,26 @@ public class TISTask extends AbstractTaskExecutor {
if (org.apache.dolphinscheduler.spi.utils.CollectionUtils.isNotEmpty(errormsg)) {
errs.append(",errs:").append(errormsg.stream().collect(Collectors.joining(",")));
}
throw new Exception("cancel TIS job faild taskId:" + triggerResult.getTaskId() + errs.toString());
throw new Exception("cancel PIGEON job faild taskId:" + triggerResult.getTaskId() + errs.toString());
}
}
}
private String getTriggerUrl() {
final String tisHost = getTisHost();
return tisConfig.getJobTriggerUrl(tisHost);
final String tisHost = getHost();
return config.getJobTriggerUrl(tisHost);
}
private String getTisHost() {
final String tisHost = taskExecutionContext.getDefinedParams().get(KEY_POOL_VAR_TIS_HOST);
if (StringUtils.isEmpty(tisHost)) {
throw new IllegalStateException("global var '" + KEY_POOL_VAR_TIS_HOST + "' can not be empty");
private String getHost() {
final String host = taskExecutionContext.getDefinedParams().get(KEY_POOL_VAR_PIGEON_HOST);
if (StringUtils.isEmpty(host)) {
throw new IllegalStateException("global var '" + KEY_POOL_VAR_PIGEON_HOST + "' can not be empty");
}
return tisHost;
return host;
}
private WebSocketClient receiveRealtimeLog(final String tisHost, String dataXName, int taskId) throws Exception {
final String applyURI = tisConfig.getJobLogsFetchUrl(tisHost, dataXName, taskId);
final String applyURI = config.getJobLogsFetchUrl(tisHost, dataXName, taskId);
logger.info("apply ws connection,uri:{}", applyURI);
WebSocketClient webSocketClient = new WebSocketClient(new URI(applyURI)) {
@Override
@ -241,8 +241,8 @@ public class TISTask extends AbstractTaskExecutor {
@Override
public AbstractParameters getParameters() {
Objects.requireNonNull(this.tisParameters, "tisParameters can not be null");
return this.tisParameters;
Objects.requireNonNull(this.parameters, "tisParameters can not be null");
return this.parameters;
}
private static class CancelResult extends AjaxResult<Object> {

13
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannel.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskChannel.java
View File

@ -15,20 +15,25 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.spi.task.AbstractTask;
import org.apache.dolphinscheduler.spi.task.TaskChannel;
import org.apache.dolphinscheduler.spi.task.request.TaskRequest;
public class TISTaskChannel implements TaskChannel {
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class PigeonTaskChannel implements TaskChannel {
private static final Logger logger = LoggerFactory.getLogger(PigeonTaskChannel.class);
@Override
public void cancelApplication(boolean status) {
logger.info("pigeon task cancel");
}
@Override
public AbstractTask createTask(TaskRequest taskRequest) {
return new TISTask(taskRequest);
return new PigeonTask(taskRequest);
}
}

14
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskChannelFactory.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskChannelFactory.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.params.base.Validate;
@ -26,24 +26,22 @@ import org.apache.dolphinscheduler.spi.task.TaskChannelFactory;
import java.util.Arrays;
import java.util.List;
/**
* TIS endpoint
**/
public class TISTaskChannelFactory implements TaskChannelFactory {
public class PigeonTaskChannelFactory implements TaskChannelFactory {
@Override
public TaskChannel create() {
return new TISTaskChannel();
return new PigeonTaskChannel();
}
@Override
public String getName() {
return "TIS";
return "PIGEON";
}
@Override
public List<PluginParams> getParams() {
InputParam webHookParam = InputParam.newBuilder(TISParamsConstants.NAME_TARGET_JOB_NAME, TISParamsConstants.TARGET_JOB_NAME)
InputParam webHookParam
= InputParam.newBuilder(PigeonParamsConstants.NAME_TARGET_JOB_NAME, PigeonParamsConstants.TARGET_JOB_NAME)
.addValidate(Validate.newBuilder().setRequired(true).build())
.build();
return Arrays.asList(webHookParam);

6
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskPlugin.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskPlugin.java
View File

@ -15,17 +15,17 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import org.apache.dolphinscheduler.spi.DolphinSchedulerPlugin;
import org.apache.dolphinscheduler.spi.task.TaskChannelFactory;
import com.google.common.collect.ImmutableList;
public class TISTaskPlugin implements DolphinSchedulerPlugin {
public class PigeonTaskPlugin implements DolphinSchedulerPlugin {
@Override
public Iterable<TaskChannelFactory> getTaskChannelFactorys() {
return ImmutableList.of(new TISTaskChannelFactory());
return ImmutableList.of(new PigeonTaskChannelFactory());
}
}

0
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/main/resources/org/apache/dolphinscheduler/plugin/task/tis/config.properties → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/main/resources/org/apache/dolphinscheduler/plugin/task/pigeon/config.properties
View File

27
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/java/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.java → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/test/java/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskTest.java
View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.plugin.task.tis;
package org.apache.dolphinscheduler.plugin.task.pigeon;
import static com.github.dreamhead.moco.Moco.file;
import static com.github.dreamhead.moco.MocoJsonRunner.jsonHttpServer;
@ -38,15 +38,14 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.dreamhead.moco.HttpServer;
public class TISTaskTest {
private static final Logger logger = LoggerFactory.getLogger(TISTaskTest.class);
private TISTask tisTask;
public class PigeonTaskTest {
private static final Logger logger = LoggerFactory.getLogger(PigeonTaskTest.class);
private PigeonTask pigeonTask;
private TaskRequest taskExecutionContext;
@ -56,7 +55,7 @@ public class TISTaskTest {
String taskParams = "{\"targetJobName\":\"mysql_elastic\"}";
taskExecutionContext = Mockito.mock(TaskRequest.class);
Mockito.when(taskExecutionContext.getTaskLogName()).thenReturn("tislogger");
Mockito.when(taskExecutionContext.getTaskLogName()).thenReturn("pigeonlogger");
Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(taskParams);
Mockito.when(taskExecutionContext.getExecutePath()).thenReturn("/tmp");
Mockito.when(taskExecutionContext.getTaskAppId()).thenReturn(UUID.randomUUID().toString());
@ -66,17 +65,17 @@ public class TISTaskTest {
Mockito.when(taskExecutionContext.getLogPath()).thenReturn("/tmp/dx");
// Mockito.when(taskExecutionContext.getVarPool())
// .thenReturn("[{\"direct\":\"IN\",\"prop\":\"" + TISTask.KEY_POOL_VAR_TIS_HOST + "\",\"type\":\"VARCHAR\",\"value\":\"127.0.0.1:8080\"}]");
Map<String, String> gloabParams = Collections.singletonMap(TISTask.KEY_POOL_VAR_TIS_HOST, "127.0.0.1:8080");
Map<String, String> gloabParams = Collections.singletonMap(PigeonTask.KEY_POOL_VAR_PIGEON_HOST, "127.0.0.1:8080");
Mockito.when(taskExecutionContext.getDefinedParams()).thenReturn(gloabParams);
tisTask = new TISTask(taskExecutionContext);
tisTask.init();
pigeonTask = new PigeonTask(taskExecutionContext);
pigeonTask.init();
}
@Test
public void testGetTISConfigParams() {
TISConfig cfg = TISConfig.getInstance();
PigeonConfig cfg = PigeonConfig.getInstance();
String tisHost = "127.0.0.1:8080";
Assert.assertEquals("http://127.0.0.1:8080/tjs/coredefine/coredefine.ajax", cfg.getJobTriggerUrl(tisHost));
String jobName = "mysql_elastic";
@ -97,7 +96,7 @@ public class TISTaskTest {
public void testInit()
throws Exception {
try {
tisTask.init();
pigeonTask.init();
} catch (Exception e) {
Assert.fail(e.getMessage());
}
@ -106,12 +105,12 @@ public class TISTaskTest {
@Test
public void testHandle()
throws Exception {
HttpServer server = jsonHttpServer(8080, file("src/test/resources/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json"));
HttpServer server = jsonHttpServer(8080, file("src/test/resources/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskTest.json"));
running(server, () -> {
tisTask.handle();
pigeonTask.handle();
Assert.assertEquals("TIS execute be success", ExecutionStatus.SUCCESS, tisTask.getExitStatus());
Assert.assertEquals("PIGEON execute be success", ExecutionStatus.SUCCESS, pigeonTask.getExitStatus());
});
}

4
dolphinscheduler-task-plugin/dolphinscheduler-task-tis/src/test/resources/org/apache/dolphinscheduler/plugin/task/tis/TISTaskTest.json → dolphinscheduler-task-plugin/dolphinscheduler-task-pigeon/src/test/resources/org/apache/dolphinscheduler/plugin/task/pigeon/PigeonTaskTest.json
View File

@ -1,6 +1,6 @@
[
{
"description": "trigger TIS task execute",
"description": "trigger task execute",
"request": {
"uri": "/tjs/coredefine/coredefine.ajax",
"method": "post",
@ -15,7 +15,7 @@
}
},
{
"description": "Get TIS task execute status",
"description": "Get task execute status",
"request": {
"uri": "/tjs/config/config.ajax",
"method": "post",

2
dolphinscheduler-task-plugin/pom.xml
View File

@ -40,7 +40,7 @@
<module>dolphinscheduler-task-sql</module>
<module>dolphinscheduler-task-sqoop</module>
<module>dolphinscheduler-task-procedure</module>
<module>dolphinscheduler-task-tis</module>
<module>dolphinscheduler-task-pigeon</module>
</modules>

BIN
dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/images/task-icos/tis.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 1.3 KiB

BIN
dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/images/task-icos/tis_hover.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 1.4 KiB

2
pom.xml
View File

@ -1103,7 +1103,7 @@
<include>**/alert/processor/AlertRequestProcessorTest.java</include>
<include>**/alert/runner/AlertSenderTest.java</include>
<include>**/alert/AlertServerTest.java</include>
<include>**/plugin/task/tis/TISTaskTest.java</include>
<include>**/plugin/task/pigeon/PigeonTaskTest.java</include>
</includes>
<!-- <skip>true</skip> -->
</configuration>