fix: Resource relate path invalid when tenant change (apache#15581)

* fix can not get correct resource related path when
  user run workflow with differnet tenant of resource
  created
* also fix can not get correct related path when we
  use `resource.storage.type=LOCAL`

dolphinscheduler-storage-plugin/dolphinscheduler-storage-api/src/main/java/org/apache/dolphinscheduler/plugin/storage/api/StorageOperate.java

@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.plugin.storage.api;
 
+import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_TYPE_FILE;
+
 import org.apache.dolphinscheduler.common.constants.Constants;
 import org.apache.dolphinscheduler.common.enums.ResUploadType;
 import org.apache.dolphinscheduler.common.utils.PropertyUtils;
@@ -72,7 +74,16 @@ public interface StorageOperate {
      */
     default String getResourceFileName(String tenantCode, String fullName) {
         String resDir = getResDir(tenantCode);
-        return fullName.replaceFirst(resDir, "");
+        String filenameReplaceResDir = fullName.replaceFirst(resDir, "");
+        if (!filenameReplaceResDir.equals(fullName)) {
+            return filenameReplaceResDir;
+        }
+
+        // Replace resource dir not effective in case of run workflow with different tenant from resource file's.
+        // this is backup solution to get related path, by split with RESOURCE_TYPE_FILE
+        return filenameReplaceResDir.contains(RESOURCE_TYPE_FILE)
+                ? filenameReplaceResDir.split(String.format("%s/", RESOURCE_TYPE_FILE))[1]
+                : filenameReplaceResDir;
     }
 
     /**

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/HdfsStorageOperator.java

@@ -71,7 +71,7 @@
 @Slf4j
 public class HdfsStorageOperator implements Closeable, StorageOperate {
 
-    private static HdfsStorageProperties hdfsProperties = new HdfsStorageProperties();
+    protected static HdfsStorageProperties hdfsProperties = new HdfsStorageProperties();
     private static final String HADOOP_UTILS_KEY = "HADOOP_UTILS_KEY";
 
     private volatile boolean yarnEnabled = false;

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/LocalStorageOperator.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.plugin.storage.hdfs;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class LocalStorageOperator extends HdfsStorageOperator {
+
+    public LocalStorageOperator() {
+        super(new HdfsStorageProperties());
+    }
+
+    public LocalStorageOperator(HdfsStorageProperties hdfsStorageProperties) {
+        super(hdfsStorageProperties);
+    }
+
+    @Override
+    public String getResourceFileName(String tenantCode, String fullName) {
+        // prefix schema `file:/` should be remove in local file mode
+        String fullNameRemoveSchema = fullName.replaceFirst(hdfsProperties.getDefaultFS(), "");
+        return super.getResourceFileName(tenantCode, fullNameRemoveSchema);
+    }
+}

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/LocalStorageOperatorFactory.java

@@ -32,7 +32,7 @@ public class LocalStorageOperatorFactory implements StorageOperateFactory {
     public StorageOperate createStorageOperate() {
         HdfsStorageProperties hdfsStorageProperties = new HdfsStorageProperties();
         hdfsStorageProperties.setDefaultFS(LOCAL_DEFAULT_FS);
-        return new HdfsStorageOperator(hdfsStorageProperties);
+        return new LocalStorageOperator(hdfsStorageProperties);
     }
 
     @Override