feat(kafka): add AWS MSK IAM authentication via SASL/OAUTHBEARER

[piclemx]

Summary

The Kafka connector enumerated the OAUTHBEARER SASL mechanism but never implemented token generation — generate_oauth_token returned "not implemented" and ENABLE_REFRESH_OAUTH_TOKEN was false. As a result Parseable could not ingest from AWS MSK clusters using IAM authentication; only PLAIN, SCRAM, and GSSAPI worked.

This PR implements SASL/OAUTHBEARER for AWS MSK IAM:

• Mints IAM auth tokens via the aws-msk-iam-sasl-signer crate, driven from the synchronous librdkafka OAuth refresh callback through a captured Tokio runtime handle. The callback runs on a librdkafka background thread (not a runtime worker), so blocking on the async signer there is safe.
• Enables ClientContext::ENABLE_REFRESH_OAUTH_TOKEN. librdkafka only invokes the callback for the OAUTHBEARER mechanism, so other mechanisms are unaffected.
• Adds an --aws-region / P_KAFKA_AWS_REGION option, falling back to the standard AWS_REGION / AWS_DEFAULT_REGION environment variables.
• Makes security validation mechanism-aware: OAUTHBEARER requires only a resolvable region (credentials come from the default AWS provider chain — IRSA, EC2 instance profile, env vars, etc.), and SASL_SSL no longer requires client certificates, which are a mutual-TLS concern rather than a SASL one.

No secrets are added to configuration — credentials are resolved from the ambient AWS provider chain at token-refresh time.

Testing

• cargo check, cargo clippy, and cargo fmt --check all clean with --features kafka.
• Added unit tests covering the new validation branches and the OAUTHBEARER rdkafka config mapping (6 tests, all passing).
• Not yet exercised against a live MSK cluster.

Notes

• Pulls in the AWS SDK crate tree (region + credential provider) as new dependencies, gated behind the existing kafka feature.

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Added AWS MSK IAM authentication for Kafka using SASL/OAUTHBEARER.
  • Introduced AWS region configuration via command line or environment variables, including automatic region normalization.
  • Kafka can now generate and refresh IAM authentication tokens for OAuth connections.

• Bug Fixes

  • Strengthened Kafka security validation with mechanism-specific credential checks (e.g., OAuthBearer, SCRAM, GSSAPI, and plain modes).
  • Enforced correct TLS/client-certificate requirements and clarified misconfiguration errors.

[github-actions]

CLA Assistant Lite bot All contributors have signed the CLA ✍️ ✅

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: f6cf6251-1817-4c2c-923c-678dfb6e0a7a

📥 Commits

Reviewing files that changed from the base of the PR and between c19477a and fd22621.

📒 Files selected for processing (1)

• src/connectors/kafka/config.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/connectors/kafka/config.rs

---

Walkthrough

Adds AWS MSK IAM OAuth bearer support to the Kafka connector through optional dependencies, AWS region configuration and validation, runtime-backed token generation, librdkafka refresh integration, and unit tests.

Changes

AWS MSK IAM Kafka authentication

Layer / File(s)	Summary
AWS authentication configuration and validation Cargo.toml, src/connectors/kafka/config.rs	Adds AWS signing dependencies, AWS region resolution, mechanism-specific security validation, updated defaults, and tests covering OAuth bearer and SASL credential behavior.
OAuth token refresh callback src/connectors/kafka/mod.rs	Enables librdkafka OAuth refreshes, captures a Tokio runtime handle, and generates AWS MSK IAM tokens through the configured AWS region.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Librdkafka
  participant KafkaContext
  participant TokioRuntime
  participant AwsMskIamSigner

  Librdkafka->>KafkaContext: Request OAuth token refresh
  KafkaContext->>KafkaContext: Resolve AWS region and validate OAuthBearer
  KafkaContext->>TokioRuntime: Run token generation
  TokioRuntime->>AwsMskIamSigner: Generate MSK IAM auth token
  AwsMskIamSigner-->>TokioRuntime: Return token and expiry
  TokioRuntime-->>KafkaContext: Return signed token
  KafkaContext-->>Librdkafka: Return OAuthToken

Loading

Poem

I’m a rabbit with tokens to hop,
AWS signs them at every Kafka stop.
Regions resolve, credentials align,
Runtime bridges the callback line.
OAuth refreshes now bloom—
sniff sniff—IAM clears the room!

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely summarizes the main change: AWS MSK IAM auth via SASL/OAUTHBEARER.
Description check	✅ Passed	The description covers the goal, approach, key changes, testing, and notes, though it doesn't follow the template exactly.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (1)

src/connectors/kafka/config.rs (1)

1077-1090: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Make the missing-region test deterministic.

When either standard AWS region variable exists, this test performs no assertion and passes without exercising validation. Isolate the environment or extract region-source resolution so the failure path is always tested.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/connectors/kafka/config.rs` around lines 1077 - 1090, Make
oauthbearer_requires_a_region deterministic by isolating or controlling
AWS_REGION and AWS_DEFAULT_REGION before invoking SecurityConfig::validate,
ensuring both region sources are absent and the test always asserts validation
failure. Preserve the existing OAuthBearer/SaslSsl configuration while avoiding
environment-dependent conditional assertions.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/connectors/kafka/config.rs`:
- Around line 814-819: Update resolved_aws_region so each candidate source is
filtered for non-empty values before applying or_else precedence. Ensure an
empty explicit aws_region does not prevent fallback to AWS_REGION or
AWS_DEFAULT_REGION, while preserving the existing source priority and
Option<String> return type.
- Around line 841-855: Update the SASL/OAUTHBEARER validation in the
protocol-checking logic around self.protocol and self.sasl_mechanism to reject
SecurityProtocol::SaslPlaintext, while preserving the existing AWS region
validation for SecurityProtocol::SaslSsl. Return a clear validation error
stating that OAUTHBEARER requires SASL_SSL.

---

Nitpick comments:
In `@src/connectors/kafka/config.rs`:
- Around line 1077-1090: Make oauthbearer_requires_a_region deterministic by
isolating or controlling AWS_REGION and AWS_DEFAULT_REGION before invoking
SecurityConfig::validate, ensuring both region sources are absent and the test
always asserts validation failure. Preserve the existing OAuthBearer/SaslSsl
configuration while avoiding environment-dependent conditional assertions.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Repository UI

Review profile: CHILL

Plan: Pro

Run ID: 273e835e-44bd-4197-a782-747e706695b0

📥 Commits

Reviewing files that changed from the base of the PR and between 879a6f6 and c19477a.

⛔ Files ignored due to path filters (1)

• Cargo.lock is excluded by !**/*.lock

📒 Files selected for processing (3)

• Cargo.toml
• src/connectors/kafka/config.rs
• src/connectors/kafka/mod.rs

[piclemx]

I have read the CLA Document and I hereby sign the CLA