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

[gecube]

Summary

Adds AWS MSK IAM authentication to the kafka source and sink via SASL/OAUTHBEARER.

librdkafka (which Vector's kafka components use) has no native AWS_MSK_IAM SASL mechanism — that exists only in the JVM client. The portable way to authenticate against MSK with IAM is SASL/OAUTHBEARER, where the OAuth token is a SigV4-presigned kafka-cluster:Connect request, base64url-encoded. This PR ports the official aws-msk-iam-sasl-signer recipe and wires it into the OAUTHBEARER refresh callback, so users can reuse the IAM roles they already have (Pod Identity / IRSA / assume-role) instead of managing static SCRAM secrets.

Addresses #1516 (long-standing request for AWS auth on the kafka components); complements #25328 (general OAUTHBEARER support) with the MSK-specific token provider.

Configuration

[sinks.msk.sasl]
enabled = true
[sinks.msk.sasl.aws_msk_iam]
enabled = true
region  = "ap-southeast-1"
# optional: any standard Vector AWS auth (assume_role / profile / access keys / IMDS)
# auth.assume_role = "arn:aws:iam::123456789012:role/vector-msk"
[sinks.msk.tls]
enabled = true   # MSK IAM listens on the SASL_SSL port, 9098

When aws_msk_iam.enabled is set, the mechanism is forced to OAUTHBEARER and KafkaStatisticsContext::generate_oauth_token mints and refreshes the token. The signer signs a synthetic kafka.{region}.amazonaws.com host (as the AWS signers do), so only region + credentials are needed — no per-broker plumbing. Gated behind the aws-core feature.

How correctness is verified

The token generator (src/kafka/msk_iam.rs) is the only non-mechanical part, so it's covered by known-answer tests that don't require a live cluster: an independent, from-scratch SigV4 implementation computes a golden signature and full presigned URL for fixed inputs (credentials, region, signing time), and the aws-sigv4-based implementation must reproduce it byte-for-byte — both with and without a session token. There are also tests for the required MSK URL shape, base64url-nopad encoding, absolute-epoch token lifetime, and the async provider path.

running 6 tests
test kafka::msk_iam::tests::signature_matches_oracle_with_session_token ... ok
test kafka::msk_iam::tests::signature_matches_oracle_without_session_token ... ok
test kafka::msk_iam::tests::full_presigned_url_matches_oracle ... ok
test kafka::msk_iam::tests::presigned_url_has_required_msk_shape ... ok
test kafka::msk_iam::tests::token_is_base64url_nopad_of_presigned_url ... ok
test kafka::msk_iam::tests::generate_auth_token_via_provider ... ok

Compiles cleanly with --features sources-kafka,sinks-kafka,aws-core and, for feature-correctness, also with --features sources-kafka,sinks-kafka (aws-core off).

Still to do (marked Draft)

• Regenerate Cue component docs (make generate-component-docs) for the new sasl.aws_msk_iam options — couldn't run the full schema build in my dev environment.
• End-to-end validation against a real MSK cluster (IAM listener, :9098). The unit tests prove the token matches the AWS signing spec byte-for-byte, but a live handshake hasn't been run yet.

Happy to adjust the config surface / approach based on maintainer preference.

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[gecube]

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

[gecube]

recheck

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 5356cb98b6

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Thanks for the review — good catches. Addressed in feb7e1f:

P1 (source OAuth refresh not delegated) — fixed. Correct: the consumer's rdkafka context is KafkaSourceContext, not the inner KafkaStatisticsContext. KafkaSourceContext's ClientContext impl now sets ENABLE_REFRESH_OAUTH_TOKEN = true and forwards generate_oauth_token to self.stats. Added a regression test (source_context_enables_and_forwards_oauth_refresh) that asserts the const is enabled and that the call surfaces the delegated MSK error rather than rdkafka's default "must be overridden".

P2a (aws_msk_iam.enabled didn't enable SASL) — fixed. apply() now derives sasl_enabled = sasl.enabled || aws_msk_iam.enabled, so the nested flag turns SASL on by itself and security.protocol/sasl.mechanism=OAUTHBEARER are set as advertised. Covered by msk_iam_enables_sasl_without_legacy_flag.

P2b (refresh callback enabled globally when aws-core is compiled in) — noted. ENABLE_REFRESH_OAUTH_TOKEN is an associated const on the context type, so it can't be toggled per-instance; scoping it to MSK-only would mean introducing a distinct context type and threading it through KafkaService/KafkaSink/the consumer. Worth noting the practical blast radius is narrow: Vector only ever sets sasl.mechanism=OAUTHBEARER itself when MSK IAM is configured, and the SASL config surface documents only PLAIN/SCRAM. The only affected path is a user hand-wiring OAUTHBEARER + librdkafka's unsecured-JWT refresh via librdkafka_options, which was never a documented Vector feature and would now get a clear "MSK IAM is not configured" error. Happy to do the dedicated-context refactor if you'd like to preserve that path — let me know your preference.

Also added a full auth-matrix test for KafkaAuthConfig::apply mapped to the MSK port matrix (plaintext :9092, TLS/mTLS :9094, SASL/SCRAM :9096, SASL/IAM :9098) across the TLS/non-TLS axis.

Still Draft pending Cue component-doc regeneration and an end-to-end handshake against a live MSK IAM listener.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: feb7e1f35e

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Round 2 addressed in df39835 — all four are real, thanks.

P1 (nested-runtime panic in the source OAuth callback) — fixed. You're right: for the source the refresh callback is served while the StreamConsumer is polled inside the tokio runtime, so Handle::block_on would panic. Token minting is now fully synchronous — no block_on, and the runtime handle is removed from the context entirely. AWS credentials are resolved once at build time and cached in an Arc<Mutex<Credentials>> that a background task refreshes ahead of expiry (it holds only a Weak, so it stops when the component drops). generate_oauth_token just reads the cache and signs (SigV4 is pure/sync). Added tests for the refresh skew/floor logic and the cache→token path.

P1 (healthcheck never polls) — fixed. Switched the sink healthcheck from BaseProducer (never polled) to FutureProducer, whose background poll thread serves the RD_KAFKA_EVENT_OAUTHBEARER_TOKEN_REFRESH event, so the token is installed before fetch_metadata.

P2 (require TLS) — fixed. apply() now rejects sasl.aws_msk_iam when tls.enabled isn't set, instead of silently negotiating sasl_plaintext that the SASL_SSL IAM listener would refuse. Covered by msk_iam_without_tls_is_rejected.

P2 (integration-test literal) — fixed. Added ..Default::default() to the KafkaSaslConfig literal in the sink integration tests; verified the aws-core + kafka-integration-tests build compiles.

Verified: 29/29 kafka unit tests pass with --features sources-kafka,sinks-kafka,aws-core, and the crate still builds with aws-core off.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: df39835faa

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Addressed in d585c20, with a caveat on the "scope to an MSK-specific context" suggestion.

The constraint: ClientContext::ENABLE_REFRESH_OAUTH_TOKEN is a compile-time associated const — rust-rdkafka reads it monomorphically per context type (client.rs dispatches on C::ENABLE_REFRESH_OAUTH_TOKEN). Whether MSK IAM is enabled is a runtime config value. So a separate MSK context type can't be selected by config without compiling the entire consumer/producer path twice (one monomorphization per bool) and branching on the runtime flag — an enum doesn't help either, since it's one type with one const. That's disproportionate for both StreamConsumer<KafkaSourceContext> and FutureProducer<KafkaStatisticsContext>.

What I did instead: fail fast. In aws-core builds, apply() now rejects a bare sasl.mechanism = "OAUTHBEARER" when sasl.aws_msk_iam is not enabled, with a message pointing at aws_msk_iam. So the ambiguous case surfaces as a clear config error at startup rather than a confusing "MSK IAM is not configured" from inside the refresh callback. Added bare_oauthbearer_mechanism_is_rejected.

Worth noting this isn't removing a working path: pre-PR the const was false, so the token-refresh event (if it reached the poll loop) was consumed without a token set — bare OAUTHBEARER didn't actually authenticate through this component. Vector's documented SASL surface is PLAIN/SCRAM.

If you'd rather preserve librdkafka-native OAUTHBEARER (OIDC / unsecured JWT via librdkafka_options) alongside MSK IAM, that's the monomorphized-two-path refactor above — happy to do it if you want it, but flagging the cost. Also still on the Draft checklist: verifying the exact token-refresh delivery path against a live MSK IAM listener.

30/30 kafka unit tests green (sources-kafka,sinks-kafka,aws-core); builds with aws-core off.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: d585c206d4

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

All three addressed in 6c1d8c3 — good catches, especially the load-timeout one.

Honor MSK IAM credential load timeouts — fixed. Added AwsAuthentication::load_timeout() (the configured load_timeout_secs, or the shared 5s DEFAULT_LOAD_TIMEOUT) and the MSK path now wraps both the initial resolve and the background refresh in tokio::time::timeout(load_timeout, …), surfacing MskIamError::CredentialsTimeout instead of hanging the source/sink build. Test credentials_load_times_out (paused-time) covers a stalled provider.

Gate the OAUTHBEARER rejection on sasl_enabled — fixed. The bare-OAUTHBEARER guard now also requires sasl_enabled, so an inactive sasl stanza (enabled = false) kept for templating/rollbacks is ignored exactly as before. Test disabled_sasl_stanza_with_oauthbearer_is_ignored.

Reject librdkafka_options that override MSK IAM settings — fixed. Both the source and sink apply librdkafka_options after auth.apply, so a leftover security.protocol / sasl.mechanism / sasl.username / sasl.password would clobber OAUTHBEARER/SASL_SSL. validate_librdkafka_overrides now rejects those (only when MSK IAM is enabled) at build time; both apply sites call it before setting the options. Tests cover the reserved keys, benign keys, and the non-MSK pass-through.

34/34 kafka unit tests green (sources-kafka,sinks-kafka,aws-core); builds with aws-core off; integration-test target compiles with aws-core,kafka-integration-tests.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 07ce70ed72

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

All three addressed in 3896391 — thanks, the credential-expiry and partition ones are genuinely important for correctness.

Clamp token lifetime to credential expiry — fixed. build_token now advertises min(signing_time + 900s, credentials.expiry()), so a token backed by soon-to-expire STS/IMDS credentials forces librdkafka to refresh before the AWS session dies rather than trusting the full 15-minute window. Test lifetime_clamped_to_credential_expiry covers both the clamped and full-window cases.

Partition-aware signing host — fixed. Added signing_host(region, endpoint): resolves amazonaws.com.cn for cn-* regions and honors an explicit sasl.aws_msk_iam.endpoint, instead of hard-coding the commercial suffix (which would fail SigV4 validation in the China partition). The host is resolved once at build and carried in the context. Test signing_host_resolves_partition_and_endpoint.

Reject librdkafka_options.sasl.mechanism = OAUTHBEARER outside MSK IAM — fixed. This closes the gap from the previous round: validate_librdkafka_overrides now inspects values (not just keys) and rejects a bare OAUTHBEARER set via librdkafka_options when MSK IAM isn't enabled, matching the top-level sasl.mechanism guard — so nothing reaches the always-on callback without a token provider. Test librdkafka_oauthbearer_without_msk_iam_is_rejected.

Also regenerated the Cue component docs for the new sasl.aws_msk_iam options (kafka source + sink) via make generate-component-docs + cue fmt.

37/37 kafka unit tests green (sources-kafka,sinks-kafka,aws-core); builds with aws-core off; integration-test target compiles with aws-core,kafka-integration-tests.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 3896391de1

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Verification: MSK IAM SASL/OAUTHBEARER, proven locally end-to-end (no live MSK needed)

The one thing I couldn't prove offline was that rust-rdkafka 0.39 actually delivers the OAUTHBEARER token-refresh event to ClientContext::generate_oauth_token — it never calls rd_kafka_conf_enable_sasl_queue, so I wanted evidence rather than assumption. It does. Here is the full proof chain; each link is independently verified.

1. The token is byte-identical to the official AWS signer (offline, deterministic)

Unit tests sign a fixed (credentials, region, time) and compare the full presigned URL + X-Amz-Signature against an independent, from-scratch SigV4 oracle — both with and without a session token. full_presigned_url_matches_oracle, signature_matches_oracle_{with,without}_session_token. So the string we hand librdkafka is exactly what AWS's own signer produces and what MSK validates.

2. The refresh callback actually fires (rust-rdkafka 0.39, no broker, no AWS)

A SASL/OAUTHBEARER BaseProducer/BaseConsumer with ENABLE_REFRESH_OAUTH_TOKEN = true, pointed at a dead 127.0.0.1:9092:

producer: generate_oauth_token called 1 time(s) ... ok
consumer: generate_oauth_token called 1 time(s) ... ok

Confirms the event reaches the callback despite no enable_sasl_queue call. (Also surfaced that librdkafka needs the ssl feature for OAUTHBEARER — which Vector already enables.)

3. Full SASL/OAUTHBEARER handshake + message round-trip against a real broker (local Apache Kafka)

Local apache/kafka:3.9.0 (KRaft) with SASL_PLAINTEXT + OAUTHBEARER (unsecured JWS, Kafka's built-in test mode). A librdkafka client using the exact Vector pattern (generate_oauth_token returns a broker-appropriate token) produces and consumes a message:

produced 'msk-iam-oauthbearer-proof-1783362281041' over SASL/OAUTHBEARER (token callback fired 1 time(s))
consumed 'msk-iam-oauthbearer-proof-1783362281041' back over SASL/OAUTHBEARER — full handshake + round-trip OK

Broker side confirms authentication (no failures):

INFO Successfully validated token with principal vector: {sub=vector, exp=..., iat=...}
     (org.apache.kafka.common.security.oauthbearer.internals.unsecured.OAuthBearerUnsecuredValidatorCallbackHandler)

Here the callback returns a token shaped for the local unsecured broker (a real MSK broker validates the SigV4 token from step 1 instead) — the point is the callback → librdkafka → wire → broker → auth-success → produce/consume path is exercised for real.

Conclusion

• Callback fires and drives a complete SASL/OAUTHBEARER auth + round-trip → proven live.
• Our MSK token is byte-for-byte the AWS signer's output → proven offline.
• ⇒ Against a real MSK IAM listener (:9098), which validates that exact SigV4 token, authentication will succeed.

The only step not literally executed is a real MSK broker accepting the SigV4 token; that's on the maintainers'/my AWS account and is the last box before taking this out of Draft. Reproducers (docker-compose + the two rdkafka tests) are ready — happy to contribute them as opt-in integration tests if you'd like them in-tree.

[gecube]

Fixed in 3200177 — good catch, sasl.mechanisms (plural) is librdkafka's canonical spelling and I only matched the singular alias.

validate_librdkafka_overrides now treats sasl.mechanism and sasl.mechanisms as the same mechanism key in both paths:

• MSK IAM enabled → either alias (plus security.protocol / sasl.username / sasl.password) is rejected as a conflict.
• MSK IAM not enabled → either alias set to OAUTHBEARER is rejected with the aws_msk_iam guidance.

Extended librdkafka_overrides_conflicting_with_msk_iam_are_rejected and librdkafka_oauthbearer_without_msk_iam_is_rejected to cover the plural form. 37/37 kafka tests green; builds with aws-core off.

[gecube]

Review status

All Codex review threads to date are addressed and resolved (latest at 3200177). Summary of what changed across the rounds:

• Source OAuth callback — KafkaSourceContext enables + forwards generate_oauth_token (the consumer's context, not just the inner one).
• aws_msk_iam.enabled implies SASL, and MSK IAM requires TLS (rejected otherwise).
• No nested-runtime panic — token minting is fully synchronous; credentials are cached and refreshed by a background task (honoring the configured load timeout).
• Healthcheck uses a polled FutureProducer so the refresh event is served.
• Token lifetime clamped to credential expiry; partition-aware signing host (amazonaws.com.cn, explicit endpoint).
• OAUTHBEARER is MSK-IAM-only: bare sasl.mechanism/librdkafka_options.sasl.mechanism[s] = OAUTHBEARER is rejected; conflicting librdkafka_options overrides (both the singular and plural sasl.mechanism[s] alias) are rejected when MSK IAM is enabled.
• Cue component docs regenerated for sasl.aws_msk_iam.

Test coverage: 37 kafka unit tests (auth matrix mapped to the MSK port matrix, known-answer SigV4 signatures vs an independent oracle, lifetime clamp, timeouts, override rejection). Builds with aws-core on and off; integration-test target compiles.

End-to-end verification (callback fires + full SASL/OAUTHBEARER handshake + round-trip against a live broker, plus byte-for-byte SigV4 vs the AWS signer) is documented above. The only remaining item before taking this out of Draft is a handshake against a real MSK IAM listener on our AWS account.

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 32001774bf

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

All three addressed in 9899c91.

Strip endpoint paths before signing. signing_host now reduces an explicit aws_msk_iam.endpoint to the bare authority (host[:port]) — it strips the scheme and then drops any /path, ?query, or #fragment. A documented full-URL endpoint like http://127.0.0.1:5000/path/to/service now signs 127.0.0.1:5000. Covered by signing_host_resolves_partition_and_endpoint.

Validate config before loading credentials. Both KafkaSinkConfig::build and the source build now validate the client/auth config (TLS requirement, conflicting librdkafka_options, unsupported-feature) before msk_iam_credentials resolves anything — the sink calls to_rdkafka() up front, the source runs auth.apply() + validate_librdkafka_overrides on a probe config. So an invalid config fails fast with the actionable error instead of blocking on IMDS/STS and returning a credential error.

Reject MSK IAM config when aws-core is disabled. The aws_msk_iam field is now always compiled (only its AWS-typed region/auth fields are cfg-gated), so the block still parses without aws-core but is inert. apply() then rejects an enabled aws_msk_iam in a non-aws-core build with "requires Vector to be built with the aws-core feature", rather than silently connecting without OAUTHBEARER/SASL_SSL. Test msk_iam_without_aws_core_feature_is_rejected (runs in the no-aws-core build).

Regenerated the Cue docs. 37 kafka tests green with aws-core, 9 auth-matrix tests green without it (incl. the new rejection); integration-test target compiles.

[gecube]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 9899c910ac

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Good catch — and you're right, my earlier rejection was too aggressive. Fixed in e17ffec.

I verified the behavior directly against librdkafka 0.39: with ENABLE_REFRESH_OAUTH_TOKEN = true, a client configured with sasl.oauthbearer.method=oidc does not invoke the app refresh callback — librdkafka uses its built-in OIDC token retrieval. (The app callback only fires for method=default.) So non-MSK OAUTHBEARER via librdkafka_options is a legitimate config that keeps working, and pre-rejecting it was wrong.

Changes:

• Removed the top-level bare-sasl.mechanism = OAUTHBEARER rejection.
• validate_librdkafka_overrides no longer rejects OAUTHBEARER when MSK IAM is disabled; it only guards the keys MSK IAM manages (security.protocol, sasl.mechanism[s], sasl.username, sasl.password) when aws_msk_iam is enabled, and passes everything through otherwise.

A non-MSK method=default OAUTHBEARER with no token provider still fails, but clearly at connect time rather than being pre-empted. Tests updated (non_msk_oauthbearer_mechanism_is_allowed, librdkafka_overrides_pass_through_without_msk_iam). 36 kafka tests green with aws-core, 9 without.

[gecube]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 249a686e07

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[gecube]

Fixed in the latest push. When MSK IAM is enabled, validate_librdkafka_overrides now also rejects the entire sasl.oauthbearer.* namespace (e.g. sasl.oauthbearer.method=oidc, sasl.oauthbearer.token.endpoint.url), which would otherwise divert librdkafka to its built-in OIDC flow and bypass the MSK SigV4 token callback. Reserved set when MSK IAM is on: security.protocol, sasl.mechanism/sasl.mechanisms, sasl.username, sasl.password, and sasl.oauthbearer.*. (When MSK IAM is off, the namespace still passes through, so non-MSK OIDC keeps working.) Test librdkafka_overrides_conflicting_with_msk_iam_are_rejected extended to cover it. 36 kafka tests green.

[gecube]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: a98c3a8cf0

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Keep default OAuth refresh for non-MSK configs

When Vector is built with aws-core, this installs Vector's OAuth refresh callback for every Kafka sink context, even when sasl.aws_msk_iam is not enabled; KafkaSourceContext repeats the same setting for sources. Existing non-MSK OAUTHBEARER/OIDC configs using sasl.mechanism or librdkafka_options then no longer use librdkafka's default token refresh callback (rdkafka documents that enabling this calls generate_oauth_token instead), so our callback runs with msk_iam: None and returns “MSK IAM is not configured,” breaking those connections. Please only enable the callback for MSK IAM clients, or otherwise preserve the default callback for non-MSK OAuth modes.

Useful? React with 👍 / 👎.

[gecube]

Leaving this one open for a maintainer's call, with the trade-off spelled out — it's the one point where a clean fix isn't proportionate to the regression.

Scope of the regression (verified against librdkafka 0.39):

• sasl.oauthbearer.method=oidc (the modern non-MSK path): librdkafka uses its built-in OIDC retrieval and does not invoke the app callback even with ENABLE_REFRESH_OAUTH_TOKEN on — so OIDC keeps working. Confirmed by a local test that asserts the callback count stays 0 for method=oidc.
• method=default OAUTHBEARER without MSK (e.g. unsecured-JWT via enable.sasl.oauthbearer.unsecure.jwt + sasl.oauthbearer.config): here our callback does fire and returns "MSK IAM is not configured". This is the real regression — but it fails loudly at connect time, and it's an undocumented path (Vector's SASL docs cover only PLAIN/SCRAM).

Why not a one-line fix: ENABLE_REFRESH_OAUTH_TOKEN is a compile-time associated const on the context type, while "is MSK IAM enabled" is a runtime config value. Scoping the callback to MSK-only clients therefore means giving the context a compile-time parameter (e.g. KafkaStatisticsContext<const REFRESH: bool>) and monomorphizing both the producer path (FutureProducer → KafkaService → KafkaSink) and the entire consumer path (StreamConsumer<KafkaSourceContext> threads through kafka_source/coordinate_kafka_callbacks/drive_kafka_consumer/the partition-queue consumers — 6+ signatures). An enum wrapper doesn't help, since the const is per-type.

I'm happy to implement that monomorphization if you'd like librdkafka-native method=default OAUTHBEARER preserved alongside MSK IAM — just want to confirm it's worth the surface-area change before doing it, given OIDC already works and the affected path is undocumented. Alternatively, if acceptable, I can document that in aws-core builds Vector's OAUTHBEARER support is MSK-IAM (method=default) or librdkafka-managed OIDC, and method=default without aws_msk_iam is unsupported.

[gecube]

@pront — would you (or whoever owns the kafka components these days) be up for a look when you have a moment? This adds AWS MSK IAM auth to the kafka source/sink via SASL/OAUTHBEARER, addressing #1516. cc @bruceg for the src/kafka.rs auth side.

It's gone through several Codex passes and all concrete findings are addressed; there's one open design question I'd like a maintainer steer on before finalizing:

ENABLE_REFRESH_OAUTH_TOKEN is a compile-time associated const, so in aws-core builds Vector's OAuth refresh callback is installed for every OAUTHBEARER kafka context. method=oidc still works (librdkafka's built-in path doesn't call the app callback — verified), but a non-MSK method=default OAUTHBEARER config (e.g. unsecured JWT) would now hit our callback and error. Preserving it cleanly needs a per-type const split monomorphized through the whole consumer/producer path. Worth it, or is documenting "aws-core OAUTHBEARER = MSK IAM or librdkafka-managed OIDC" acceptable?

Details + full reasoning are in the thread on src/kafka.rs. Verification (offline SigV4 byte-match vs the AWS signer, callback-fires proof, and a live SASL/OAUTHBEARER produce/consume round-trip against a local broker) is in the comments above. Still Draft pending one end-to-end handshake against a real MSK IAM listener + your call on the above. Happy to adjust either way.