zixi0825 commented on code in PR #16863: URL: https://github.com/apache/dolphinscheduler/pull/16863#discussion_r3791730563
########## dolphinscheduler-task-plugin/dolphinscheduler-task-datavines/src/main/java/org/apache/dolphinscheduler/plugin/task/datavines/utils/RequestUtils.java: ########## @@ -0,0 +1,123 @@ +/* + * 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.plugin.task.datavines.utils; + +import org.apache.dolphinscheduler.plugin.task.datavines.DatavinesTaskConstants; + +import org.apache.http.HttpResponse; +import org.apache.http.HttpStatus; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.utils.URIBuilder; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.util.EntityUtils; + +import java.net.URI; + +import lombok.extern.slf4j.Slf4j; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +@Slf4j +public class RequestUtils { + + private RequestUtils() { + throw new IllegalStateException("Utility class"); + } + + public static JsonNode executeJob(String address, String jobId, String token) { + return parse(doPost(address + DatavinesTaskConstants.EXECUTE_JOB + jobId, token)); + } + + public static JsonNode getJobExecutionStatus(String address, String jobExecutionId, String token) { + return parse(doGet(address + DatavinesTaskConstants.GET_JOB_EXECUTION_STATUS + jobExecutionId, token)); + } + + public static JsonNode getJobExecutionResult(String address, String jobExecutionId, String token) { + return parse(doGet(address + DatavinesTaskConstants.GET_JOB_EXECUTION_RESULT + jobExecutionId, token)); + } + + public static void killJobExecution(String address, String jobExecutionId, String token) { + parse(doPost(address + DatavinesTaskConstants.JOB_EXECUTION_KILL + jobExecutionId, token)); + } + + public static JsonNode parse(String res) { + ObjectMapper mapper = new ObjectMapper(); + JsonNode result = null; + try { + result = mapper.readTree(res); + } catch (JsonProcessingException e) { + log.error("datavines task submit failed with error", e); + } + return result; + } + + public static String doGet(String url, String token) { + String result = ""; + HttpClient httpClient = HttpClientBuilder.create().build(); + HttpGet httpGet = null; + try { + URIBuilder uriBuilder = new URIBuilder(url); + URI uri = uriBuilder.build(); + httpGet = new HttpGet(uri); + httpGet.setHeader("Authorization", "Bearer " + token); + log.info("access url: {}", uri); + HttpResponse response = httpClient.execute(httpGet); + if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) { + result = EntityUtils.toString(response.getEntity()); + log.info("datavines task succeed with results: {}", result); + } else { + log.error("datavines task terminated,response: {}", response); + } + } catch (IllegalArgumentException ie) { + log.error("datavines task terminated: {}", ie.getMessage()); + } catch (Exception e) { + log.error("datavines task terminated: ", e); + } finally { + if (null != httpGet) { + httpGet.releaseConnection(); + } + } + return result; Review Comment: Thanks. `RequestUtils` now uses a shared static `CloseableHttpClient` and closes the response with try-with-resources, so the polling in `trackApplicationStatus` no longer leaks connections (commit 9b02337514). ########## dolphinscheduler-task-plugin/dolphinscheduler-task-datavines/src/test/java/org/apache/dolphinscheduler/plugin/task/datavines/DatavinesTaskTest.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.plugin.task.datavines; + +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE; +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS; +import static org.junit.Assert.assertThrows; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.mockito.Mockito.when; + +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.datavines.utils.RequestUtils; + +import java.lang.reflect.Field; +import java.util.stream.Stream; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; + +import com.fasterxml.jackson.databind.JsonNode; + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +class DatavinesTaskTest { + + @Mock + private TaskExecutionContext taskExecutionContext; + + private DatavinesTask datavinesTask; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + datavinesTask = new DatavinesTask(taskExecutionContext); + } + + @Test + void initValidParametersInitializesSuccessfully() { + when(taskExecutionContext.getTaskParams()) + .thenReturn("{\"address\":\"http://localhost\",\"jobId\":\"1\",\"token\":\"token\"}";); + datavinesTask.init(); + Assertions.assertNotNull(datavinesTask.getParameters()); + } + + @Test + void initInvalidParametersThrowsException() { + when(taskExecutionContext.getTaskParams()).thenReturn("{}"); + assertThrows(DatavinesTaskException.class, () -> datavinesTask.init()); + } + + @Test + void submitApplicationExecutesJobSuccessfully() { + when(taskExecutionContext.getTaskParams()) + .thenReturn("{\"address\":\"http://localhost\",\"jobId\":\"1\",\"token\":\"token\"}";); + datavinesTask.init(); + assertDoesNotThrow(() -> datavinesTask.submitApplication()); + } + + @Test + void trackApplicationStatusJobExecutionSuccessSetsExitCodeSuccess() throws TaskException { + JsonNode executeJobResult = RequestUtils.parse("{\"code\":\"200\",\"data\":\"1\"}"); + JsonNode executeStatus = RequestUtils.parse("{\"code\":\"200\",\"data\":\"SUCCESS\"}"); + JsonNode executeResult = RequestUtils.parse("{\"code\":\"200\",\"data\":\"1\"}"); + try (MockedStatic<RequestUtils> requestUtilsStatic = Mockito.mockStatic(RequestUtils.class)) { + when(taskExecutionContext.getTaskParams()) + .thenReturn("{\"address\":\"http://localhost\",\"jobId\":\"1\",\"token\":\"token\"}";); + datavinesTask.init(); + + requestUtilsStatic.when(() -> RequestUtils.executeJob(Mockito.any(), Mockito.any(), Mockito.any())) + .thenReturn(executeJobResult); + datavinesTask.submitApplication(); + + requestUtilsStatic + .when(() -> RequestUtils.getJobExecutionStatus(Mockito.any(), Mockito.any(), Mockito.any())) + .thenReturn(executeStatus); + + requestUtilsStatic + .when(() -> RequestUtils.getJobExecutionResult(Mockito.any(), Mockito.any(), Mockito.any())) + .thenReturn(executeResult); + + datavinesTask.trackApplicationStatus(); + Assertions.assertEquals(EXIT_CODE_SUCCESS, datavinesTask.getExitStatusCode()); + } + } + + @Test + void trackApplicationStatusJobExecutionFailureSetsExitCodeFailure() throws TaskException { + when(taskExecutionContext.getTaskParams()) + .thenReturn("{\"address\":\"http://localhost\",\"jobId\":\"1\",\"token\":\"token\"}";); + datavinesTask.init(); + datavinesTask.submitApplication(); + datavinesTask.trackApplicationStatus(); + Assertions.assertEquals(EXIT_CODE_FAILURE, datavinesTask.getExitStatusCode()); Review Comment: Thanks. All tests now mock the static `RequestUtils` methods, so no real HTTP requests are made during unit tests (commits 9b02337514 and 4cb4274b0e). ########## dolphinscheduler-ui/src/views/projects/task/components/node/types.ts: ########## @@ -476,6 +476,9 @@ interface ITaskParams { yarnQueue?: string awsRegion?: string kubeConfig?: string + jobId?: string + token?: string + failureBlock?: string Review Comment: Thanks. `failureBlock` is now typed as `boolean` in the UI model (commit 9b02337514). ########## dolphinscheduler-task-plugin/dolphinscheduler-task-datavines/src/main/java/org/apache/dolphinscheduler/plugin/task/datavines/DatavinesTask.java: ########## @@ -0,0 +1,205 @@ +/* + * 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.plugin.task.datavines; + +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE; +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask; +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; +import org.apache.dolphinscheduler.plugin.task.datavines.utils.RequestUtils; + +import java.util.Collections; +import java.util.List; + +import lombok.extern.slf4j.Slf4j; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.MissingNode; +import com.fasterxml.jackson.databind.node.NullNode; + +@Slf4j +public class DatavinesTask extends AbstractRemoteTask { + + private final TaskExecutionContext taskExecutionContext; + + private DatavinesParameters datavinesParameters; + private String jobExecutionId; + private boolean executionStatus; + + protected DatavinesTask(TaskExecutionContext taskExecutionContext) { + super(taskExecutionContext); + this.taskExecutionContext = taskExecutionContext; + } + + @Override + public List<String> getApplicationIds() throws TaskException { + return Collections.emptyList(); + } + + @Override + public void init() { + final String taskParams = taskExecutionContext.getTaskParams(); + this.datavinesParameters = JSONUtils.parseObject(taskParams, DatavinesParameters.class); + log.info("initialize datavines task params : {}", JSONUtils.toPrettyJsonString(datavinesParameters)); + if (this.datavinesParameters == null || !this.datavinesParameters.checkParameters()) { + throw new DatavinesTaskException("datavines task params is not valid"); + } + } + + @Override + public void submitApplication() throws TaskException { + executeJob(); + } + + @Override + public void trackApplicationStatus() throws TaskException { + trackApplicationStatusInner(); + } + + private void executeJob() { + try { + String address = this.datavinesParameters.getAddress(); + String jobId = this.datavinesParameters.getJobId(); + String token = this.datavinesParameters.getToken(); + String apiResultDataKey = DatavinesTaskConstants.API_RESULT_DATA; + JsonNode result = RequestUtils.executeJob(address, jobId, token); + if (checkResult(result)) { + jobExecutionId = result.get(apiResultDataKey).asText(); + executionStatus = true; + } + } catch (Exception ex) { + Thread.currentThread().interrupt(); + log.error(DatavinesTaskConstants.SUBMIT_FAILED_MSG, ex); + setExitStatusCode(EXIT_CODE_FAILURE); + throw new TaskException(DatavinesTaskConstants.SUBMIT_FAILED_MSG, ex); + } + } + + public void trackApplicationStatusInner() throws TaskException { + try { + String address = this.datavinesParameters.getAddress(); + if (executionStatus && jobExecutionId == null) { + // Use address-taskId as app id + setAppIds(String.format(DatavinesTaskConstants.APPIDS_FORMAT, address, this.jobExecutionId)); + setExitStatusCode(mapStatusToExitCode(false)); + log.info("datavines task failed."); + return; + } + String apiResultDataKey = DatavinesTaskConstants.API_RESULT_DATA; + boolean finishFlag = false; + while (!finishFlag) { + JsonNode jobExecutionStatus = + RequestUtils.getJobExecutionStatus(address, jobExecutionId, + this.datavinesParameters.getToken()); + if (!checkResult(jobExecutionStatus)) { + break; + } + String jobExecutionStatusStr = jobExecutionStatus.get(apiResultDataKey).asText(); + switch (jobExecutionStatusStr) { + case DatavinesTaskConstants.STATUS_SUCCESS: + setAppIds(String.format(DatavinesTaskConstants.APPIDS_FORMAT, address, this.jobExecutionId)); + JsonNode jobExecutionResult = + RequestUtils.getJobExecutionResult(address, jobExecutionId, + this.datavinesParameters.getToken()); + if (!checkResult(jobExecutionResult)) { + break; + } + + String jobExecutionResultStr = jobExecutionResult.get(apiResultDataKey).asText(); + boolean checkResult = true; + if (this.datavinesParameters.isFailureBlock()) { + checkResult = DatavinesTaskConstants.STATUS_SUCCESS.equals(jobExecutionResultStr); + } + + setExitStatusCode(mapStatusToExitCode(checkResult)); + log.info("datavines task finished, execution status is {} and check result is {}", + jobExecutionStatusStr, jobExecutionResultStr); + finishFlag = true; + break; + case DatavinesTaskConstants.STATUS_FAILURE: + case DatavinesTaskConstants.STATUS_KILL: + errorHandle("task execution status: " + jobExecutionStatusStr); + finishFlag = true; + break; + default: + Thread.sleep(DatavinesTaskConstants.SLEEP_MILLIS); + } + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + log.error(DatavinesTaskConstants.TRACK_FAILED_MSG, ex); + setExitStatusCode(EXIT_CODE_FAILURE); + throw new TaskException(DatavinesTaskConstants.TRACK_FAILED_MSG, ex); + } + } + + /** + * map datavines task status to exitStatusCode + * + * @param status datavines job status + * @return exitStatusCode + */ + private int mapStatusToExitCode(boolean status) { + if (status) { + return EXIT_CODE_SUCCESS; + } else { + return EXIT_CODE_FAILURE; + } + } + + private boolean checkResult(JsonNode result) { + boolean isCorrect = true; + if (result instanceof MissingNode || result instanceof NullNode) { + errorHandle(DatavinesTaskConstants.API_ERROR_MSG); + isCorrect = false; + } else if (result.get(DatavinesTaskConstants.API_RESULT_CODE) + .asInt() != DatavinesTaskConstants.API_RESULT_CODE_SUCCESS) { + errorHandle(result.get(DatavinesTaskConstants.API_RESULT_MSG)); Review Comment: Thanks. `checkResult` now handles null/MissingNode/NullNode responses and requires the `data` field to be present, treating missing data as an API failure instead of throwing an NPE (commit 4cb4274b0e). ########## dolphinscheduler-task-plugin/dolphinscheduler-task-datavines/src/main/java/org/apache/dolphinscheduler/plugin/task/datavines/DatavinesTask.java: ########## @@ -0,0 +1,205 @@ +/* + * 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.plugin.task.datavines; + +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE; +import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS; + +import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.plugin.task.api.AbstractRemoteTask; +import org.apache.dolphinscheduler.plugin.task.api.TaskException; +import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; +import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; +import org.apache.dolphinscheduler.plugin.task.datavines.utils.RequestUtils; + +import java.util.Collections; +import java.util.List; + +import lombok.extern.slf4j.Slf4j; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.MissingNode; +import com.fasterxml.jackson.databind.node.NullNode; + +@Slf4j +public class DatavinesTask extends AbstractRemoteTask { + + private final TaskExecutionContext taskExecutionContext; + + private DatavinesParameters datavinesParameters; + private String jobExecutionId; + private boolean executionStatus; + + protected DatavinesTask(TaskExecutionContext taskExecutionContext) { + super(taskExecutionContext); + this.taskExecutionContext = taskExecutionContext; + } + + @Override + public List<String> getApplicationIds() throws TaskException { + return Collections.emptyList(); + } + + @Override + public void init() { + final String taskParams = taskExecutionContext.getTaskParams(); + this.datavinesParameters = JSONUtils.parseObject(taskParams, DatavinesParameters.class); + log.info("initialize datavines task params : {}", JSONUtils.toPrettyJsonString(datavinesParameters)); + if (this.datavinesParameters == null || !this.datavinesParameters.checkParameters()) { + throw new DatavinesTaskException("datavines task params is not valid"); + } + } + + @Override + public void submitApplication() throws TaskException { + executeJob(); + } + + @Override + public void trackApplicationStatus() throws TaskException { + trackApplicationStatusInner(); + } + + private void executeJob() { + try { + String address = this.datavinesParameters.getAddress(); + String jobId = this.datavinesParameters.getJobId(); + String token = this.datavinesParameters.getToken(); + String apiResultDataKey = DatavinesTaskConstants.API_RESULT_DATA; + JsonNode result = RequestUtils.executeJob(address, jobId, token); + if (checkResult(result)) { + jobExecutionId = result.get(apiResultDataKey).asText(); + executionStatus = true; + } + } catch (Exception ex) { + Thread.currentThread().interrupt(); + log.error(DatavinesTaskConstants.SUBMIT_FAILED_MSG, ex); + setExitStatusCode(EXIT_CODE_FAILURE); + throw new TaskException(DatavinesTaskConstants.SUBMIT_FAILED_MSG, ex); + } + } + + public void trackApplicationStatusInner() throws TaskException { + try { + String address = this.datavinesParameters.getAddress(); + if (executionStatus && jobExecutionId == null) { + // Use address-taskId as app id + setAppIds(String.format(DatavinesTaskConstants.APPIDS_FORMAT, address, this.jobExecutionId)); + setExitStatusCode(mapStatusToExitCode(false)); + log.info("datavines task failed."); + return; + } + String apiResultDataKey = DatavinesTaskConstants.API_RESULT_DATA; + boolean finishFlag = false; + while (!finishFlag) { + JsonNode jobExecutionStatus = + RequestUtils.getJobExecutionStatus(address, jobExecutionId, + this.datavinesParameters.getToken()); + if (!checkResult(jobExecutionStatus)) { + break; + } + String jobExecutionStatusStr = jobExecutionStatus.get(apiResultDataKey).asText(); + switch (jobExecutionStatusStr) { + case DatavinesTaskConstants.STATUS_SUCCESS: + setAppIds(String.format(DatavinesTaskConstants.APPIDS_FORMAT, address, this.jobExecutionId)); + JsonNode jobExecutionResult = + RequestUtils.getJobExecutionResult(address, jobExecutionId, + this.datavinesParameters.getToken()); + if (!checkResult(jobExecutionResult)) { + break; + } + + String jobExecutionResultStr = jobExecutionResult.get(apiResultDataKey).asText(); + boolean checkResult = true; + if (this.datavinesParameters.isFailureBlock()) { + checkResult = DatavinesTaskConstants.STATUS_SUCCESS.equals(jobExecutionResultStr); + } + + setExitStatusCode(mapStatusToExitCode(checkResult)); + log.info("datavines task finished, execution status is {} and check result is {}", + jobExecutionStatusStr, jobExecutionResultStr); + finishFlag = true; + break; + case DatavinesTaskConstants.STATUS_FAILURE: + case DatavinesTaskConstants.STATUS_KILL: + errorHandle("task execution status: " + jobExecutionStatusStr); + finishFlag = true; + break; + default: + Thread.sleep(DatavinesTaskConstants.SLEEP_MILLIS); + } + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + log.error(DatavinesTaskConstants.TRACK_FAILED_MSG, ex); + setExitStatusCode(EXIT_CODE_FAILURE); + throw new TaskException(DatavinesTaskConstants.TRACK_FAILED_MSG, ex); + } + } + + /** + * map datavines task status to exitStatusCode + * + * @param status datavines job status + * @return exitStatusCode + */ + private int mapStatusToExitCode(boolean status) { + if (status) { + return EXIT_CODE_SUCCESS; + } else { + return EXIT_CODE_FAILURE; + } + } + + private boolean checkResult(JsonNode result) { + boolean isCorrect = true; + if (result instanceof MissingNode || result instanceof NullNode) { + errorHandle(DatavinesTaskConstants.API_ERROR_MSG); + isCorrect = false; + } else if (result.get(DatavinesTaskConstants.API_RESULT_CODE) + .asInt() != DatavinesTaskConstants.API_RESULT_CODE_SUCCESS) { + errorHandle(result.get(DatavinesTaskConstants.API_RESULT_MSG)); + isCorrect = false; + } + return isCorrect; + } + + private void errorHandle(Object msg) { + setExitStatusCode(EXIT_CODE_FAILURE); + log.error("datavines task execute failed with error: {}", msg); + } + + @Override + public AbstractParameters getParameters() { + return datavinesParameters; + } + + @Override + public void cancelApplication() throws TaskException { + String address = this.datavinesParameters.getAddress(); + log.info("trying terminate datavines task, taskId: {}, address: {}, taskId: {}", + this.taskExecutionContext.getTaskInstanceId(), + address, + jobExecutionId); + RequestUtils.killJobExecution(address, jobExecutionId, this.datavinesParameters.getToken()); + log.warn("datavines task terminated, taskId: {}, address: {}, jobExecutionId: {}", + this.taskExecutionContext.getTaskInstanceId(), + address, + jobExecutionId); Review Comment: Thanks. `cancelApplication` now skips the kill API call (with a warning log) when `jobExecutionId` is empty, so it no longer hits the kill endpoint with a null id after a failed submit (commit 3924769190). ########## dolphinscheduler-task-plugin/dolphinscheduler-task-datavines/src/main/java/org/apache/dolphinscheduler/plugin/task/datavines/DatavinesTaskConstants.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.plugin.task.datavines; + +/** + * Custom DinkyTaskConstants Review Comment: Thanks. The class comment now reads `Constants for Datavines tasks.` (commit 9b02337514). ########## docs/docs/en/guide/task/datavines.md: ########## @@ -0,0 +1,30 @@ +# Datavines + +## Overview + +Use `Datavines Task` to create a datavines-type task and support data quality job in Datavines. When the worker executes `Datavines Task`, +it will call `Datavines API` to trigger datavines job. Click [here](https://datavane.github.io/datavines-website/) for details about `Datavines`. + +## Create Task + +- Click Project Management-Project Name-Workflow Definition, and click the "Create Workflow" button to enter the DAG editing page. +- Drag <img src="../../../../img/tasks/icons/datavines.png" width="15"/> from the toolbar to the canvas. + +## Task Parameter + +- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md) `Default Task Parameters` section for default parameters. + +| **Parameter** | **Description** | +|-------------------|-------------------------------------------------------------------------------------------------------| +| Datavines Address | The URL for the Datavines service, e.g., http://localhost:5600. | +| Datavines Job ID | The unique job id for a datavines job. | +| Datavines token | The Datawines service access token can be obtained through token management on the Datavines service. | Review Comment: Thanks. The `Datawines` typo has been fixed to `Datavines` (commit 9b02337514). -- 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]
