feat(source): use the default_provider for AWS-related connectors (#1…

…7933)

Co-authored-by: Phil Sheets <[email protected]>
Co-authored-by: psheets <[email protected]>

src/connector/src/connector_common/common.rs

@@ -46,6 +46,10 @@ pub const PRIVATE_LINK_TARGETS_KEY: &str = "privatelink.targets";
 
 const AWS_MSK_IAM_AUTH: &str = "AWS_MSK_IAM";
 
+/// The environment variable to disable using default credential from environment.
+/// It's recommended to set this variable to `false` in cloud hosting environment.
+const DISABLE_DEFAULT_CREDENTIAL: &str = "DISABLE_DEFAULT_CREDENTIAL";
+
 #[derive(Debug, Clone, Deserialize)]
 pub struct AwsPrivateLinkItem {
     pub az_id: Option<String>,
@@ -57,6 +61,7 @@ use aws_config::sts::AssumeRoleProvider;
 use aws_credential_types::provider::SharedCredentialsProvider;
 use aws_types::region::Region;
 use aws_types::SdkConfig;
+use risingwave_common::util::env_var::env_var_is_true;
 
 /// A flatten config map for aws auth.
 #[derive(Deserialize, Debug, Clone, WithOptions)]
@@ -104,7 +109,7 @@ impl AwsAuthProps {
         }
     }
 
-    fn build_credential_provider(&self) -> ConnectorResult<SharedCredentialsProvider> {
+    async fn build_credential_provider(&self) -> ConnectorResult<SharedCredentialsProvider> {
         if self.access_key.is_some() && self.secret_key.is_some() {
             Ok(SharedCredentialsProvider::new(
                 aws_credential_types::Credentials::from_keys(
@@ -113,6 +118,10 @@ impl AwsAuthProps {
                     self.session_token.clone(),
                 ),
             ))
+        } else if !env_var_is_true(DISABLE_DEFAULT_CREDENTIAL) {
+            Ok(SharedCredentialsProvider::new(
+                aws_config::default_provider::credentials::default_provider().await,
+            ))
         } else {
             bail!("Both \"access_key\" and \"secret_key\" are required.")
         }
@@ -140,7 +149,7 @@ impl AwsAuthProps {
     pub async fn build_config(&self) -> ConnectorResult<SdkConfig> {
         let region = self.build_region().await?;
         let credentials_provider = self
-            .with_role_provider(self.build_credential_provider()?)
+            .with_role_provider(self.build_credential_provider().await?)
             .await?;
         let mut config_loader = aws_config::from_env()
             .region(region)