[Improvement-3327][api]support re-upload the resource file (#3395)

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

@@ -16,6 +16,11 @@
  */
 package org.apache.dolphinscheduler.api.controller;
 
+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 org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.exceptions.ApiException;
 import org.apache.dolphinscheduler.api.service.ResourcesService;
@@ -26,11 +31,6 @@ 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.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 org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -63,23 +63,15 @@ public class ResourcesController extends BaseController {
     private UdfFuncService udfFuncService;
 
     /**
-     * create resource
+     * create directory
      *
-     * @param loginUser login user
-     * @param alias alias
-     * @param description description
-     * @param type type
-     * @return create result code
-     */
-
-    /**
      * @param loginUser   login user
      * @param type        type
      * @param alias       alias
      * @param description description
      * @param pid         parent id
      * @param currentDir  current directory
-     * @return
+     * @return create result code
      */
     @ApiOperation(value = "createDirctory", notes = "CREATE_RESOURCE_NOTES")
     @ApiImplicitParams({
@@ -140,6 +132,7 @@ public class ResourcesController extends BaseController {
      * @param resourceId  resource id
      * @param type        resource type
      * @param description description
+     * @param file        resource file
      * @return update result code
      */
     @ApiOperation(value = "updateResource", notes = "UPDATE_RESOURCE_NOTES")
@@ -147,7 +140,8 @@ public class ResourcesController extends BaseController {
             @ApiImplicitParam(name = "id", value = "RESOURCE_ID", required = true, dataType = "Int", example = "100"),
             @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 = "description", value = "RESOURCE_DESC", dataType = "String"),
+            @ApiImplicitParam(name = "file", value = "RESOURCE_FILE", required = true, dataType = "MultipartFile")
     })
     @PostMapping(value = "/update")
     @ApiException(UPDATE_RESOURCE_ERROR)
@@ -155,10 +149,11 @@ public class ResourcesController extends BaseController {
                                  @RequestParam(value = "id") int resourceId,
                                  @RequestParam(value = "type") ResourceType type,
                                  @RequestParam(value = "name") String alias,
-                                 @RequestParam(value = "description", required = false) String description) {
+                                 @RequestParam(value = "description", required = false) String description,
-        logger.info("login user {}, update resource, type: {}, resource alias: {}, desc: {}",
+                                 @RequestParam(value = "file" ,required = false) MultipartFile file) {
-                loginUser.getUserName(), type, alias, description);
+        logger.info("login user {}, update resource, type: {}, resource alias: {}, desc: {}, file: {}",
-        return resourceService.updateResource(loginUser, resourceId, alias, description, type);
+                loginUser.getUserName(), type, alias, description, file);
+        return resourceService.updateResource(loginUser, resourceId, alias, description, type, file);
     }
 
     /**

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java

@@ -287,6 +287,7 @@ public class ResourcesService extends BaseService {
      * @param name          name
      * @param desc          description
      * @param type          resource type
+     * @param file          resource file
      * @return  update result code
      */
     @Transactional(rollbackFor = Exception.class)
@@ -294,7 +295,8 @@ public class ResourcesService extends BaseService {
                                  int resourceId,
                                  String name,
                                  String desc,
-                                 ResourceType type) {
+                                 ResourceType type,
+                                 MultipartFile file) {
         Result result = new Result();
 
         // if resource upload startup
@@ -330,6 +332,42 @@ public class ResourcesService extends BaseService {
             return result;
         }
 
+        if (file != null) {
+
+            // file is empty
+            if (file.isEmpty()) {
+                logger.error("file is empty: {}", file.getOriginalFilename());
+                putMsg(result, Status.RESOURCE_FILE_IS_EMPTY);
+                return result;
+            }
+
+            // file suffix
+            String fileSuffix = FileUtils.suffix(file.getOriginalFilename());
+            String nameSuffix = FileUtils.suffix(name);
+
+            // determine file suffix
+            if (!(StringUtils.isNotEmpty(fileSuffix) && fileSuffix.equalsIgnoreCase(nameSuffix))) {
+                /**
+                 * rename file suffix and original suffix must be consistent
+                 */
+                logger.error("rename file suffix and original suffix must be consistent: {}", file.getOriginalFilename());
+                putMsg(result, Status.RESOURCE_SUFFIX_FORBID_CHANGE);
+                return result;
+            }
+
+            //If resource type is UDF, only jar packages are allowed to be uploaded, and the suffix must be .jar
+            if (Constants.UDF.equals(type.name()) && !JAR.equalsIgnoreCase(FileUtils.suffix(originFullName))) {
+                logger.error(Status.UDF_RESOURCE_SUFFIX_NOT_JAR.getMsg());
+                putMsg(result, Status.UDF_RESOURCE_SUFFIX_NOT_JAR);
+                return result;
+            }
+            if (file.getSize() > Constants.MAX_FILE_SIZE) {
+                logger.error("file size is too large: {}", file.getOriginalFilename());
+                putMsg(result, Status.RESOURCE_SIZE_EXCEED_LIMIT);
+                return result;
+            }
+        }
+
         // query tenant by user id
         String tenantCode = getTenantCode(resource.getUserId(),result);
         if (StringUtils.isEmpty(tenantCode)){
@@ -379,26 +417,33 @@ public class ResourcesService extends BaseService {
         }
 
         // updateResource data
-        List<Integer> childrenResource = listAllChildren(resource,false);
+
         Date now = new Date();
 
         resource.setAlias(name);
         resource.setFullName(fullName);
         resource.setDescription(desc);
         resource.setUpdateTime(now);
+        if (file != null) {
+            resource.setFileName(file.getOriginalFilename());
+            resource.setSize(file.getSize());
+        }
 
         try {
             resourcesMapper.updateById(resource);
-            if (resource.isDirectory() && CollectionUtils.isNotEmpty(childrenResource)) {
+            if (resource.isDirectory()) {
-                String matcherFullName = Matcher.quoteReplacement(fullName);
+                List<Integer> childrenResource = listAllChildren(resource,false);
-                List<Resource> childResourceList = new ArrayList<>();
+                if (CollectionUtils.isNotEmpty(childrenResource)) {
-                List<Resource> resourceList = resourcesMapper.listResourceByIds(childrenResource.toArray(new Integer[childrenResource.size()]));
+                    String matcherFullName = Matcher.quoteReplacement(fullName);
-                childResourceList = resourceList.stream().map(t -> {
+                    List<Resource> childResourceList = new ArrayList<>();
-                    t.setFullName(t.getFullName().replaceFirst(originFullName, matcherFullName));
+                    List<Resource> resourceList = resourcesMapper.listResourceByIds(childrenResource.toArray(new Integer[childrenResource.size()]));
-                    t.setUpdateTime(now);
+                    childResourceList = resourceList.stream().map(t -> {
-                    return t;
+                        t.setFullName(t.getFullName().replaceFirst(originFullName, matcherFullName));
-                }).collect(Collectors.toList());
+                        t.setUpdateTime(now);
-                resourcesMapper.batchUpdateResource(childResourceList);
+                        return t;
+                    }).collect(Collectors.toList());
+                    resourcesMapper.batchUpdateResource(childResourceList);
+                }
             }
 
             putMsg(result, Status.SUCCESS);
@@ -414,11 +459,31 @@ public class ResourcesService extends BaseService {
             logger.error(Status.UPDATE_RESOURCE_ERROR.getMsg(), e);
             throw new ServiceException(Status.UPDATE_RESOURCE_ERROR);
         }
+
         // if name unchanged, return directly without moving on HDFS
-        if (originResourceName.equals(name)) {
+        if (originResourceName.equals(name) && file == null) {
             return result;
         }
 
+        if (file != null) {
+            // fail upload
+            if (!upload(loginUser, fullName, file, type)) {
+                logger.error("upload resource: {} file: {} failed.", name, file.getOriginalFilename());
+                putMsg(result, Status.HDFS_OPERATION_ERROR);
+                throw new RuntimeException(String.format("upload resource: %s file: %s failed.", name, file.getOriginalFilename()));
+            }
+            if (!fullName.equals(originFullName)) {
+                try {
+                    HadoopUtils.getInstance().delete(originHdfsFileName,false);
+                } catch (IOException e) {
+                    logger.error(e.getMessage(),e);
+                    throw new RuntimeException(String.format("delete resource: %s failed.", originFullName));
+                }
+            }
+            return result;
+        }
+
+
         // get the path of dest file in hdfs
         String destHdfsFileName = HadoopUtils.getHdfsFileName(resource.getType(),tenantCode,fullName);
 

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java

@@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.api.service;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import org.apache.dolphinscheduler.api.enums.Status;
-import org.apache.dolphinscheduler.api.exceptions.ServiceException;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
 import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
@@ -159,19 +158,19 @@ public class ResourcesServiceTest {
         PowerMockito.when(PropertyUtils.getResUploadStartupState()).thenReturn(false);
         User user = new User();
         //HDFS_NOT_STARTUP
-        Result result = resourcesService.updateResource(user,1,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE);
+        Result result = resourcesService.updateResource(user,1,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.HDFS_NOT_STARTUP.getMsg(),result.getMsg());
 
         //RESOURCE_NOT_EXIST
         Mockito.when(resourcesMapper.selectById(1)).thenReturn(getResource());
         PowerMockito.when(PropertyUtils.getResUploadStartupState()).thenReturn(true);
-        result = resourcesService.updateResource(user,0,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE);
+        result = resourcesService.updateResource(user,0,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.RESOURCE_NOT_EXIST.getMsg(),result.getMsg());
 
         //USER_NO_OPERATION_PERM
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest","ResourcesServiceTest",ResourceType.FILE,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.USER_NO_OPERATION_PERM.getMsg(),result.getMsg());
 
@@ -186,7 +185,7 @@ public class ResourcesServiceTest {
         } catch (IOException e) {
             logger.error(e.getMessage(),e);
         }
-        result = resourcesService.updateResource(user, 1, "ResourcesServiceTest1.jar", "ResourcesServiceTest", ResourceType.UDF);
+        result = resourcesService.updateResource(user, 1, "ResourcesServiceTest1.jar", "ResourcesServiceTest", ResourceType.UDF,null);
         Assert.assertEquals(Status.RESOURCE_NOT_EXIST.getMsg(),result.getMsg());
 
         //SUCCESS
@@ -199,25 +198,25 @@ public class ResourcesServiceTest {
             logger.error(e.getMessage(),e);
         }
 
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest.jar","ResourcesServiceTest",ResourceType.FILE);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest.jar","ResourcesServiceTest",ResourceType.FILE,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.SUCCESS.getMsg(),result.getMsg());
 
         //RESOURCE_EXIST
         Mockito.when(resourcesMapper.queryResourceList("/ResourcesServiceTest1.jar", 0, 0)).thenReturn(getResourceList());
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.FILE);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.FILE,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.RESOURCE_EXIST.getMsg(),result.getMsg());
         //USER_NOT_EXIST
         Mockito.when(userMapper.selectById(Mockito.anyInt())).thenReturn(null);
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.UDF);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.UDF,null);
         logger.info(result.toString());
         Assert.assertTrue(Status.USER_NOT_EXIST.getCode() == result.getCode());
 
         //TENANT_NOT_EXIST
         Mockito.when(userMapper.selectById(1)).thenReturn(getUser());
         Mockito.when(tenantMapper.queryById(Mockito.anyInt())).thenReturn(null);
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.UDF);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest",ResourceType.UDF,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.TENANT_NOT_EXIST.getMsg(),result.getMsg());
 
@@ -231,7 +230,7 @@ public class ResourcesServiceTest {
             logger.error(e.getMessage(),e);
         }
 
-        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest1.jar",ResourceType.UDF);
+        result = resourcesService.updateResource(user,1,"ResourcesServiceTest1.jar","ResourcesServiceTest1.jar",ResourceType.UDF,null);
         logger.info(result.toString());
         Assert.assertEquals(Status.SUCCESS.getMsg(),result.getMsg());