[core] RESTCatalog: update parse region from dlf uri

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogOptions.java

@@ -50,6 +50,12 @@ public class RESTCatalogOptions {
                     .noDefaultValue()
                     .withDescription("REST Catalog auth token provider.");
 
+    public static final ConfigOption<String> DLF_REGION =
+            ConfigOptions.key("dlf.region")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("REST Catalog auth DLF region.");
+
     public static final ConfigOption<String> DLF_TOKEN_PATH =
             ConfigOptions.key("dlf.token-path")
                     .stringType()

paimon-core/src/main/java/org/apache/paimon/rest/auth/DLFAuthProviderFactory.java

@@ -21,6 +21,10 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.rest.RESTCatalogOptions;
 
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_PATH;
 import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN_REFRESH_TIME;
 import static org.apache.paimon.rest.RESTCatalogOptions.URI;
@@ -35,7 +39,8 @@ public String identifier() {
 
     @Override
     public AuthProvider create(Options options) {
-        String region = getRegion(options);
+        String region =
+                getRegion(options.getOptional(RESTCatalogOptions.DLF_REGION), options.get(URI));
         if (options.getOptional(RESTCatalogOptions.DLF_TOKEN_PATH).isPresent()) {
             String tokenFilePath = options.get(DLF_TOKEN_PATH);
             long tokenRefreshInMills = options.get(TOKEN_REFRESH_TIME).toMillis();
@@ -51,15 +56,22 @@ public AuthProvider create(Options options) {
         throw new IllegalArgumentException("DLF token path or AK must be set for DLF Auth.");
     }
 
-    private static String getRegion(Options options) {
-        String region = "undefined";
+    protected static String getRegion(Optional<String> region, String uri) {
+        if (region.isPresent()) {
+            return region.get();
+        }
         try {
-            String[] paths = options.get(URI).split("\\.");
-            if (paths.length > 1) {
-                region = paths[1];
+            String regex = "dlf-(?:pre-)?([a-z]+-[a-z]+(?:-\\d+)?)(?:-internal)?";
+
+            Pattern pattern = Pattern.compile(regex);
+            Matcher matcher = pattern.matcher(uri);
+
+            if (matcher.find()) {
+                return matcher.group(1);
             }
         } catch (Exception ignore) {
         }
-        return region;
+        throw new IllegalArgumentException(
+                "Could not get region from conf or uri, please check your config.");
     }
 }

paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java

@@ -43,6 +43,7 @@
 
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_ACCESS_KEY_ID;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_ACCESS_KEY_SECRET;
+import static org.apache.paimon.rest.RESTCatalogOptions.DLF_REGION;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_SECURITY_TOKEN;
 import static org.apache.paimon.rest.RESTCatalogOptions.DLF_TOKEN_PATH;
 import static org.apache.paimon.rest.RESTCatalogOptions.TOKEN;
@@ -171,6 +172,7 @@ public void testCreateDLFAuthProviderByStsToken() throws IOException {
         options.set(DLF_ACCESS_KEY_ID.key(), token.getAccessKeyId());
         options.set(DLF_ACCESS_KEY_SECRET.key(), token.getAccessKeySecret());
         options.set(DLF_SECURITY_TOKEN.key(), token.getSecurityToken());
+        options.set(DLF_REGION.key(), "cn-hangzhou");
         AuthProvider authProvider =
                 AuthProviderFactory.createAuthProvider(AuthProviderEnum.DLF.identifier(), options);
         AuthSession session = AuthSession.fromRefreshAuthProvider(null, authProvider);
@@ -187,6 +189,7 @@ public void testCreateDLFAuthProviderByAk() throws IOException {
         DLFToken token = new DLFToken(akId, akSecret, null, null);
         options.set(DLF_ACCESS_KEY_ID.key(), token.getAccessKeyId());
         options.set(DLF_ACCESS_KEY_SECRET.key(), token.getAccessKeySecret());
+        options.set(DLF_REGION.key(), "cn-hangzhou");
         AuthProvider authProvider =
                 AuthProviderFactory.createAuthProvider(AuthProviderEnum.DLF.identifier(), options);
         AuthSession session = AuthSession.fromRefreshAuthProvider(null, authProvider);
@@ -224,7 +227,7 @@ public void testDLFAuthProviderAuthHeaderWhenDataIsNotEmpty() throws Exception {
         String fileName = UUID.randomUUID().toString();
         Pair<File, String> tokenFile2Token = generateTokenAndWriteToFile(fileName);
         String tokenStr = tokenFile2Token.getRight();
-        String serverUrl = "https://dlf.cn-hangzhou.aliyuncs.com";
+        String serverUrl = "https://dlf-cn-hangzhou.aliyuncs.com";
         AuthProvider authProvider = generateDLFAuthProvider(Optional.empty(), fileName, serverUrl);
         DLFToken token = OBJECT_MAPPER_INSTANCE.readValue(tokenStr, DLFToken.class);
         Map<String, String> parameters = new HashMap<>();
@@ -280,6 +283,7 @@ private AuthProvider generateDLFAuthProvider(
         Options options = new Options();
         options.set(DLF_TOKEN_PATH.key(), folder.getRoot().getPath() + "/" + fileName);
         options.set(RESTCatalogOptions.URI.key(), serverUrl);
+        options.set(DLF_REGION.key(), "cn-hangzhou");
         tokenRefreshInMillsOpt.ifPresent(
                 tokenRefreshInMills ->
                         options.set(TOKEN_REFRESH_TIME.key(), tokenRefreshInMills + "ms"));

paimon-core/src/test/java/org/apache/paimon/rest/auth/DLFAuthProviderFactoryTest.java

@@ -0,0 +1,51 @@
+/*
+ * 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.paimon.rest.auth;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Test for {@link DLFAuthProviderFactory}. */
+class DLFAuthProviderFactoryTest {
+
+    @Test
+    void getRegion() {
+        String region = "cn-hangzhou";
+        String ipPortUri = "http://127.0.0.1:8080";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.of(region), ipPortUri));
+        String url = "https://dlf-" + region + "-internal.aliyuncs.com";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.empty(), url));
+        url = "https://dlf-" + region + ".aliyuncs.com";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.empty(), url));
+        url = "https://dlf-pre-" + region + ".aliyuncs.com";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.empty(), url));
+        region = "us-east-1";
+        url = "https://dlf-" + region + ".aliyuncs.com";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.empty(), url));
+        url = "https://dlf-" + region + "-internal.aliyuncs.com";
+        assertEquals(region, DLFAuthProviderFactory.getRegion(Optional.empty(), url));
+        assertThrows(
+                IllegalArgumentException.class,
+                () -> DLFAuthProviderFactory.getRegion(Optional.empty(), ipPortUri));
+    }
+}