[#9564] fix(authn): Fix Flink connector oauth2 mode

Author: jerqi

docs/flink-connector/flink-authentication-with-gravitino.md

@@ -39,7 +39,7 @@ table.catalog-store.gravitino.gravitino.metalake: my_metalake
 table.catalog-store.gravitino.gravitino.client.auth.type: oauth2
 table.catalog-store.gravitino.gravitino.client.oauth2.serverUri: https://oauth-server.example.com
 table.catalog-store.gravitino.gravitino.client.oauth2.tokenPath: /oauth/token
-table.catalog-store.gravitino.gravitino.client.oauth2.credential: your-client-credentials
+table.catalog-store.gravitino.gravitino.client.oauth2.credential: client-id:client-secret
 table.catalog-store.gravitino.gravitino.client.oauth2.scope: your-scope
 ```
 

flink-connector/flink-runtime/build.gradle.kts

@@ -72,6 +72,7 @@ shadowJarTask.configure {
   relocate("com.google", "org.apache.gravitino.shaded.com.google")
   relocate("google", "org.apache.gravitino.shaded.google")
   relocate("org.apache.hc", "org.apache.gravitino.shaded.org.apache.hc")
+  relocate("org.apache.commons.lang3", "org.apache.gravitino.shaded.org.apache.commons.lang3")
 }
 
 tasks.test {

flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/GravitinoCatalogManager.java

@@ -25,9 +25,7 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.Catalog;
-import org.apache.gravitino.auth.AuthenticatorType;
 import org.apache.gravitino.client.DefaultOAuth2TokenProvider;
 import org.apache.gravitino.client.GravitinoAdminClient;
 import org.apache.gravitino.client.GravitinoMetalake;
@@ -66,7 +64,7 @@ private GravitinoCatalogManager(
 
     // Only OAuth is explicitly configured; otherwise follow Flink security (Kerberos if enabled,
     // simple auth otherwise).
-    if (AuthenticatorType.OAUTH.name().equalsIgnoreCase(authType)) {
+    if (GravitinoCatalogStoreFactoryOptions.OAUTH2.equalsIgnoreCase(authType)) {
       this.gravitinoClient = buildOAuthClient(gravitinoUri, gravitinoClientConfig);
     } else {
       if (authType != null) {
@@ -247,14 +245,19 @@ private static GravitinoAdminClient buildOAuthClient(
     String credential = config.get(GravitinoCatalogStoreFactoryOptions.OAUTH2_CREDENTIAL);
     String path = config.get(GravitinoCatalogStoreFactoryOptions.OAUTH2_TOKEN_PATH);
     String scope = config.get(GravitinoCatalogStoreFactoryOptions.OAUTH2_SCOPE);
-    Preconditions.checkArgument(
-        StringUtils.isNoneBlank(serverUri, credential, path, scope),
-        String.format(
-            "OAuth2 authentication requires: %s, %s, %s, and %s",
+
+    // Remove OAuth config from client config map, Because
+    // GravitinoAdminClient.builder().withClientConfig() does not recognize these keys.
+    Set<String> oauthConfigKeys =
+        Sets.newHashSet(
+            GravitinoCatalogStoreFactoryOptions.AUTH_TYPE,
             GravitinoCatalogStoreFactoryOptions.OAUTH2_SERVER_URI,
             GravitinoCatalogStoreFactoryOptions.OAUTH2_CREDENTIAL,
             GravitinoCatalogStoreFactoryOptions.OAUTH2_TOKEN_PATH,
-            GravitinoCatalogStoreFactoryOptions.OAUTH2_SCOPE));
+            GravitinoCatalogStoreFactoryOptions.OAUTH2_SCOPE);
+    for (String key : oauthConfigKeys) {
+      config.remove(key);
+    }
 
     DefaultOAuth2TokenProvider provider =
         DefaultOAuth2TokenProvider.builder()

flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/store/GravitinoCatalogStoreFactoryOptions.java

@@ -48,6 +48,7 @@ private GravitinoCatalogStoreFactoryOptions() {}
           .withDescription("The config of Gravitino client");
 
   public static final String AUTH_TYPE = "gravitino.client.auth.type";
+  public static final String OAUTH2 = "oauth2";
 
   // OAuth2 config keys
   public static final String OAUTH2_SERVER_URI = "gravitino.client.oauth2.serverUri";