Repository: risingwavelabs/risingwave
File: src/connector/src/connector_common/iceberg/mod.rs (Rust)

Code before:
            }

            match self.catalog_type() {
                "rest" => {
                    // Handle security type for REST catalog (Iceberg 1.10+)
                    if let Some(security) = &self.catalog_security {
                        match security.to_lowercase().as_str() {
                            "google" => {
                                // Google AuthManager (Iceberg 1.10+) - uses Google ADC or service account
                                java_catalog_configs.insert(
                                    "rest.auth.type".to_owned(),
                                    "org.apache.iceberg.gcp.auth.GoogleAuthManager".to_owned(),
                                );
                                // Set GCP credentials path if provided
                                if let Some(gcp_credentials_path) = &self.gcp_credentials_path {
                                    java_catalog_configs.insert(
                                        "gcp.auth.credentials-path".to_owned(),
                                        gcp_credentials_path.clone(),
                                    );
                                }
                                // Set GCP auth scopes if provided
                                if let Some(gcp_auth_scopes) = &self.gcp_auth_scopes {
                                    java_catalog_configs.insert(
                                        "gcp.auth.scopes".to_owned(),
                                        gcp_auth_scopes.clone(),
                                    );
                                }
                            }
                            "oauth2" => {
                                // Standard OAuth2 authentication
                                if let Some(credential) = &self.catalog_credential {
                                    java_catalog_configs
                                        .insert("credential".to_owned(), credential.clone());
                                }
                                if let Some(token) = &self.catalog_token {
                                    java_catalog_configs.insert("token".to_owned(), token.clone());
                                }
                                if let Some(oauth2_server_uri) = &self.catalog_oauth2_server_uri {
                                    java_catalog_configs.insert(
                                        "oauth2-server-uri".to_owned(),
                                        oauth2_server_uri.clone(),
                                    );
                                }
                                if let Some(scope) = &self.catalog_scope {
                                    java_catalog_configs.insert("scope".to_owned(), scope.clone());
                                }
                            }
                            "none" | "" => {
                                // No authentication
                            }
                            _ => {

Review comment (security):
I have took a look at the implementation of `GoogleAuthManager` and it does require a  `gcp.auth.credentials-path` to pass the credentails, however, in RisingWave especially in cloud, users are unable to upload such a file. So I think we might consider to support a `GoogleAuthManagerV2` to allow receive credentials content directly. It also aligns with RisingWave's gcs warehouse credentials provided way. https://github.com/apache/iceberg/blob/40c0a73c6513711b2a58bb159cf8c562fe56c826/gcp/src/main/java/org/apache/iceberg/gcp/auth/GoogleAuthManager.java#L85-L88

Code after:
            if let Some(secret_key) = &self.s3_secret_key {
                java_catalog_configs.insert("s3.secret-access-key".to_owned(), secret_key.clone());
            }

            if let Some(path_style_access) = &self.s3_path_style_access {
                java_catalog_configs.insert(
                    "s3.path-style-access".to_owned(),
                    path_style_access.to_string(),
                );
            }

            let headers = self.headers()?;
            for (header_name, header_value) in headers {
                java_catalog_configs.insert(format!("header.{}", header_name), header_value);
            }

            match self.catalog_type() {
                "rest" => {
                    // Handle security type for REST catalog (Iceberg 1.10+)
                    if let Some(security) = &self.catalog_security {
                        match security.to_lowercase().as_str() {
                            "google" => {
                                // Google AuthManager (Iceberg 1.10+) - uses Google ADC
                                java_catalog_configs.insert(
                                    "rest.auth.type".to_owned(),
                                    "org.apache.iceberg.gcp.auth.GoogleAuthManager".to_owned(),
                                );
                                // Set GCP auth scopes if provided
                                if let Some(gcp_auth_scopes) = &self.gcp_auth_scopes {
                                    java_catalog_configs.insert(
                                        "gcp.auth.scopes".to_owned(),
                                        gcp_auth_scopes.clone(),
                                    );
                                }
                            }
                            "oauth2" => {
                                // Standard OAuth2 authentication
                                if let Some(credential) = &self.catalog_credential {
                                    java_catalog_configs
                                        .insert("credential".to_owned(), credential.clone());
                                }
                                if let Some(token) = &self.catalog_token {
                                    java_catalog_configs.insert("token".to_owned(), token.clone());
                                }
                                if let Some(oauth2_server_uri) = &self.catalog_oauth2_server_uri {
                                    java_catalog_configs.insert(
                                        "oauth2-server-uri".to_owned(),
                                        oauth2_server_uri.clone(),
                                    );
                                }
                                if let Some(scope) = &self.catalog_scope {