[fix](multi-catalog)fix compatibility issue for s3 endpoint when use incompatible client

fe/fe-core/src/main/java/org/apache/doris/datasource/property/PropertyConverter.java

@@ -123,28 +123,40 @@ public static Map<String, String> convertToHadoopFSProperties(Map<String, String
         } else if (props.containsKey(MinioProperties.ENDPOINT)) {
             return convertToMinioProperties(props, MinioProperties.getCredential(props));
         } else if (props.containsKey(S3Properties.ENDPOINT)) {
-            CloudCredential credential = S3Properties.getCredential(props);
+            CloudCredential s3Credential = S3Properties.getCredential(props);
+            Map<String, String> s3Properties = convertToS3Properties(props, s3Credential);
             String s3CliEndpoint = props.get(S3Properties.ENDPOINT);
-            if (s3CliEndpoint.contains(CosProperties.COS_PREFIX)) {
-                props.putIfAbsent(CosProperties.ENDPOINT, s3CliEndpoint);
-                // CosN is not compatible with S3, when use s3 properties, will convert to cosn properties.
-                return convertToCOSProperties(props, credential);
-            }
-            return convertToS3Properties(props, S3Properties.getCredential(props));
+            return convertToCompatibleS3Properties(props, s3CliEndpoint, s3Credential, s3Properties);
         } else if (props.containsKey(S3Properties.Env.ENDPOINT)) {
             // checkout env in the end
             // compatible with the s3,obs,oss,cos when they use aws client.
             CloudCredentialWithEndpoint envCredentials = S3Properties.getEnvironmentCredentialWithEndpoint(props);
-            if (envCredentials.getEndpoint().contains(CosProperties.COS_PREFIX)) {
-                props.putIfAbsent(CosProperties.ENDPOINT, envCredentials.getEndpoint());
-                // CosN is not compatible with S3, when use s3 properties, will convert to cosn properties.
-                return convertToCOSProperties(props, envCredentials);
-            }
-            return convertToS3EnvProperties(props, envCredentials, false);
+            Map<String, String> s3Properties = convertToS3EnvProperties(props, envCredentials, false);
+            String s3CliEndpoint = envCredentials.getEndpoint();
+            return convertToCompatibleS3Properties(props, s3CliEndpoint, envCredentials, s3Properties);
         }
         return props;
     }
 
+    private static Map<String, String> convertToCompatibleS3Properties(Map<String, String> props,
+                                                                       String s3CliEndpoint,
+                                                                       CloudCredential credential,
+                                                                       Map<String, String> s3Properties) {
+        Map<String, String> heteroProps = new HashMap<>(s3Properties);
+        if (s3CliEndpoint.contains(CosProperties.COS_PREFIX)) {
+            props.putIfAbsent(CosProperties.ENDPOINT, s3CliEndpoint);
+            // CosN is not compatible with S3, when use s3 properties, will convert to cosn properties.
+            heteroProps.putAll(convertToCOSProperties(props, credential));
+        } else if (s3CliEndpoint.contains(ObsProperties.OBS_PREFIX)) {
+            props.putIfAbsent(ObsProperties.ENDPOINT, s3CliEndpoint);
+            heteroProps.putAll(convertToOBSProperties(props, credential));
+        } else if (s3CliEndpoint.contains(OssProperties.OSS_REGION_PREFIX)) {
+            props.putIfAbsent(OssProperties.ENDPOINT, s3CliEndpoint);
+            heteroProps.putAll(convertToOSSProperties(props, credential));
+        }
+        return heteroProps;
+    }
+
 
     private static Map<String, String> convertToOBSProperties(Map<String, String> props,
                                                               CloudCredential credential) {
@@ -237,7 +249,7 @@ private static void setS3FsAccess(Map<String, String> s3Properties, Map<String,
                                       CloudCredential credential) {
         s3Properties.put(Constants.MAX_ERROR_RETRIES, "2");
         s3Properties.put("fs.s3.impl.disable.cache", "true");
-        s3Properties.put("fs.s3.impl", S3AFileSystem.class.getName());
+        s3Properties.putIfAbsent("fs.s3.impl", S3AFileSystem.class.getName());
         String defaultProviderList = String.join(",", S3Properties.AWS_CREDENTIALS_PROVIDERS);
         String credentialsProviders = s3Properties
                 .getOrDefault(S3Properties.CREDENTIALS_PROVIDER, defaultProviderList);

fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/OssProperties.java

@@ -26,6 +26,7 @@
 public class OssProperties extends BaseProperties {
 
     public static final String OSS_PREFIX = "oss.";
+    public static final String OSS_REGION_PREFIX = "oss-";
     public static final String OSS_FS_PREFIX = "fs.oss";
 
     public static final String ENDPOINT = "oss.endpoint";