verify whether the filename start with '/' (#2544)

* fix #2442 when the resource name contains '$',need translate it to '\$'

* fix #2442 when the resource name contains '$',need translate it to '/$'

* verify whether the filename start with '/'

Co-authored-by: qiaozhanwei <qiaozhanwei@outlook.com>
Co-authored-by: dailidong <dailidong66@gmail.com>

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/HadoopUtils.java

@@ -503,6 +503,9 @@ public class HadoopUtils implements Closeable {
      * @return hdfs file name
      */
     public static String getHdfsFileName(ResourceType resourceType, String tenantCode, String fileName) {
+        if (fileName.startsWith("/")) {
+            fileName = fileName.replaceFirst("/","");
+        }
         return String.format("%s/%s", getHdfsDir(resourceType,tenantCode), fileName);
     }
 
@@ -514,6 +517,9 @@ public class HadoopUtils implements Closeable {
      * @return get absolute path and name for file on hdfs
      */
     public static String getHdfsResourceFileName(String tenantCode, String fileName) {
+        if (fileName.startsWith("/")) {
+            fileName = fileName.replaceFirst("/","");
+        }
         return String.format("%s/%s", getHdfsResDir(tenantCode), fileName);
     }
 
@@ -525,6 +531,9 @@ public class HadoopUtils implements Closeable {
      * @return get absolute path and name for udf file on hdfs
      */
     public static String getHdfsUdfFileName(String tenantCode, String fileName) {
+        if (fileName.startsWith("/")) {
+            fileName = fileName.replaceFirst("/","");
+        }
         return String.format("%s/%s", getHdfsUdfDir(tenantCode), fileName);
     }
 

dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/HadoopUtilsTest.java

@@ -127,6 +127,18 @@ public class HadoopUtilsTest {
         Assert.assertEquals("/dolphinscheduler/11000/resources/aa.txt", result);
     }
 
+    @Test
+    public void getHdfsResourceFileName() {
+        String result = hadoopUtils.getHdfsResourceFileName("11000","aa.txt");
+        Assert.assertEquals("/dolphinscheduler/11000/resources/aa.txt", result);
+    }
+
+    @Test
+    public void getHdfsUdfFileName() {
+        String result = hadoopUtils.getHdfsFileName(ResourceType.UDF,"11000","aa.txt");
+        Assert.assertEquals("/dolphinscheduler/11000/udfs/aa.txt", result);
+    }
+
     @Test
     public void isYarnEnabled() {
         boolean result = hadoopUtils.isYarnEnabled();