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

Merge branch 'dev' into 133-merge-dev

133-merge-dev
bao liang 2020-11-09 10:08:19 +08:00 committed by GitHub
parent 536974d351 4b502e361f
commit 7239979167
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
65 changed files with 568 additions and 534 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

192
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/ResourcesController.java
View File

@ -14,8 +14,33 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.api.controller;
import static org.apache.dolphinscheduler.api.enums.Status.AUTHORIZED_FILE_RESOURCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.AUTHORIZED_UDF_FUNCTION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.AUTHORIZE_RESOURCE_TREE;
import static org.apache.dolphinscheduler.api.enums.Status.CREATE_RESOURCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.CREATE_RESOURCE_FILE_ON_LINE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.CREATE_UDF_FUNCTION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.DELETE_RESOURCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.DELETE_UDF_FUNCTION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.DOWNLOAD_RESOURCE_FILE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.EDIT_RESOURCE_FILE_ON_LINE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_DATASOURCE_BY_TYPE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_RESOURCES_LIST_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_RESOURCES_LIST_PAGING;
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_UDF_FUNCTION_LIST_PAGING_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.RESOURCE_FILE_IS_EMPTY;
import static org.apache.dolphinscheduler.api.enums.Status.RESOURCE_NOT_EXIST;
import static org.apache.dolphinscheduler.api.enums.Status.UNAUTHORIZED_UDF_FUNCTION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.UPDATE_RESOURCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.UPDATE_UDF_FUNCTION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.VERIFY_RESOURCE_BY_NAME_AND_TYPE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.VERIFY_UDF_FUNCTION_NAME_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.VIEW_RESOURCE_FILE_ON_LINE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.VIEW_UDF_FUNCTION_ERROR;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.ResourcesService;
@ -26,12 +51,11 @@ import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.common.enums.ResourceType;
import org.apache.dolphinscheduler.common.enums.UdfType;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.entity.User;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.apache.commons.lang.StringUtils;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -39,14 +63,22 @@ import org.springframework.core.io.Resource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import springfox.documentation.annotations.ApiIgnore;
import java.util.Map;
import static org.apache.dolphinscheduler.api.enums.Status.*;
/**
* resources controller
*/
@ -57,14 +89,12 @@ public class ResourcesController extends BaseController {
private static final Logger logger = LoggerFactory.getLogger(ResourcesController.class);
@Autowired
private ResourcesService resourceService;
@Autowired
private UdfFuncService udfFuncService;
/**
* create directory
*
* @param loginUser login user
* @param type type
@ -79,7 +109,8 @@ public class ResourcesController extends BaseController {
@ApiImplicitParam(name = "type", value = "RESOURCE_TYPE", required = true, dataType = "ResourceType"),
@ApiImplicitParam(name = "name", value = "RESOURCE_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "description", value = "RESOURCE_DESC", dataType = "String"),
@ApiImplicitParam(name = "file", value = "RESOURCE_FILE", required = true, dataType = "MultipartFile")
@ApiImplicitParam(name = "pid", value = "RESOURCE_PID", required = true, dataType = "Int"),
@ApiImplicitParam(name = "currentDir", value = "RESOURCE_CURRENTDIR", required = true, dataType = "String")
})
@PostMapping(value = "/directory/create")
@ApiException(CREATE_RESOURCE_ERROR)
@ -96,12 +127,13 @@ public class ResourcesController extends BaseController {
/**
* create resource
*
* @param loginUser login user
* @param alias alias
* @param description description
* @param type type
* @param file file
* @param loginUser
* @param type
* @param alias
* @param description
* @param file
* @param pid
* @param currentDir
* @return create result code
*/
@ApiOperation(value = "createResource", notes = "CREATE_RESOURCE_NOTES")
@ -109,7 +141,9 @@ public class ResourcesController extends BaseController {
@ApiImplicitParam(name = "type", value = "RESOURCE_TYPE", required = true, dataType = "ResourceType"),
@ApiImplicitParam(name = "name", value = "RESOURCE_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "description", value = "RESOURCE_DESC", dataType = "String"),
@ApiImplicitParam(name = "file", value = "RESOURCE_FILE", required = true, dataType = "MultipartFile")
@ApiImplicitParam(name = "file", value = "RESOURCE_FILE", required = true, dataType = "MultipartFile"),
@ApiImplicitParam(name = "pid", value = "RESOURCE_PID", required = true, dataType = "Int"),
@ApiImplicitParam(name = "currentDir", value = "RESOURCE_CURRENTDIR", required = true, dataType = "String")
})
@PostMapping(value = "/create")
@ApiException(CREATE_RESOURCE_ERROR)
@ -128,10 +162,10 @@ public class ResourcesController extends BaseController {
/**
* update resource
*
* @param loginUser login user
* @param alias alias
* @param resourceId resource id
* @param type resource type
* @param loginUser login user
* @param alias alias
* @param resourceId resource id
* @param type resource type
* @param description description
* @param file resource file
* @return update result code
@ -161,7 +195,7 @@ public class ResourcesController extends BaseController {
* query resources list
*
* @param loginUser login user
* @param type resource type
* @param type resource type
* @return resource list
*/
@ApiOperation(value = "queryResourceList", notes = "QUERY_RESOURCE_LIST_NOTES")
@ -183,10 +217,10 @@ public class ResourcesController extends BaseController {
* query resources list paging
*
* @param loginUser login user
* @param type resource type
* @param type resource type
* @param searchVal search value
* @param pageNo page number
* @param pageSize page size
* @param pageNo page number
* @param pageSize page size
* @return resource list page
*/
@ApiOperation(value = "queryResourceListPaging", notes = "QUERY_RESOURCE_LIST_PAGING_NOTES")
@ -223,7 +257,7 @@ public class ResourcesController extends BaseController {
/**
* delete resource
*
* @param loginUser login user
* @param loginUser login user
* @param resourceId resource id
* @return delete result code
*/
@ -247,8 +281,8 @@ public class ResourcesController extends BaseController {
* verify resource by alias and type
*
* @param loginUser login user
* @param fullName resource full name
* @param type resource type
* @param fullName resource full name
* @param type resource type
* @return true if the resource name not exists, otherwise return false
*/
@ApiOperation(value = "verifyResourceName", notes = "VERIFY_RESOURCE_NAME_NOTES")
@ -273,7 +307,7 @@ public class ResourcesController extends BaseController {
* query resources jar list
*
* @param loginUser login user
* @param type resource type
* @param type resource type
* @return resource list
*/
@ApiOperation(value = "queryResourceByProgramType", notes = "QUERY_RESOURCE_LIST_NOTES")
@ -299,14 +333,16 @@ public class ResourcesController extends BaseController {
* query resource by full name and type
*
* @param loginUser login user
* @param fullName resource full name
* @param type resource type
* @param fullName resource full name
* @param type resource type
* @param id resource id
* @return true if the resource name not exists, otherwise return false
*/
@ApiOperation(value = "queryResource", notes = "QUERY_BY_RESOURCE_NAME")
@ApiImplicitParams({
@ApiImplicitParam(name = "type", value = "RESOURCE_TYPE", required = true, dataType = "ResourceType"),
@ApiImplicitParam(name = "fullName", value = "RESOURCE_FULL_NAME", required = true, dataType = "String")
@ApiImplicitParam(name = "fullName", value = "RESOURCE_FULL_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "id", value = "RESOURCE_ID", required = false, dataType = "Int")
})
@GetMapping(value = "/queryResource")
@ResponseStatus(HttpStatus.OK)
@ -325,10 +361,10 @@ public class ResourcesController extends BaseController {
/**
* view resource file online
*
* @param loginUser login user
* @param resourceId resource id
* @param loginUser login user
* @param resourceId resource id
* @param skipLineNum skip line number
* @param limit limit
* @param limit limit
* @return resource content
*/
@ApiOperation(value = "viewResource", notes = "VIEW_RESOURCE_BY_ID_NOTES")
@ -352,13 +388,14 @@ public class ResourcesController extends BaseController {
/**
* create resource file online
*
* @param loginUser login user
* @param type resource type
* @param fileName file name
* @param fileSuffix file suffix
* @param description description
* @param content content
* @param loginUser
* @param type
* @param fileName
* @param fileSuffix
* @param description
* @param content
* @param pid
* @param currentDir
* @return create result code
*/
@ApiOperation(value = "onlineCreateResource", notes = "ONLINE_CREATE_RESOURCE_NOTES")
@ -367,7 +404,9 @@ public class ResourcesController extends BaseController {
@ApiImplicitParam(name = "fileName", value = "RESOURCE_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "suffix", value = "SUFFIX", required = true, dataType = "String"),
@ApiImplicitParam(name = "description", value = "RESOURCE_DESC", dataType = "String"),
@ApiImplicitParam(name = "content", value = "CONTENT", required = true, dataType = "String")
@ApiImplicitParam(name = "content", value = "CONTENT", required = true, dataType = "String"),
@ApiImplicitParam(name = "pid", value = "RESOURCE_PID", required = true, dataType = "Int"),
@ApiImplicitParam(name = "currentDir", value = "RESOURCE_CURRENTDIR", required = true, dataType = "String")
})
@PostMapping(value = "/online-create")
@ApiException(CREATE_RESOURCE_FILE_ON_LINE_ERROR)
@ -392,9 +431,9 @@ public class ResourcesController extends BaseController {
/**
* edit resource file online
*
* @param loginUser login user
* @param loginUser login user
* @param resourceId resource id
* @param content content
* @param content content
* @return update result code
*/
@ApiOperation(value = "updateResourceContent", notes = "UPDATE_RESOURCE_NOTES")
@ -420,7 +459,7 @@ public class ResourcesController extends BaseController {
/**
* download resource file
*
* @param loginUser login user
* @param loginUser login user
* @param resourceId resource id
* @return resource content
*/
@ -449,14 +488,14 @@ public class ResourcesController extends BaseController {
/**
* create udf function
*
* @param loginUser login user
* @param type udf type
* @param funcName function name
* @param argTypes argument types
* @param database database
* @param loginUser login user
* @param type udf type
* @param funcName function name
* @param argTypes argument types
* @param database database
* @param description description
* @param className class name
* @param resourceId resource id
* @param className class name
* @param resourceId resource id
* @return create result code
*/
@ApiOperation(value = "createUdfFunc", notes = "CREATE_UDF_FUNCTION_NOTES")
@ -490,12 +529,12 @@ public class ResourcesController extends BaseController {
* view udf function
*
* @param loginUser login user
* @param id resource id
* @param id resource id
* @return udf function detail
*/
@ApiOperation(value = "viewUIUdfFunction", notes = "VIEW_UDF_FUNCTION_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "resourceId", value = "RESOURCE_ID", required = true, dataType = "Int", example = "100")
@ApiImplicitParam(name = "id", value = "RESOURCE_ID", required = true, dataType = "Int", example = "100")
})
@GetMapping(value = "/udf-func/update-ui")
@ -512,26 +551,27 @@ public class ResourcesController extends BaseController {
/**
* update udf function
*
* @param loginUser login user
* @param type resource type
* @param funcName function name
* @param argTypes argument types
* @param database data base
* @param loginUser login user
* @param type resource type
* @param funcName function name
* @param argTypes argument types
* @param database data base
* @param description description
* @param resourceId resource id
* @param className class name
* @param udfFuncId udf function id
* @param resourceId resource id
* @param className class name
* @param udfFuncId udf function id
* @return update result code
*/
@ApiOperation(value = "updateUdfFunc", notes = "UPDATE_UDF_FUNCTION_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "id", value = "UDF_ID", required = true, dataType = "Int"),
@ApiImplicitParam(name = "type", value = "UDF_TYPE", required = true, dataType = "UdfType"),
@ApiImplicitParam(name = "funcName", value = "FUNC_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "suffix", value = "CLASS_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "className", value = "CLASS_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "argTypes", value = "ARG_TYPES", dataType = "String"),
@ApiImplicitParam(name = "database", value = "DATABASE_NAME", dataType = "String"),
@ApiImplicitParam(name = "description", value = "UDF_DESC", dataType = "String"),
@ApiImplicitParam(name = "id", value = "RESOURCE_ID", required = true, dataType = "Int", example = "100")
@ApiImplicitParam(name = "resourceId", value = "RESOURCE_ID", required = true, dataType = "Int", example = "100")
})
@PostMapping(value = "/udf-func/update")
@ -556,8 +596,8 @@ public class ResourcesController extends BaseController {
*
* @param loginUser login user
* @param searchVal search value
* @param pageNo page number
* @param pageSize page size
* @param pageNo page number
* @param pageSize page size
* @return udf function list page
*/
@ApiOperation(value = "queryUdfFuncListPaging", notes = "QUERY_UDF_FUNCTION_LIST_PAGING_NOTES")
@ -589,7 +629,7 @@ public class ResourcesController extends BaseController {
* query udf func list by type
*
* @param loginUser login user
* @param type resource type
* @param type resource type
* @return resource list
*/
@ApiOperation(value = "queryUdfFuncList", notes = "QUERY_UDF_FUNC_LIST_NOTES")
@ -612,7 +652,7 @@ public class ResourcesController extends BaseController {
* verify udf function name can use or not
*
* @param loginUser login user
* @param name name
* @param name name
* @return true if the name can user, otherwise return false
*/
@ApiOperation(value = "verifyUdfFuncName", notes = "VERIFY_UDF_FUNCTION_NAME_NOTES")
@ -657,7 +697,7 @@ public class ResourcesController extends BaseController {
* authorized file resource list
*
* @param loginUser login user
* @param userId user id
* @param userId user id
* @return authorized result
*/
@ApiOperation(value = "authorizedFile", notes = "AUTHORIZED_FILE_NOTES")
@ -679,7 +719,7 @@ public class ResourcesController extends BaseController {
* unauthorized file resource list
*
* @param loginUser login user
* @param userId user id
* @param userId user id
* @return unauthorized result code
*/
@ApiOperation(value = "authorizeResourceTree", notes = "AUTHORIZE_RESOURCE_TREE_NOTES")
@ -701,7 +741,7 @@ public class ResourcesController extends BaseController {
* unauthorized udf function
*
* @param loginUser login user
* @param userId user id
* @param userId user id
* @return unauthorized result code
*/
@ApiOperation(value = "unauthUDFFunc", notes = "UNAUTHORIZED_UDF_FUNC_NOTES")
@ -724,7 +764,7 @@ public class ResourcesController extends BaseController {
* authorized udf function
*
* @param loginUser login user
* @param userId user id
* @param userId user id
* @return authorized result code
*/
@ApiOperation(value = "authUDFFunc", notes = "AUTHORIZED_UDF_FUNC_NOTES")

23
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/TenantController.java
View File

@ -29,6 +29,7 @@ import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.TenantService;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.dao.entity.User;
@ -72,7 +73,6 @@ public class TenantController extends BaseController {
*
* @param loginUser login user
* @param tenantCode tenant code
* @param tenantName tenant name
* @param queueId queue id
* @param description description
* @return create result code
@ -80,7 +80,6 @@ public class TenantController extends BaseController {
@ApiOperation(value = "createTenant", notes = "CREATE_TENANT_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "tenantCode", value = "TENANT_CODE", required = true, dataType = "String"),
@ApiImplicitParam(name = "tenantName", value = "TENANT_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "queueId", value = "QUEUE_ID", required = true, dataType = "Int", example = "100"),
@ApiImplicitParam(name = "description", value = "TENANT_DESC", dataType = "String")
@ -90,12 +89,13 @@ public class TenantController extends BaseController {
@ApiException(CREATE_TENANT_ERROR)
public Result createTenant(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam(value = "tenantCode") String tenantCode,
@RequestParam(value = "tenantName") String tenantName,
@RequestParam(value = "queueId") int queueId,
@RequestParam(value = "description", required = false) String description) throws Exception {
logger.info("login user {}, create tenant, tenantCode: {}, tenantName: {}, queueId: {}, desc: {}",
loginUser.getUserName(), tenantCode, tenantName, queueId, description);
Map<String, Object> result = tenantService.createTenant(loginUser, tenantCode, tenantName, queueId, description);
String userReplace = StringUtils.replaceNRTtoUnderline(loginUser.getUserName());
String tenantCodeReplace = StringUtils.replaceNRTtoUnderline(tenantCode);
String descReplace = StringUtils.replaceNRTtoUnderline(description);
logger.info("login user {}, create tenant, tenantCode: {}, queueId: {}, desc: {}", userReplace, tenantCodeReplace, queueId, descReplace);
Map<String, Object> result = tenantService.createTenant(loginUser, tenantCode, queueId, description);
return returnDataList(result);
}
@ -157,7 +157,6 @@ public class TenantController extends BaseController {
* @param loginUser login user
* @param id tennat id
* @param tenantCode tennat code
* @param tenantName tennat name
* @param queueId queue id
* @param description description
* @return update result code
@ -166,7 +165,6 @@ public class TenantController extends BaseController {
@ApiImplicitParams({
@ApiImplicitParam(name = "ID", value = "TENANT_ID", required = true, dataType = "Int", example = "100"),
@ApiImplicitParam(name = "tenantCode", value = "TENANT_CODE", required = true, dataType = "String"),
@ApiImplicitParam(name = "tenantName", value = "TENANT_NAME", required = true, dataType = "String"),
@ApiImplicitParam(name = "queueId", value = "QUEUE_ID", required = true, dataType = "Int", example = "100"),
@ApiImplicitParam(name = "description", value = "TENANT_DESC", type = "String")
@ -177,12 +175,13 @@ public class TenantController extends BaseController {
public Result updateTenant(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam(value = "id") int id,
@RequestParam(value = "tenantCode") String tenantCode,
@RequestParam(value = "tenantName") String tenantName,
@RequestParam(value = "queueId") int queueId,
@RequestParam(value = "description", required = false) String description) throws Exception {
logger.info("login user {}, updateProcessInstance tenant, tenantCode: {}, tenantName: {}, queueId: {}, description: {}",
loginUser.getUserName(), tenantCode, tenantName, queueId, description);
Map<String, Object> result = tenantService.updateTenant(loginUser, id, tenantCode, tenantName, queueId, description);
String userReplace = StringUtils.replaceNRTtoUnderline(loginUser.getUserName());
String tenantCodeReplace = StringUtils.replaceNRTtoUnderline(tenantCode);
String descReplace = StringUtils.replaceNRTtoUnderline(description);
logger.info("login user {}, create tenant, tenantCode: {}, queueId: {}, desc: {}", userReplace, tenantCodeReplace, queueId, descReplace);
Map<String, Object> result = tenantService.updateTenant(loginUser, id, tenantCode, queueId, description);
return returnDataList(result);
}

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
View File

@ -36,7 +36,7 @@ public enum Status {
USER_NAME_NULL(10004, "user name is null", "用户名不能为空"),
HDFS_OPERATION_ERROR(10006, "hdfs operation error", "hdfs操作错误"),
TASK_INSTANCE_NOT_FOUND(10008, "task instance not found", "任务实例不存在"),
TENANT_NAME_EXIST(10009, "tenant code {0} already exists", "租户编码[{0}]已存在"),
TENANT_CODE_EXIST(10009, "tenant code {0} already exists", "租户编码[{0}]已存在"),
USER_NOT_EXIST(10010, "user {0} not exists", "用户[{0}]不存在"),
ALERT_GROUP_NOT_EXIST(10011, "alarm group not found", "告警组不存在"),
ALERT_GROUP_EXIST(10012, "alarm group already exists", "告警组名称已存在"),

5
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
View File

@ -32,7 +32,6 @@ public interface TenantService {
*
* @param loginUser login user
* @param tenantCode tenant code
* @param tenantName tenant name
* @param queueId queue id
* @param desc description
* @return create result code
@ -40,7 +39,6 @@ public interface TenantService {
*/
Map<String, Object> createTenant(User loginUser,
String tenantCode,
String tenantName,
int queueId,
String desc) throws Exception;
@ -61,13 +59,12 @@ public interface TenantService {
* @param loginUser login user
* @param id tennat id
* @param tenantCode tennat code
* @param tenantName tennat name
* @param queueId queue id
* @param desc description
* @return update result code
* @throws Exception exception
*/
Map<String, Object> updateTenant(User loginUser, int id, String tenantCode, String tenantName, int queueId,
Map<String, Object> updateTenant(User loginUser, int id, String tenantCode, int queueId,
String desc) throws Exception;
/**

12
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
View File

@ -76,7 +76,6 @@ public class TenantServiceImpl extends BaseService implements TenantService {
*
* @param loginUser login user
* @param tenantCode tenant code
* @param tenantName tenant name
* @param queueId queue id
* @param desc description
* @return create result code
@ -85,7 +84,6 @@ public class TenantServiceImpl extends BaseService implements TenantService {
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> createTenant(User loginUser,
String tenantCode,
String tenantName,
int queueId,
String desc) throws Exception {
@ -113,7 +111,6 @@ public class TenantServiceImpl extends BaseService implements TenantService {
return result;
}
tenant.setTenantCode(tenantCode);
tenant.setTenantName(tenantName);
tenant.setQueueId(queueId);
tenant.setDescription(desc);
tenant.setCreateTime(now);
@ -166,13 +163,12 @@ public class TenantServiceImpl extends BaseService implements TenantService {
* @param loginUser login user
* @param id tennat id
* @param tenantCode tennat code
* @param tenantName tennat name
* @param queueId queue id
* @param desc description
* @return update result code
* @throws Exception exception
*/
public Map<String, Object> updateTenant(User loginUser, int id, String tenantCode, String tenantName, int queueId,
public Map<String, Object> updateTenant(User loginUser, int id, String tenantCode, int queueId,
String desc) throws Exception {
Map<String, Object> result = new HashMap<>(5);
@ -215,10 +211,6 @@ public class TenantServiceImpl extends BaseService implements TenantService {
tenant.setTenantCode(tenantCode);
}
if (StringUtils.isNotEmpty(tenantName)) {
tenant.setTenantName(tenantName);
}
if (queueId != 0) {
tenant.setQueueId(queueId);
}
@ -337,7 +329,7 @@ public class TenantServiceImpl extends BaseService implements TenantService {
public Result verifyTenantCode(String tenantCode) {
Result result = new Result();
if (checkTenantExists(tenantCode)) {
putMsg(result, Status.TENANT_NAME_EXIST, tenantCode);
putMsg(result, Status.TENANT_CODE_EXIST, tenantCode);
} else {
putMsg(result, Status.SUCCESS);
}

1
dolphinscheduler-api/src/main/resources/i18n/messages.properties
View File

@ -117,7 +117,6 @@ VERIFY_QUEUE_NOTES=verify queue
TENANT_TAG=tenant related operation
CREATE_TENANT_NOTES=create tenant
TENANT_CODE=tenant code
TENANT_NAME=tenant name
QUEUE_NAME=queue name
PASSWORD=password
DATA_SOURCE_OTHER=jdbc connection params, format:{"key1":"value1",...}

1
dolphinscheduler-api/src/main/resources/i18n/messages_en_US.properties
View File

@ -117,7 +117,6 @@ VERIFY_QUEUE_NOTES=verify queue
TENANT_TAG=tenant related operation
CREATE_TENANT_NOTES=create tenant
TENANT_CODE=tenant code
TENANT_NAME=tenant name
QUEUE_NAME=queue name
PASSWORD=password
DATA_SOURCE_OTHER=jdbc connection params, format:{"key1":"value1",...}

1
dolphinscheduler-api/src/main/resources/i18n/messages_zh_CN.properties
View File

@ -116,7 +116,6 @@ VERIFY_QUEUE_NOTES=验证队列
TENANT_TAG=租户相关操作
CREATE_TENANT_NOTES=创建租户
TENANT_CODE=租户编码
TENANT_NAME=租户名称
QUEUE_NAME=队列名
PASSWORD=密码
DATA_SOURCE_OTHER=jdbc连接参数，格式为:{"key1":"value1",...}

8
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/TenantControllerTest.java
View File

@ -43,7 +43,6 @@ public class TenantControllerTest extends AbstractControllerTest{
public void testCreateTenant() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("tenantCode","tenantCode");
paramsMap.add("tenantName","tenantName");
paramsMap.add("queueId","1");
paramsMap.add("description","tenant description");
@ -84,7 +83,6 @@ public class TenantControllerTest extends AbstractControllerTest{
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("id","9");
paramsMap.add("tenantCode","cxc_te");
paramsMap.add("tenantName","tenant_update_2");
paramsMap.add("queueId","1");
paramsMap.add("description","tenant description");
@ -96,7 +94,7 @@ public class TenantControllerTest extends AbstractControllerTest{
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
Assert.assertEquals(Status.TENANT_NOT_EXIST.getCode(),result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@ -133,7 +131,7 @@ public class TenantControllerTest extends AbstractControllerTest{
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertEquals(Status.TENANT_NAME_EXIST.getCode(), result.getCode().intValue());
Assert.assertEquals(Status.TENANT_CODE_EXIST.getCode(), result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
@ -165,7 +163,7 @@ public class TenantControllerTest extends AbstractControllerTest{
.andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
Assert.assertEquals(Status.TENANT_NOT_EXIST.getCode(),result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
}
}

13
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
View File

@ -73,8 +73,6 @@ public class TenantServiceTest {
private static final String tenantCode = "TenantServiceTest";
private static final String tenantName = "TenantServiceTest";
@Test
public void testCreateTenant() {
@ -83,17 +81,17 @@ public class TenantServiceTest {
try {
//check tenantCode
Map<String, Object> result =
tenantService.createTenant(getLoginUser(), "%!1111", tenantName, 1, "TenantServiceTest");
tenantService.createTenant(getLoginUser(), "%!1111", 1, "TenantServiceTest");
logger.info(result.toString());
Assert.assertEquals(Status.VERIFY_TENANT_CODE_ERROR, result.get(Constants.STATUS));
//check exist
result = tenantService.createTenant(loginUser, tenantCode, tenantName, 1, "TenantServiceTest");
result = tenantService.createTenant(loginUser, tenantCode, 1, "TenantServiceTest");
logger.info(result.toString());
Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
// success
result = tenantService.createTenant(loginUser, "test", "test", 1, "TenantServiceTest");
result = tenantService.createTenant(loginUser, "test", 1, "TenantServiceTest");
logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
@ -126,11 +124,11 @@ public class TenantServiceTest {
try {
// id not exist
Map<String, Object> result =
tenantService.updateTenant(getLoginUser(), 912222, tenantCode, tenantName, 1, "desc");
tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, "desc");
logger.info(result.toString());
// success
Assert.assertEquals(Status.TENANT_NOT_EXIST, result.get(Constants.STATUS));
result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, "TenantServiceTest001", 1, "desc");
result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, "desc");
logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
} catch (Exception e) {
@ -236,7 +234,6 @@ public class TenantServiceTest {
Tenant tenant = new Tenant();
tenant.setId(id);
tenant.setTenantCode(tenantCode);
tenant.setTenantName(tenantName);
return tenant;
}

39
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
View File

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.common.utils;
import static java.nio.charset.StandardCharsets.UTF_8;
import static com.fasterxml.jackson.databind.DeserializationFeature.ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT;
import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
import static com.fasterxml.jackson.databind.DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL;
@ -126,6 +128,22 @@ public class JSONUtils {
return null;
}
/**
* deserialize
*
* @param src byte array
* @param clazz class
* @param <T> deserialize type
* @return deserialize type
*/
public static <T> T parseObject(byte[] src, Class<T> clazz) {
if (src == null) {
return null;
}
String json = new String(src, UTF_8);
return parseObject(json, clazz);
}
/**
* json to list
*
@ -253,6 +271,27 @@ public class JSONUtils {
}
}
/**
* serialize to json byte
*
* @param obj object
* @param <T> object type
* @return byte array
*/
public static <T> byte[] toJsonByteArray(T obj) {
if (obj == null) {
return null;
}
String json = "";
try {
json = toJsonString(obj);
} catch (Exception e) {
logger.error("json serialize exception.", e);
}
return json.getBytes(UTF_8);
}
public static ObjectNode parseObject(String text) {
try {
return (ObjectNode) objectMapper.readTree(text);

35
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/JSONUtilsTest.java
View File

@ -14,13 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.utils;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.process.Property;
import java.util.ArrayList;
import java.util.HashMap;
@ -28,13 +28,15 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.process.Property;
import org.junit.Assert;
import org.junit.Test;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
public class JSONUtilsTest {
@Test
@ -108,9 +110,8 @@ public class JSONUtilsTest {
Assert.assertEquals(Direct.IN, direct);
}
@Test
public void String2MapTest() {
public void string2MapTest() {
String str = list2String();
List<LinkedHashMap> maps = JSONUtils.toList(str,
@ -145,6 +146,18 @@ public class JSONUtilsTest {
Assert.assertNull(JSONUtils.parseObject("foo", String.class));
}
@Test
public void testJsonByteArray() {
String str = "foo";
byte[] serializeByte = JSONUtils.toJsonByteArray(str);
String deserialize = JSONUtils.parseObject(serializeByte, String.class);
Assert.assertEquals(str, deserialize);
str = null;
serializeByte = JSONUtils.toJsonByteArray(str);
deserialize = JSONUtils.parseObject(serializeByte, String.class);
Assert.assertNull(deserialize);
}
@Test
public void testToList() {
Assert.assertEquals(new ArrayList(),

14
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java
View File

@ -42,11 +42,6 @@ public class Tenant {
*/
private String tenantCode;
/**
* tenant name
*/
private String tenantName;
/**
* description
*/
@ -97,14 +92,6 @@ public class Tenant {
this.tenantCode = tenantCode;
}
public String getTenantName() {
return tenantName;
}
public void setTenantName(String tenantName) {
this.tenantName = tenantName;
}
public int getQueueId() {
return queueId;
}
@ -150,7 +137,6 @@ public class Tenant {
return "Tenant{" +
"id=" + id +
", tenantCode='" + tenantCode + '\'' +
", tenantName='" + tenantName + '\'' +
", queueId=" + queueId +
", queueName='" + queueName + '\'' +
", queue='" + queue + '\'' +

15
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/User.java
View File

@ -79,12 +79,6 @@ public class User {
@TableField(exist = false)
private String tenantCode;
/**
* tenant name
*/
@TableField(exist = false)
private String tenantName;
/**
* queue name
*/
@ -203,14 +197,6 @@ public class User {
this.alertGroup = alertGroup;
}
public String getTenantName() {
return tenantName;
}
public void setTenantName(String tenantName) {
this.tenantName = tenantName;
}
public String getTenantCode() {
return tenantCode;
}
@ -272,7 +258,6 @@ public class User {
", tenantId=" + tenantId +
", state=" + state +
", tenantCode='" + tenantCode + '\'' +
", tenantName='" + tenantName + '\'' +
", queueName='" + queueName + '\'' +
", alertGroup='" + alertGroup + '\'' +
", queue='" + queue + '\'' +

2
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml
View File

@ -25,7 +25,7 @@
and pd.name = #{processDefinitionName}
</select>
<select id="queryByDefineName" resultType="org.apache.dolphinscheduler.dao.entity.ProcessDefinition">
select pd.*,u.user_name,p.name as project_name,t.tenant_code,t.tenant_name,q.queue,q.queue_name
select pd.*,u.user_name,p.name as project_name,t.tenant_code,q.queue,q.queue_name
from t_ds_process_definition pd
JOIN t_ds_user u ON pd.user_id = u.id
JOIN t_ds_project p ON pd.project_id = p.id

2
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TenantMapper.xml
View File

@ -34,7 +34,7 @@
FROM t_ds_tenant t,t_ds_queue q
WHERE t.queue_id = q.id
<if test="searchVal != null and searchVal != ''">
and t.tenant_name like concat('%', #{searchVal}, '%')
and t.tenant_code like concat('%', #{searchVal}, '%')
</if>
order by t.update_time desc
</select>

4
dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/UserMapper.xml
View File

@ -32,7 +32,7 @@
</select>
<select id="queryUserPaging" resultType="org.apache.dolphinscheduler.dao.entity.User">
select u.id,u.user_name,u.user_password,u.user_type,u.email,u.phone,u.tenant_id,u.create_time,
u.update_time,t.tenant_name,u.state,
u.update_time,t.tenant_code,u.state,
case when u.queue <![CDATA[ <> ]]> '' then u.queue else q.queue_name end as queue, q.queue_name
from t_ds_user u
left join t_ds_tenant t on u.tenant_id=t.id
@ -44,7 +44,7 @@
order by u.update_time desc
</select>
<select id="queryDetailsById" resultType="org.apache.dolphinscheduler.dao.entity.User">
select u.*, t.tenant_name,
select u.*, t.tenant_code,
case when u.queue <![CDATA[ <> ]]> '' then u.queue else q.queue_name end as queue_name
from t_ds_user u
left join t_ds_tenant t on u.tenant_id=t.id

1
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ResourceMapperTest.java
View File

@ -308,7 +308,6 @@ public class ResourceMapperTest {
Tenant tenant = new Tenant();
tenant.setTenantName("ut tenant ");
tenant.setTenantCode("ut tenant code for resource");
int tenantInsertStatus = tenantMapper.insert(tenant);

4
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/TenantMapperTest.java
View File

@ -137,12 +137,12 @@ public class TenantMapperTest {
Tenant tenant = insertOne();
tenant.setTenantCode("ut code");
tenant.setTenantName("ut name");
tenant.setQueueId(queue.getId());
tenantMapper.updateById(tenant);
Page<Tenant> page = new Page(1,3);
IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, tenant.getTenantName());
//tenant.getTenantCode() used instead of tenant.getTenantName()
IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, tenant.getTenantCode());
Assert.assertNotEquals(tenantIPage.getTotal(), 0);
}

2
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/UserMapperTest.java
View File

@ -170,7 +170,6 @@ public class UserMapperTest {
private Tenant insertOneTenant(){
Tenant tenant = new Tenant();
tenant.setTenantCode("dolphin");
tenant.setTenantName("dolphin test");
tenant.setDescription("dolphin user use");
tenant.setQueue("1");
tenant.setCreateTime(new Date());
@ -186,7 +185,6 @@ public class UserMapperTest {
private Tenant insertOneTenant(Queue queue){
Tenant tenant = new Tenant();
tenant.setTenantCode("dolphin");
tenant.setTenantName("dolphin test");
tenant.setDescription("dolphin user use");
tenant.setQueueId(queue.getId());
tenant.setQueue(queue.getQueue());

8
dolphinscheduler-remote/pom.xml
View File

@ -35,6 +35,10 @@
</properties>
<dependencies>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-common</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
@ -48,10 +52,6 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
</dependencies>
</project>

28
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskExecuteAckCommand.java
View File

@ -14,14 +14,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import com.fasterxml.jackson.annotation.JsonFormat;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.Serializable;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
/**
* execute task request command
*/
@ -35,7 +37,7 @@ public class TaskExecuteAckCommand implements Serializable {
/**
* startTime
*/
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone = "GMT+8")
private Date startTime;
/**
@ -111,23 +113,23 @@ public class TaskExecuteAckCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.TASK_EXECUTE_ACK);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "TaskExecuteAckCommand{" +
"taskInstanceId=" + taskInstanceId +
", startTime=" + startTime +
", host='" + host + '\'' +
", status=" + status +
", logPath='" + logPath + '\'' +
", executePath='" + executePath + '\'' +
'}';
return "TaskExecuteAckCommand{"
+ "taskInstanceId=" + taskInstanceId
+ ", startTime=" + startTime
+ ", host='" + host + '\''
+ ", status=" + status
+ ", logPath='" + logPath + '\''
+ ", executePath='" + executePath + '\''
+ '}';
}
}

13
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskExecuteRequestCommand.java
View File

@ -14,9 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.Serializable;
@ -50,18 +51,18 @@ public class TaskExecuteRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.TASK_EXECUTE_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "TaskExecuteRequestCommand{" +
"taskExecutionContext='" + taskExecutionContext + '\'' +
'}';
return "TaskExecuteRequestCommand{"
+ "taskExecutionContext='" + taskExecutionContext + '\''
+ '}';
}
}

27
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskExecuteResponseCommand.java
View File

@ -14,20 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import com.fasterxml.jackson.annotation.JsonFormat;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.Serializable;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
/**
* execute task response command
*/
public class TaskExecuteResponseCommand implements Serializable {
public TaskExecuteResponseCommand() {
}
@ -49,7 +50,7 @@ public class TaskExecuteResponseCommand implements Serializable {
/**
* end time
*/
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date endTime;
@ -120,22 +121,22 @@ public class TaskExecuteResponseCommand implements Serializable {
* package response command
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.TASK_EXECUTE_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "TaskExecuteResponseCommand{" +
"taskInstanceId=" + taskInstanceId +
", status=" + status +
", endTime=" + endTime +
", processId=" + processId +
", appIds='" + appIds + '\'' +
'}';
return "TaskExecuteResponseCommand{"
+ "taskInstanceId=" + taskInstanceId
+ ", status=" + status
+ ", endTime=" + endTime
+ ", processId=" + processId
+ ", appIds='" + appIds + '\''
+ '}';
}
}

14
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskKillRequestCommand.java
View File

@ -14,9 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.Serializable;
@ -30,7 +31,6 @@ public class TaskKillRequestCommand implements Serializable {
*/
private int taskInstanceId;
public int getTaskInstanceId() {
return taskInstanceId;
}
@ -44,18 +44,18 @@ public class TaskKillRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.TASK_KILL_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "TaskKillRequestCommand{" +
"taskInstanceId=" + taskInstanceId +
'}';
return "TaskKillRequestCommand{"
+ "taskInstanceId=" + taskInstanceId
+ '}';
}
}

22
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskKillResponseCommand.java
View File

@ -14,9 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.Serializable;
import java.util.List;
@ -52,7 +53,6 @@ public class TaskKillResponseCommand implements Serializable {
*/
protected List<String> appIds;
public int getTaskInstanceId() {
return taskInstanceId;
}
@ -98,22 +98,22 @@ public class TaskKillResponseCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.TASK_KILL_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}
@Override
public String toString() {
return "TaskKillResponseCommand{" +
"taskInstanceId=" + taskInstanceId +
", host='" + host + '\'' +
", status=" + status +
", processId=" + processId +
", appIds=" + appIds +
'}';
return "TaskKillResponseCommand{"
+ "taskInstanceId=" + taskInstanceId
+ ", host='" + host + '\''
+ ", status=" + status
+ ", processId=" + processId
+ ", appIds=" + appIds
+ '}';
}
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/GetLogBytesRequestCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -53,10 +53,10 @@ public class GetLogBytesRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.GET_LOG_BYTES_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/GetLogBytesResponseCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -54,10 +54,10 @@ public class GetLogBytesResponseCommand implements Serializable {
* @param opaque request unique identification
* @return command
*/
public Command convert2Command(long opaque){
public Command convert2Command(long opaque) {
Command command = new Command(opaque);
command.setType(CommandType.GET_LOG_BYTES_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/RemoveTaskLogRequestCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -53,10 +53,10 @@ public class RemoveTaskLogRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.REMOVE_TAK_LOG_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/RemoveTaskLogResponseCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -53,10 +53,10 @@ public class RemoveTaskLogResponseCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(long opaque){
public Command convert2Command(long opaque) {
Command command = new Command(opaque);
command.setType(CommandType.REMOVE_TAK_LOG_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/RollViewLogRequestCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -81,10 +81,10 @@ public class RollViewLogRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.ROLL_VIEW_LOG_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/RollViewLogResponseCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -54,10 +54,10 @@ public class RollViewLogResponseCommand implements Serializable {
* @param opaque request unique identification
* @return command
*/
public Command convert2Command(long opaque){
public Command convert2Command(long opaque) {
Command command = new Command(opaque);
command.setType(CommandType.ROLL_VIEW_LOG_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/ViewLogRequestCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -53,10 +53,10 @@ public class ViewLogRequestCommand implements Serializable {
*
* @return command
*/
public Command convert2Command(){
public Command convert2Command() {
Command command = new Command();
command.setType(CommandType.VIEW_WHOLE_LOG_REQUEST);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

6
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/log/ViewLogResponseCommand.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.remote.command.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
@ -54,10 +54,10 @@ public class ViewLogResponseCommand implements Serializable {
* @param opaque request unique identification
* @return command
*/
public Command convert2Command(long opaque){
public Command convert2Command(long opaque) {
Command command = new Command(opaque);
command.setType(CommandType.VIEW_WHOLE_LOG_RESPONSE);
byte[] body = JsonSerializer.serialize(this);
byte[] body = JSONUtils.toJsonByteArray(this);
command.setBody(body);
return command;
}

92
dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/utils/JsonSerializer.java
View File

@ -1,92 +0,0 @@
/*
* 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.remote.utils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
/**
* json serialize or deserialize
*/
public class JsonSerializer {
private static final ObjectMapper objectMapper = new ObjectMapper();
private static final Logger logger = LoggerFactory.getLogger(JsonSerializer.class);
private JsonSerializer(){
}
/**
* serialize to byte
*
* @param obj object
* @param <T> object type
* @return byte array
*/
public static <T> byte[] serialize(T obj) {
String json = "";
try {
json = objectMapper.writeValueAsString(obj);
} catch (JsonProcessingException e) {
logger.error("serializeToString exception!", e);
}
return json.getBytes(Constants.UTF8);
}
/**
* serialize to string
* @param obj object
* @param <T> object type
* @return string
*/
public static <T> String serializeToString(T obj) {
String json = "";
try {
json = objectMapper.writeValueAsString(obj);
} catch (JsonProcessingException e) {
logger.error("serializeToString exception!", e);
}
return json;
}
/**
* deserialize
*
* @param src byte array
* @param clazz class
* @param <T> deserialize type
* @return deserialize type
*/
public static <T> T deserialize(byte[] src, Class<T> clazz) {
String json = new String(src, StandardCharsets.UTF_8);
try {
return objectMapper.readValue(json, clazz);
} catch (IOException e) {
logger.error("deserialize exception!", e);
return null;
}
}
}

58
dolphinscheduler-remote/src/test/java/org/apache/dolphinscheduler/remote/JsonSerializerTest.java
View File

@ -1,58 +0,0 @@
/*
* 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.remote;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.junit.Assert;
import org.junit.Test;
public class JsonSerializerTest {
@Test
public void testSerialize(){
TestObj testObj = new TestObj();
testObj.setAge(12);
byte[] serializeByte = JsonSerializer.serialize(testObj);
//
TestObj deserialize = JsonSerializer.deserialize(serializeByte, TestObj.class);
Assert.assertEquals(testObj.getAge(), deserialize.getAge());
}
static class TestObj {
private int age;
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public String toString() {
return "TestObj{" +
"age=" + age +
'}';
}
}
}

4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskExecutionContext.java
View File

@ -18,9 +18,9 @@
package org.apache.dolphinscheduler.server.entity;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.TaskExecuteRequestCommand;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import java.io.Serializable;
import java.util.Date;
@ -475,7 +475,7 @@ public class TaskExecutionContext implements Serializable {
public Command toCommand() {
TaskExecuteRequestCommand requestCommand = new TaskExecuteRequestCommand();
requestCommand.setTaskExecutionContext(JsonSerializer.serializeToString(this));
requestCommand.setTaskExecutionContext(JSONUtils.toJsonString(this));
return requestCommand.convert2Command();
}

64
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/LoggerRequestProcessor.java
View File

@ -14,19 +14,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.log;
import io.netty.channel.Channel;
import org.apache.dolphinscheduler.common.utils.IOUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.log.*;
import org.apache.dolphinscheduler.remote.command.log.GetLogBytesRequestCommand;
import org.apache.dolphinscheduler.remote.command.log.GetLogBytesResponseCommand;
import org.apache.dolphinscheduler.remote.command.log.RemoveTaskLogRequestCommand;
import org.apache.dolphinscheduler.remote.command.log.RemoveTaskLogResponseCommand;
import org.apache.dolphinscheduler.remote.command.log.RollViewLogRequestCommand;
import org.apache.dolphinscheduler.remote.command.log.RollViewLogResponseCommand;
import org.apache.dolphinscheduler.remote.command.log.ViewLogRequestCommand;
import org.apache.dolphinscheduler.remote.command.log.ViewLogResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Collections;
@ -38,6 +49,11 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
/**
* logger request process logic
*/
@ -47,7 +63,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
private final ThreadPoolExecutor executor;
public LoggerRequestProcessor(){
public LoggerRequestProcessor() {
this.executor = new ThreadPoolExecutor(4, 4, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100));
}
@ -59,35 +75,35 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
* reuqest task log command type
*/
final CommandType commandType = command.getType();
switch (commandType){
switch (commandType) {
case GET_LOG_BYTES_REQUEST:
GetLogBytesRequestCommand getLogRequest = JsonSerializer.deserialize(
GetLogBytesRequestCommand getLogRequest = JSONUtils.parseObject(
command.getBody(), GetLogBytesRequestCommand.class);
byte[] bytes = getFileContentBytes(getLogRequest.getPath());
GetLogBytesResponseCommand getLogResponse = new GetLogBytesResponseCommand(bytes);
channel.writeAndFlush(getLogResponse.convert2Command(command.getOpaque()));
break;
case VIEW_WHOLE_LOG_REQUEST:
ViewLogRequestCommand viewLogRequest = JsonSerializer.deserialize(
ViewLogRequestCommand viewLogRequest = JSONUtils.parseObject(
command.getBody(), ViewLogRequestCommand.class);
String msg = readWholeFileContent(viewLogRequest.getPath());
ViewLogResponseCommand viewLogResponse = new ViewLogResponseCommand(msg);
channel.writeAndFlush(viewLogResponse.convert2Command(command.getOpaque()));
break;
case ROLL_VIEW_LOG_REQUEST:
RollViewLogRequestCommand rollViewLogRequest = JsonSerializer.deserialize(
RollViewLogRequestCommand rollViewLogRequest = JSONUtils.parseObject(
command.getBody(), RollViewLogRequestCommand.class);
List<String> lines = readPartFileContent(rollViewLogRequest.getPath(),
rollViewLogRequest.getSkipLineNum(), rollViewLogRequest.getLimit());
StringBuilder builder = new StringBuilder();
for (String line : lines){
for (String line : lines) {
builder.append(line + "\r\n");
}
RollViewLogResponseCommand rollViewLogRequestResponse = new RollViewLogResponseCommand(builder.toString());
channel.writeAndFlush(rollViewLogRequestResponse.convert2Command(command.getOpaque()));
break;
case REMOVE_TAK_LOG_REQUEST:
RemoveTaskLogRequestCommand removeTaskLogRequest = JsonSerializer.deserialize(
RemoveTaskLogRequestCommand removeTaskLogRequest = JSONUtils.parseObject(
command.getBody(), RemoveTaskLogRequestCommand.class);
String taskLogPath = removeTaskLogRequest.getPath();
@ -95,10 +111,10 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
File taskLogFile = new File(taskLogPath);
Boolean status = true;
try {
if (taskLogFile.exists()){
if (taskLogFile.exists()) {
status = taskLogFile.delete();
}
}catch (Exception e){
} catch (Exception e) {
status = false;
}
@ -110,7 +126,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
}
}
public ExecutorService getExecutor(){
public ExecutorService getExecutor() {
return this.executor;
}
@ -121,7 +137,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
* @return byte array of file
* @throws Exception exception
*/
private byte[] getFileContentBytes(String filePath){
private byte[] getFileContentBytes(String filePath) {
InputStream in = null;
ByteArrayOutputStream bos = null;
try {
@ -133,9 +149,9 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
bos.write(buf, 0, len);
}
return bos.toByteArray();
}catch (IOException e){
} catch (IOException e) {
logger.error("get file bytes error",e);
}finally {
} finally {
IOUtils.closeQuietly(bos);
IOUtils.closeQuietly(in);
}
@ -152,7 +168,7 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
*/
private List<String> readPartFileContent(String filePath,
int skipLine,
int limit){
int limit) {
try (Stream<String> stream = Files.lines(Paths.get(filePath))) {
return stream.skip(skipLine).limit(limit).collect(Collectors.toList());
} catch (IOException e) {
@ -167,19 +183,19 @@ public class LoggerRequestProcessor implements NettyRequestProcessor {
* @param filePath file path
* @return whole file content
*/
private String readWholeFileContent(String filePath){
private String readWholeFileContent(String filePath) {
BufferedReader br = null;
String line;
StringBuilder sb = new StringBuilder();
try {
br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
while ((line = br.readLine()) != null){
while ((line = br.readLine()) != null) {
sb.append(line + "\r\n");
}
return sb.toString();
}catch (IOException e){
} catch (IOException e) {
logger.error("read file error",e);
}finally {
} finally {
IOUtils.closeQuietly(br);
}
return "";

16
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessor.java
View File

@ -17,10 +17,12 @@
package org.apache.dolphinscheduler.server.master.processor;
import io.netty.channel.Channel;
import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.Preconditions;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.Command;
@ -28,17 +30,17 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.ChannelUtils;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.dolphinscheduler.common.Constants.*;
import io.netty.channel.Channel;
/**
* task ack processor
@ -63,7 +65,7 @@ public class TaskAckProcessor implements NettyRequestProcessor {
*/
private ProcessService processService;
public TaskAckProcessor(){
public TaskAckProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
this.processService = SpringApplicationContext.getBean(ProcessService.class);
@ -77,7 +79,7 @@ public class TaskAckProcessor implements NettyRequestProcessor {
@Override
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_EXECUTE_ACK == command.getType(), String.format("invalid command type : %s", command.getType()));
TaskExecuteAckCommand taskAckCommand = JsonSerializer.deserialize(command.getBody(), TaskExecuteAckCommand.class);
TaskExecuteAckCommand taskAckCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteAckCommand.class);
logger.info("taskAckCommand : {}", taskAckCommand);
taskInstanceCacheManager.cacheTaskInstance(taskAckCommand);
@ -96,10 +98,10 @@ public class TaskAckProcessor implements NettyRequestProcessor {
taskResponseService.addResponse(taskResponseEvent);
while (Stopper.isRunning()){
while (Stopper.isRunning()) {
TaskInstance taskInstance = processService.findTaskInstanceById(taskAckCommand.getTaskInstanceId());
if (taskInstance != null && ackStatus.typeIsRunning()){
if (taskInstance != null && ackStatus.typeIsRunning()) {
break;
}
ThreadUtils.sleep(SLEEP_TIME_MILLIS);

9
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskKillResponseProcessor.java
View File

@ -17,16 +17,18 @@
package org.apache.dolphinscheduler.server.master.processor;
import io.netty.channel.Channel;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.Preconditions;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskKillResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
/**
* task response processor
*/
@ -45,9 +47,8 @@ public class TaskKillResponseProcessor implements NettyRequestProcessor {
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_KILL_RESPONSE == command.getType(), String.format("invalid command type : %s", command.getType()));
TaskKillResponseCommand responseCommand = JsonSerializer.deserialize(command.getBody(), TaskKillResponseCommand.class);
TaskKillResponseCommand responseCommand = JSONUtils.parseObject(command.getBody(), TaskKillResponseCommand.class);
logger.info("received task kill response command : {}", responseCommand);
}
}

17
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
View File

@ -17,27 +17,29 @@
package org.apache.dolphinscheduler.server.master.processor;
import io.netty.channel.Channel;
import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.Preconditions;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.server.master.cache.TaskInstanceCacheManager;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.dolphinscheduler.common.Constants.*;
import io.netty.channel.Channel;
/**
* task response processor
@ -61,7 +63,7 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
*/
private ProcessService processService;
public TaskResponseProcessor(){
public TaskResponseProcessor() {
this.taskResponseService = SpringApplicationContext.getBean(TaskResponseService.class);
this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
this.processService = SpringApplicationContext.getBean(ProcessService.class);
@ -78,7 +80,7 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_EXECUTE_RESPONSE == command.getType(), String.format("invalid command type : %s", command.getType()));
TaskExecuteResponseCommand responseCommand = JsonSerializer.deserialize(command.getBody(), TaskExecuteResponseCommand.class);
TaskExecuteResponseCommand responseCommand = JSONUtils.parseObject(command.getBody(), TaskExecuteResponseCommand.class);
logger.info("received command : {}", responseCommand);
taskInstanceCacheManager.cacheTaskInstance(responseCommand);
@ -95,15 +97,14 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
taskResponseService.addResponse(taskResponseEvent);
while (Stopper.isRunning()){
while (Stopper.isRunning()) {
TaskInstance taskInstance = processService.findTaskInstanceById(taskResponseEvent.getTaskInstanceId());
if (taskInstance != null && responseStatus.typeIsFinished()){
if (taskInstance != null && responseStatus.typeIsFinished()) {
break;
}
ThreadUtils.sleep(SLEEP_TIME_MILLIS);
}
}
}

15
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
View File

@ -32,7 +32,6 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteRequestCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
import org.apache.dolphinscheduler.server.utils.LogUtils;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
@ -73,7 +72,7 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
*/
private final TaskCallbackService taskCallbackService;
public TaskExecuteProcessor(){
public TaskExecuteProcessor() {
this.taskCallbackService = SpringApplicationContext.getBean(TaskCallbackService.class);
this.workerConfig = SpringApplicationContext.getBean(WorkerConfig.class);
this.workerExecService = ThreadUtils.newDaemonFixedThreadExecutor("Worker-Execute-Thread", workerConfig.getWorkerExecThreads());
@ -84,12 +83,12 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
Preconditions.checkArgument(CommandType.TASK_EXECUTE_REQUEST == command.getType(),
String.format("invalid command type : %s", command.getType()));
TaskExecuteRequestCommand taskRequestCommand = JsonSerializer.deserialize(
TaskExecuteRequestCommand taskRequestCommand = JSONUtils.parseObject(
command.getBody(), TaskExecuteRequestCommand.class);
logger.info("received command : {}", taskRequestCommand);
if(taskRequestCommand == null){
if (taskRequestCommand == null) {
logger.error("task execute request command is null");
return;
}
@ -97,7 +96,7 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
String contextJson = taskRequestCommand.getTaskExecutionContext();
TaskExecutionContext taskExecutionContext = JSONUtils.parseObject(contextJson, TaskExecutionContext.class);
if(taskExecutionContext == null){
if (taskExecutionContext == null) {
logger.error("task execution context is null");
return;
}
@ -162,9 +161,9 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
ackCommand.setLogPath(LogUtils.getTaskLogPath(taskExecutionContext));
ackCommand.setHost(taskExecutionContext.getHost());
ackCommand.setStartTime(taskExecutionContext.getStartTime());
if(taskExecutionContext.getTaskType().equals(TaskType.SQL.name()) || taskExecutionContext.getTaskType().equals(TaskType.PROCEDURE.name())){
if (taskExecutionContext.getTaskType().equals(TaskType.SQL.name()) || taskExecutionContext.getTaskType().equals(TaskType.PROCEDURE.name())) {
ackCommand.setExecutePath(null);
}else{
} else {
ackCommand.setExecutePath(taskExecutionContext.getExecutePath());
}
taskExecutionContext.setLogPath(ackCommand.getLogPath());
@ -176,7 +175,7 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
* @param taskExecutionContext taskExecutionContext
* @return execute local path
*/
private String getExecLocalPath(TaskExecutionContext taskExecutionContext){
private String getExecLocalPath(TaskExecutionContext taskExecutionContext) {
return FileUtils.getProcessExecDir(taskExecutionContext.getProjectId(),
taskExecutionContext.getProcessDefineId(),
taskExecutionContext.getProcessInstanceId(),

23
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskKillProcessor.java
View File

@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.server.worker.processor;
import io.netty.channel.Channel;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.LoggerUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.Preconditions;
@ -29,7 +29,6 @@ import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskKillRequestCommand;
import org.apache.dolphinscheduler.remote.command.TaskKillResponseCommand;
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.Pair;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
@ -39,12 +38,15 @@ import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContext
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.log.LogClientService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
/**
* task kill processor
*/
@ -67,8 +69,7 @@ public class TaskKillProcessor implements NettyRequestProcessor {
*/
private TaskExecutionContextCacheManager taskExecutionContextCacheManager;
public TaskKillProcessor(){
public TaskKillProcessor() {
this.taskCallbackService = SpringApplicationContext.getBean(TaskCallbackService.class);
this.workerConfig = SpringApplicationContext.getBean(WorkerConfig.class);
this.taskExecutionContextCacheManager = SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
@ -83,7 +84,7 @@ public class TaskKillProcessor implements NettyRequestProcessor {
@Override
public void process(Channel channel, Command command) {
Preconditions.checkArgument(CommandType.TASK_KILL_REQUEST == command.getType(), String.format("invalid command type : %s", command.getType()));
TaskKillRequestCommand killCommand = JsonSerializer.deserialize(command.getBody(), TaskKillRequestCommand.class);
TaskKillRequestCommand killCommand = JSONUtils.parseObject(command.getBody(), TaskKillRequestCommand.class);
logger.info("received kill command : {}", killCommand);
Pair<Boolean, List<String>> result = doKill(killCommand);
@ -101,14 +102,14 @@ public class TaskKillProcessor implements NettyRequestProcessor {
* @param killCommand
* @return kill result
*/
private Pair<Boolean, List<String>> doKill(TaskKillRequestCommand killCommand){
private Pair<Boolean, List<String>> doKill(TaskKillRequestCommand killCommand) {
List<String> appIds = Collections.EMPTY_LIST;
try {
TaskExecutionContext taskExecutionContext = taskExecutionContextCacheManager.getByTaskInstanceId(killCommand.getTaskInstanceId());
Integer processId = taskExecutionContext.getProcessId();
if (processId == null || processId.equals(0)){
if (processId == null || processId.equals(0)) {
logger.error("process kill failed, process id :{}, task id:{}", processId, killCommand.getTaskInstanceId());
return Pair.of(false, appIds);
}
@ -145,7 +146,7 @@ public class TaskKillProcessor implements NettyRequestProcessor {
taskKillResponseCommand.setStatus(result.getLeft() ? ExecutionStatus.SUCCESS.getCode() : ExecutionStatus.FAILURE.getCode());
taskKillResponseCommand.setAppIds(result.getRight());
TaskExecutionContext taskExecutionContext = taskExecutionContextCacheManager.getByTaskInstanceId(killCommand.getTaskInstanceId());
if(taskExecutionContext != null){
if (taskExecutionContext != null) {
taskKillResponseCommand.setTaskInstanceId(taskExecutionContext.getTaskInstanceId());
taskKillResponseCommand.setHost(taskExecutionContext.getHost());
taskKillResponseCommand.setProcessId(taskExecutionContext.getProcessId());
@ -183,7 +184,7 @@ public class TaskKillProcessor implements NettyRequestProcessor {
} catch (Exception e) {
logger.error("kill yarn job error",e);
} finally {
if(logClient != null){
if (logClient != null) {
logClient.close();
}
}

1
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
View File

@ -78,7 +78,6 @@ public class TaskPriorityQueueConsumerTest {
Tenant tenant = new Tenant();
tenant.setId(1);
tenant.setTenantCode("journey");
tenant.setTenantName("journey");
tenant.setDescription("journey");
tenant.setQueueId(1);
tenant.setCreateTime(new Date());

101
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/TaskAckProcessorTest.java Normal file
View File

@ -0,0 +1,101 @@
/*
* 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.master.processor;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.TaskResponseService;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import java.net.InetSocketAddress;
import java.util.Date;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import io.netty.channel.Channel;
/**
* task ack processor test
*/
@RunWith(PowerMockRunner.class)
@PrepareForTest({SpringApplicationContext.class, TaskResponseEvent.class})
public class TaskAckProcessorTest {
private TaskAckProcessor taskAckProcessor;
private TaskResponseService taskResponseService;
private TaskInstanceCacheManagerImpl taskInstanceCacheManager;
private ProcessService processService;
private TaskExecuteAckCommand taskExecuteAckCommand;
private TaskResponseEvent taskResponseEvent;
private Channel channel;
@Before
public void before() {
PowerMockito.mockStatic(SpringApplicationContext.class);
taskResponseService = PowerMockito.mock(TaskResponseService.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskResponseService.class)).thenReturn(taskResponseService);
taskInstanceCacheManager = PowerMockito.mock(TaskInstanceCacheManagerImpl.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class)).thenReturn(taskInstanceCacheManager);
processService = PowerMockito.mock(ProcessService.class);
PowerMockito.when(SpringApplicationContext.getBean(ProcessService.class)).thenReturn(processService);
taskAckProcessor = new TaskAckProcessor();
channel = PowerMockito.mock(Channel.class);
taskResponseEvent = PowerMockito.mock(TaskResponseEvent.class);
taskExecuteAckCommand = new TaskExecuteAckCommand();
taskExecuteAckCommand.setStatus(1);
taskExecuteAckCommand.setExecutePath("/dolphinscheduler/worker");
taskExecuteAckCommand.setHost("localhost");
taskExecuteAckCommand.setLogPath("/temp/worker.log");
taskExecuteAckCommand.setStartTime(new Date());
taskExecuteAckCommand.setTaskInstanceId(1);
}
@Test
public void testProcess() {
Command command = taskExecuteAckCommand.convert2Command();
Assert.assertEquals(CommandType.TASK_EXECUTE_ACK,command.getType());
InetSocketAddress socketAddress = new InetSocketAddress("localhost",12345);
PowerMockito.when(channel.remoteAddress()).thenReturn(socketAddress);
PowerMockito.mockStatic(TaskResponseEvent.class);
PowerMockito.when(TaskResponseEvent.newAck(Mockito.any(), Mockito.any(), Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.anyInt()))
.thenReturn(taskResponseEvent);
TaskInstance taskInstance = PowerMockito.mock(TaskInstance.class);
PowerMockito.when(processService.findTaskInstanceById(Mockito.any())).thenReturn(taskInstance);
taskAckProcessor.process(channel,command);
}
}

64
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/TaskKillResponseProcessorTest.java Normal file
View File

@ -0,0 +1,64 @@
/*
* 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.master.processor;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.TaskKillResponseCommand;
import java.util.ArrayList;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.powermock.api.mockito.PowerMockito;
import io.netty.channel.Channel;
/**
* task response processor test
*/
public class TaskKillResponseProcessorTest {
private TaskKillResponseProcessor taskKillResponseProcessor;
private TaskKillResponseCommand taskKillResponseCommand;
private Channel channel;
@Before
public void before() {
taskKillResponseProcessor = new TaskKillResponseProcessor();
channel = PowerMockito.mock(Channel.class);
taskKillResponseCommand = new TaskKillResponseCommand();
taskKillResponseCommand.setAppIds(
new ArrayList<String>() {{ add("task_1"); }});
taskKillResponseCommand.setHost("localhost");
taskKillResponseCommand.setProcessId(1);
taskKillResponseCommand.setStatus(1);
taskKillResponseCommand.setTaskInstanceId(1);
}
@Test
public void testProcess() {
Command command = taskKillResponseCommand.convert2Command();
Assert.assertEquals(CommandType.TASK_KILL_RESPONSE,command.getType());
taskKillResponseProcessor.process(channel,command);
}
}

13
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/log/LogClientService.java
View File

@ -14,8 +14,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.service.log;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.remote.NettyRemotingClient;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.log.GetLogBytesRequestCommand;
@ -28,12 +30,10 @@ import org.apache.dolphinscheduler.remote.command.log.ViewLogRequestCommand;
import org.apache.dolphinscheduler.remote.command.log.ViewLogResponseCommand;
import org.apache.dolphinscheduler.remote.config.NettyClientConfig;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.JsonSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* log client
*/
@ -90,7 +90,7 @@ public class LogClientService {
Command command = request.convert2Command();
Command response = this.client.sendSync(address, command, LOG_REQUEST_TIMEOUT);
if (response != null) {
RollViewLogResponseCommand rollReviewLog = JsonSerializer.deserialize(
RollViewLogResponseCommand rollReviewLog = JSONUtils.parseObject(
response.getBody(), RollViewLogResponseCommand.class);
return rollReviewLog.getMsg();
}
@ -119,7 +119,7 @@ public class LogClientService {
Command command = request.convert2Command();
Command response = this.client.sendSync(address, command, LOG_REQUEST_TIMEOUT);
if (response != null) {
ViewLogResponseCommand viewLog = JsonSerializer.deserialize(
ViewLogResponseCommand viewLog = JSONUtils.parseObject(
response.getBody(), ViewLogResponseCommand.class);
return viewLog.getMsg();
}
@ -148,7 +148,7 @@ public class LogClientService {
Command command = request.convert2Command();
Command response = this.client.sendSync(address, command, LOG_REQUEST_TIMEOUT);
if (response != null) {
GetLogBytesResponseCommand getLog = JsonSerializer.deserialize(
GetLogBytesResponseCommand getLog = JSONUtils.parseObject(
response.getBody(), GetLogBytesResponseCommand.class);
return getLog.getData();
}
@ -160,7 +160,6 @@ public class LogClientService {
return result;
}
/**
* remove task log
*
@ -178,7 +177,7 @@ public class LogClientService {
Command command = request.convert2Command();
Command response = this.client.sendSync(address, command, LOG_REQUEST_TIMEOUT);
if (response != null) {
RemoveTaskLogResponseCommand taskLogResponse = JsonSerializer.deserialize(
RemoveTaskLogResponseCommand taskLogResponse = JSONUtils.parseObject(
response.getBody(), RemoveTaskLogResponseCommand.class);
return taskLogResponse.getStatus();
}

2
dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/udp/_source/selectTenant.vue
View File

@ -24,7 +24,7 @@
v-for="item in itemList"
:key="item.id"
:value="item.id"
:label="item.tenantName">
:label="item.tenantCode">
</x-option>
</x-select>
</template>

28
dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/_source/createTenement.vue
View File

@ -34,18 +34,6 @@
</x-input>
</template>
</m-list-box-f>
<m-list-box-f>
<template slot="name"><strong>*</strong>{{$t('Tenant Name')}}</template>
<template slot="content">
<x-input
type="input"
v-model="tenantName"
maxlength="60"
:placeholder="$t('Please enter tenant Name')"
autocomplete="off">
</x-input>
</template>
</m-list-box-f>
<m-list-box-f>
<template slot="name"><strong>*</strong>{{$t('Queue')}}</template>
<template slot="content">
@ -88,8 +76,7 @@
store,
queueList: [],
queueId: '',
tenantCode: null,
tenantName: '',
tenantCode: '',
description: '',
}
},
@ -144,24 +131,12 @@
return false
}
if (!this.tenantName.replace(/\s*/g,"")) {
this.$message.warning(`${i18n.$t('Please enter tenant Name')}`)
return false
}
// Verify tenant name cannot contain special characters
let isSpecial = /[~#^$@%&!*()<>《》:;'"{}【】 ]/gi
if (isSpecial.test(this.tenantName)) {
this.$message.warning(`${i18n.$t('Please enter tenant name without special characters')}`)
return false
}
return true
},
_submit () {
// 提交
let param = {
tenantCode: this.tenantCode,
tenantName: this.tenantName,
queueId: this.queueId,
description: this.description
}
@ -191,7 +166,6 @@
this.queueId = this.item.queueId
})
this.tenantCode = this.item.tenantCode
this.tenantName = this.item.tenantName
this.description = this.item.description
}
})

8
dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/_source/list.vue
View File

@ -25,9 +25,6 @@
<th>
<span>{{$t('Tenant Code')}}</span>
</th>
<th>
<span>{{$t('Tenant Name')}}</span>
</th>
<th>
<span>{{$t('Description')}}</span>
</th>
@ -53,11 +50,6 @@
{{item.tenantCode}}
</span>
</td>
<td>
<span>
{{item.tenantName}}
</span>
</td>
<td>
<span v-if="item.description" class="ellipsis" v-tooltip.large.top.start.light="{text: item.description, maxWidth: '500px'}">{{item.description}}</span>
<span v-else>-</span>

2
dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/createUser.vue
View File

@ -232,7 +232,7 @@
this.tenantList = _.map(arr, v => {
return {
id: v.id,
code: v.tenantName
code: v.tenantCode
}
})
this.$nextTick(() => {

2
dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/list.vue
View File

@ -65,7 +65,7 @@
<td>
<span>{{item.userType === 'GENERAL_USER' ? `${$t('Ordinary users')}` : `${$t('Administrator')}`}}</span>
</td>
<td><span>{{item.tenantName || '-'}}</span></td>
<td><span>{{item.tenantCode || '-'}}</span></td>
<td><span>{{item.queue || '-'}}</span></td>
<td>
<span>{{item.email || '-'}}</span>

2
dolphinscheduler-ui/src/js/conf/home/pages/user/pages/account/_source/info.vue
View File

@ -43,7 +43,7 @@
<m-list-box-f v-ps="['GENERAL_USER']">
<template slot="name">{{$t('Tenant')}}</template>
<template slot="content">
<span class="sp1">{{userInfo.tenantName}}</span>
<span class="sp1">{{userInfo.tenantCode}}</span>
</template>
</m-list-box-f>
<m-list-box-f v-ps="['GENERAL_USER']">

2
dolphinscheduler-ui/src/js/conf/home/store/security/actions.js
View File

@ -282,7 +282,7 @@ export default {
const list = res.data
list.unshift({
id: -1,
tenantName: 'default'
tenantCode: 'default'
})
state.tenantAllList = list
resolve(list)

2
dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
View File

@ -182,12 +182,10 @@ export default {
'Last heartbeat time': 'Last heartbeat time',
'Edit Tenant': 'Edit Tenant',
'Tenant Code': 'Tenant Code',
'Tenant Name': 'Tenant Name',
Queue: 'Yarn Queue',
'Please select a queue': 'default is tenant association queue',
'Please enter the tenant code in English': 'Please enter the tenant code in English',
'Please enter tenant code in English': 'Please enter tenant code in English',
'Please enter tenant name without special characters': 'Please enter tenant name without special characters',
'Edit User': 'Edit User',
Tenant: 'Tenant',
Email: 'Email',

2
dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
View File

@ -185,11 +185,9 @@ export default {
'Last heartbeat time': '最后心跳时间',
'Edit Tenant': '编辑租户',
'Tenant Code': '租户编码',
'Tenant Name': '租户名称',
Queue: 'Yarn 队列',
'Please enter the tenant code in English': '请输入租户编码只允许英文',
'Please enter tenant code in English': '请输入英文租户编码',
'Please enter tenant name without special characters': '请输入不包含特殊字符的租户名称',
'Edit User': '编辑用户',
Tenant: '租户',
Email: '邮件',

4
e2e/src/test/java/org/apache/dolphinscheduler/locator/security/TenantManageLocator.java
View File

@ -27,15 +27,13 @@ public class TenantManageLocator{
public static final By TENANT_INPUT_CODE = By.xpath("//div[2]/div/div/div[2]/div/input");
public static final By TENANT_INPUT_NAME = By.xpath("//div[2]/div[2]/div/input");
public static final By QUEUE = By.xpath("//textarea");
public static final By DESCRIPTION = By.xpath("//textarea");
public static final By SUBMIT_BUTTON = By.xpath("//div[3]/button[2]/span");
public static final By DELETE_TENANT_BUTTON = By.xpath("//div[3]/div[1]/div/table/tr[2]/td[8]/span/button");
public static final By DELETE_TENANT_BUTTON = By.xpath("//div[3]/div[1]/div/table/tr[2]/td[7]/span/button");
public static final By CONFIRM_DELETE_TENANT_BUTTON = By.xpath("//div[2]/div/button[2]/span");

1
e2e/src/test/java/org/apache/dolphinscheduler/page/security/TenantManagePage.java
View File

@ -55,7 +55,6 @@ public class TenantManagePage extends PageCommon {
// tenant data
sendInput(TenantManageLocator.TENANT_INPUT_CODE, tenantManageData.getTenantData("tenantCode"));
sendInput(TenantManageLocator.TENANT_INPUT_NAME, tenantManageData.getTenantData("tenantName"));
sendInput(TenantManageLocator.QUEUE, tenantManageData.getTenantData("queue"));
sendInput(TenantManageLocator.DESCRIPTION, tenantManageData.getTenantData("description"));

1
e2e/src/test/resources/testData/security_zh_cn.yml
View File

@ -29,7 +29,6 @@ userManage:
tenantManage:
tenantCode: selenium_tenant_code_1
tenantName: selenium_tenant_Name
queue: default
description: create tenant test
tenantTitle: 租户管理 - DolphinScheduler

3
pom.xml
View File

@ -710,6 +710,7 @@
<!--<include>**/alert/utils/MailUtilsTest.java</include>-->
<include>**/alert/plugin/EmailAlertPluginTest.java</include>
<include>**/api/controller/ProcessDefinitionControllerTest.java</include>
<include>**/api/controller/TenantControllerTest.java</include>
<include>**/api/dto/resources/filter/ResourceFilterTest.java</include>
<include>**/api/dto/resources/visitor/ResourceTreeVisitorTest.java</include>
<includeDataxTaskTest>**/api/enums/testGetEnum.java</includeDataxTaskTest>
@ -830,6 +831,8 @@
<include>**/server/master/MasterExecThreadTest.java</include>
<include>**/server/master/ParamsTest.java</include>
<include>**/server/master/SubProcessTaskTest.java</include>
<include>**/server/master/processor/TaskAckProcessorTest.java</include>
<include>**/server/master/processor/TaskKillResponseProcessorTest.java</include>
<include>**/server/register/ZookeeperNodeManagerTest.java</include>
<include>**/server/utils/DataxUtilsTest.java</include>
<include>**/server/utils/ExecutionContextTestUtils.java</include>

2
script/dolphinscheduler-daemon.sh
View File

@ -16,7 +16,7 @@
# limitations under the License.
#
usage="Usage: dolphinscheduler-daemon.sh (start|stop|status) <command> "
usage="Usage: dolphinscheduler-daemon.sh (start|stop|status) <api-server|master-server|worker-server|alert-server> "
# if no args specified, show usage
if [ $# -le 1 ]; then

1
sql/create/release-1.0.0_schema/mysql/dolphinscheduler_ddl.sql
View File

@ -381,7 +381,6 @@ DROP TABLE IF EXISTS `t_escheduler_tenant`;
CREATE TABLE `t_escheduler_tenant` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`tenant_code` varchar(64) DEFAULT NULL COMMENT 'tenant code',
`tenant_name` varchar(64) DEFAULT NULL COMMENT 'tenant name',
`desc` varchar(256) DEFAULT NULL COMMENT 'description',
`queue_id` int(11) DEFAULT NULL COMMENT 'queue id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',

1
sql/create/release-1.2.0_schema/postgresql/dolphinscheduler_ddl.sql
View File

@ -584,7 +584,6 @@ DROP TABLE IF EXISTS t_ds_tenant;
CREATE TABLE t_ds_tenant (
id int NOT NULL ,
tenant_code varchar(64) DEFAULT NULL ,
tenant_name varchar(64) DEFAULT NULL ,
description varchar(256) DEFAULT NULL ,
queue_id int DEFAULT NULL ,
create_time timestamp DEFAULT NULL ,

1
sql/dolphinscheduler-postgre.sql
View File

@ -608,7 +608,6 @@ DROP TABLE IF EXISTS t_ds_tenant;
CREATE TABLE t_ds_tenant (
id int NOT NULL ,
tenant_code varchar(64) DEFAULT NULL ,
tenant_name varchar(64) DEFAULT NULL ,
description varchar(256) DEFAULT NULL ,
queue_id int DEFAULT NULL ,
create_time timestamp DEFAULT NULL ,

25
sql/dolphinscheduler_mysql.sql
View File

@ -29,7 +29,7 @@ CREATE TABLE `QRTZ_BLOB_TRIGGERS` (
PRIMARY KEY (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
KEY `SCHED_NAME` (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
CONSTRAINT `QRTZ_BLOB_TRIGGERS_ibfk_1` FOREIGN KEY (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`) REFERENCES `QRTZ_TRIGGERS` (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_BLOB_TRIGGERS
@ -44,7 +44,7 @@ CREATE TABLE `QRTZ_CALENDARS` (
`CALENDAR_NAME` varchar(200) NOT NULL,
`CALENDAR` blob NOT NULL,
PRIMARY KEY (`SCHED_NAME`,`CALENDAR_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_CALENDARS
@ -62,7 +62,7 @@ CREATE TABLE `QRTZ_CRON_TRIGGERS` (
`TIME_ZONE_ID` varchar(80) DEFAULT NULL,
PRIMARY KEY (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
CONSTRAINT `QRTZ_CRON_TRIGGERS_ibfk_1` FOREIGN KEY (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`) REFERENCES `QRTZ_TRIGGERS` (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_CRON_TRIGGERS
@ -93,7 +93,7 @@ CREATE TABLE `QRTZ_FIRED_TRIGGERS` (
KEY `IDX_QRTZ_FT_JG` (`SCHED_NAME`,`JOB_GROUP`),
KEY `IDX_QRTZ_FT_T_G` (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
KEY `IDX_QRTZ_FT_TG` (`SCHED_NAME`,`TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_FIRED_TRIGGERS
@ -117,7 +117,7 @@ CREATE TABLE `QRTZ_JOB_DETAILS` (
PRIMARY KEY (`SCHED_NAME`,`JOB_NAME`,`JOB_GROUP`),
KEY `IDX_QRTZ_J_REQ_RECOVERY` (`SCHED_NAME`,`REQUESTS_RECOVERY`),
KEY `IDX_QRTZ_J_GRP` (`SCHED_NAME`,`JOB_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_JOB_DETAILS
@ -131,7 +131,7 @@ CREATE TABLE `QRTZ_LOCKS` (
`SCHED_NAME` varchar(120) NOT NULL,
`LOCK_NAME` varchar(40) NOT NULL,
PRIMARY KEY (`SCHED_NAME`,`LOCK_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_LOCKS
@ -145,7 +145,7 @@ CREATE TABLE `QRTZ_PAUSED_TRIGGER_GRPS` (
`SCHED_NAME` varchar(120) NOT NULL,
`TRIGGER_GROUP` varchar(200) NOT NULL,
PRIMARY KEY (`SCHED_NAME`,`TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_PAUSED_TRIGGER_GRPS
@ -161,7 +161,7 @@ CREATE TABLE `QRTZ_SCHEDULER_STATE` (
`LAST_CHECKIN_TIME` bigint(13) NOT NULL,
`CHECKIN_INTERVAL` bigint(13) NOT NULL,
PRIMARY KEY (`SCHED_NAME`,`INSTANCE_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_SCHEDULER_STATE
@ -180,7 +180,7 @@ CREATE TABLE `QRTZ_SIMPLE_TRIGGERS` (
`TIMES_TRIGGERED` bigint(10) NOT NULL,
PRIMARY KEY (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
CONSTRAINT `QRTZ_SIMPLE_TRIGGERS_ibfk_1` FOREIGN KEY (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`) REFERENCES `QRTZ_TRIGGERS` (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_SIMPLE_TRIGGERS
@ -207,7 +207,7 @@ CREATE TABLE `QRTZ_SIMPROP_TRIGGERS` (
`BOOL_PROP_2` varchar(1) DEFAULT NULL,
PRIMARY KEY (`SCHED_NAME`,`TRIGGER_NAME`,`TRIGGER_GROUP`),
CONSTRAINT `QRTZ_SIMPROP_TRIGGERS_ibfk_1` FOREIGN KEY (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`) REFERENCES `QRTZ_TRIGGERS` (`SCHED_NAME`, `TRIGGER_NAME`, `TRIGGER_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_SIMPROP_TRIGGERS
@ -248,7 +248,7 @@ CREATE TABLE `QRTZ_TRIGGERS` (
KEY `IDX_QRTZ_T_NFT_ST_MISFIRE` (`SCHED_NAME`,`MISFIRE_INSTR`,`NEXT_FIRE_TIME`,`TRIGGER_STATE`),
KEY `IDX_QRTZ_T_NFT_ST_MISFIRE_GRP` (`SCHED_NAME`,`MISFIRE_INSTR`,`NEXT_FIRE_TIME`,`TRIGGER_GROUP`,`TRIGGER_STATE`),
CONSTRAINT `QRTZ_TRIGGERS_ibfk_1` FOREIGN KEY (`SCHED_NAME`, `JOB_NAME`, `JOB_GROUP`) REFERENCES `QRTZ_JOB_DETAILS` (`SCHED_NAME`, `JOB_NAME`, `JOB_GROUP`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-- ----------------------------
-- Records of QRTZ_TRIGGERS
@ -729,7 +729,7 @@ CREATE TABLE `t_ds_task_instance` (
`alert_flag` tinyint(4) DEFAULT NULL COMMENT 'whether alert',
`retry_times` int(4) DEFAULT '0' COMMENT 'task retry times',
`pid` int(4) DEFAULT NULL COMMENT 'pid of task',
`app_link` varchar(255) DEFAULT NULL COMMENT 'yarn app id',
`app_link` text COMMENT 'yarn app id',
`flag` tinyint(4) DEFAULT '1' COMMENT '0 not available, 1 available',
`retry_interval` int(4) DEFAULT NULL COMMENT 'retry interval when task failed ',
`max_retry_times` int(2) DEFAULT NULL COMMENT 'max retry times',
@ -756,7 +756,6 @@ DROP TABLE IF EXISTS `t_ds_tenant`;
CREATE TABLE `t_ds_tenant` (
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
`tenant_code` varchar(64) DEFAULT NULL COMMENT 'tenant code',
`tenant_name` varchar(64) DEFAULT NULL COMMENT 'tenant name',
`description` varchar(256) DEFAULT NULL,
`queue_id` int(11) DEFAULT NULL COMMENT 'queue id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',