Adds a configurable OAuth provider for Kafka SASL/OAUTHBEARER

[nikhilsinhaparseable]

and fixes the MSK IAM token refresh, which panicked at runtime on first refresh.

Provider	Token source	Required config
aws-msk (default when a region resolves)	aws-msk-iam-sasl-signer, app-managed refresh	region via --aws-region/AWS_REGION, or AWS profile/IMDS

Existing MSK IAM configs work unchanged; the provider is inferred when unset.

• Nested Handle::block_on panic in the OAUTHBEARER refresh callback (fired inside the consumer poll loop's block_on) — signer now runs on a dedicated thread with its own single-use runtime, 30s timeout.
• Invalid Kafka config aborted only with a warn! and silently skipped ingestion — now a hard startup failure when Kafka is configured.
• AWS region resolution falls back to the SDK default chain (profile/IMDS), matching where credentials come from; resolved once and cached.
• Consumer creation no longer logs the full ClientConfig (contains secrets).

rdkafka/curl + curl-static added for librdkafka's OIDC handler; aws-config added (already transitive via the signer).

Summary by CodeRabbit

• New Features

  • Added Kafka OAuth/OIDC authentication support, including AWS MSK IAM and OIDC providers.
  • Enhanced interactive Kafka setup to prompt mechanism-specific SASL settings (OAuth/OAUTHBEARER, GSSAPI, and others) and request provider/region/endpoint/credentials only when required.
  • Expanded parsing to accept hyphenated Kafka security/SASL values (e.g., SASL-SSL, OAUTH-BEARER).

• Bug Fixes

  • Kafka startup now aborts when explicitly configured, but warns and continues when not configured.
  • Avoided logging sensitive Kafka consumer configuration.
  • Applied auto.offset.reset only for earliest/latest sources.

• Tests

  • Added coverage for safe dropping of the Kafka sink worker runtime in async contexts.

[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: b1f625d9-ea03-4ce8-9932-b576f843c320

📥 Commits

Reviewing files that changed from the base of the PR and between 91318bc and 1353e07.

📒 Files selected for processing (2)

• src/connectors/kafka/config.rs
• src/interactive.rs

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

• src/interactive.rs
• src/connectors/kafka/config.rs

---

Walkthrough

Kafka security now supports AWS MSK and OIDC OAuthBearer flows, expanded parsing, mechanism-specific prompts, bounded token generation, conditional startup validation, safer logging, and shutdown-safe worker runtimes.

Changes

Kafka OAuth and configuration

Layer / File(s)	Summary
Security configuration contracts Cargo.toml, src/connectors/kafka/config.rs	Kafka features enable AWS configuration and curl support; OAuth fields and expanded protocol, mechanism, and provider parsing are added.
OAuth resolution and validation src/connectors/kafka/config.rs	Provider inference, region normalization, OIDC librdkafka settings, offset handling, bootstrap validation, and tests are updated.
AWS MSK token generation src/connectors/kafka/mod.rs	MSK token generation resolves and caches the AWS region, runs signing on a bounded runtime, and validates the AWS MSK provider.
Interactive prompting and startup handling src/interactive.rs, src/connectors/mod.rs, src/connectors/kafka/consumer.rs	Prompts branch by protocol and SASL mechanism, startup distinguishes absent from invalid Kafka configuration, and consumer secrets are not logged.
Kafka worker runtime lifecycle src/connectors/kafka/sink.rs	Kafka sink runtime ownership uses a shutdown-safe wrapper with asynchronous drop coverage.

Estimated code review effort: 4 (Complex) | ~60 minutes

Sequence Diagram(s)

sequenceDiagram
  participant librdkafka
  participant KafkaContext
  participant DefaultRegionChain
  participant MSKIAMSigner
  librdkafka->>KafkaContext: request OAuth token
  KafkaContext->>DefaultRegionChain: resolve signing region
  DefaultRegionChain-->>KafkaContext: return region
  KafkaContext->>MSKIAMSigner: generate signed token
  MSKIAMSigner-->>KafkaContext: return token
  KafkaContext-->>librdkafka: provide OAuthToken

Loading

Possibly related PRs

• parseablehq/parseable#1730: Modifies the Kafka AWS MSK SASL/OAUTHBEARER configuration and token-generation flow.

Poem

I’m a rabbit with OAuth in my ear,
AWS tokens now hop without fear.
OIDC fields join the parade,
Secrets stay tucked, never displayed.
Kafka hops along, configs align.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly matches the main change: adding configurable OAuth provider support for Kafka SASL/OAUTHBEARER.
Description check	✅ Passed	The description covers the goal, rationale, key changes, and behavior impacts, though it omits the template headings and checklist format.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
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

---

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