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

Code before:
                                    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
                            }
                            _ => {
                                tracing::warn!(
                                    "Unknown catalog.security value: {}. Supported values: none, oauth2, google",
                                    security
                                );
                            }
                        }
                    } else {
                        // Legacy behavior: use individual OAuth2 properties if security type not specified
                        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());
                        }
                    }
                    if let Some(rest_signing_region) = &self.rest_signing_region {
                        java_catalog_configs.insert(
                            "rest.signing-region".to_owned(),
                            rest_signing_region.clone(),
                        );
                    }
                    if let Some(rest_signing_name) = &self.rest_signing_name {
                        java_catalog_configs
                            .insert("rest.signing-name".to_owned(), rest_signing_name.clone());
                    }
                    if let Some(rest_sigv4_enabled) = self.rest_sigv4_enabled {
                        java_catalog_configs.insert(
                            "rest.sigv4-enabled".to_owned(),
                            rest_sigv4_enabled.to_string(),
                        );

Review comment (security):
The Google authentication configuration (catalog.security, gcp.credentials.path, gcp.auth.scopes) is only applied when catalog type is "rest", but there's no validation to warn users if they configure these properties with a non-REST catalog type. Consider adding validation to return an error or warning if catalog.security is set with a catalog type other than "rest" or "rest_rust", similar to how other cloud storage options like "gcs.credential" are validated at lines 347-348.

Code after:
                                    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
                            }
                            _ => {
                                tracing::warn!(
                                    "Unknown catalog.security value: {}. Supported values: none, oauth2, google",
                                    security
                                );
                            }
                        }
                    } else {
                        // Legacy behavior: use individual OAuth2 properties if security type not specified
                        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());
                        }
                    }
                    if let Some(rest_signing_region) = &self.rest_signing_region {
                        java_catalog_configs.insert(
                            "rest.signing-region".to_owned(),
                            rest_signing_region.clone(),
                        );
                    }
                    if let Some(rest_signing_name) = &self.rest_signing_name {
                        java_catalog_configs