feat(kafka source, kafka sink): support AWS MSK IAM auth via SASL/OAUTHBEARER

Cargo.toml

@@ -650,6 +650,7 @@ aws-core = [
   "dep:aws-smithy-runtime",
   "dep:aws-smithy-runtime-api",
   "dep:aws-sdk-sts",
+  "dep:base64",
 ]
 
 # Anything that requires Protocol Buffers.

changelog.d/kafka_msk_iam_auth.feature.md

@@ -0,0 +1,3 @@
+The `kafka` source and sink now support AWS MSK IAM authentication via SASL/OAUTHBEARER. Configure it under `sasl.aws_msk_iam` with a region and the standard AWS authentication options (static keys, profile, assumed role, or IMDS). The OAuth token is a SigV4-presigned `kafka-cluster:Connect` request, generated and refreshed automatically, so no static SCRAM credentials are needed. Requires TLS (MSK IAM listens on the SASL_SSL port, `9098`).
+
+authors: gecube

src/aws/auth.rs

@@ -212,6 +212,23 @@ fn default_profile() -> String {
 }
 
 impl AwsAuthentication {
+    /// The effective credential load timeout for this authentication mechanism (the configured
+    /// `load_timeout_secs`, or the shared 5s default). Mirrors the timeout applied by
+    /// [`AwsAuthentication::credentials_cache`] for components that resolve credentials directly.
+    pub(crate) fn load_timeout(&self) -> Duration {
+        match self {
+            AwsAuthentication::Role {
+                load_timeout_secs, ..
+            }
+            | AwsAuthentication::Default {
+                load_timeout_secs, ..
+            } => load_timeout_secs
+                .map(Duration::from_secs)
+                .unwrap_or(DEFAULT_LOAD_TIMEOUT),
+            _ => DEFAULT_LOAD_TIMEOUT,
+        }
+    }
+
     /// Creates the identity cache to store credentials based on the authentication mechanism chosen.
     pub(super) async fn credentials_cache(&self) -> crate::Result<SharedIdentityCache> {
         match self {