feat(kafka): add SASL OAUTHBEARER method=default RFC 6749 token fetch callback

[dvilaverde]

Summary

Adds sasl.oauthbearer.method=default support to the Kafka sink and source. Today the only supported external token path is method=oidc, which requires librdkafka's built-in OIDC handler to validate the access token as a JWT with a standard exp claim. That handler fails with:

Failed to acquire SASL OAUTHBEARER token: Expected JSON JWT response with "exp" field

This makes method=oidc unusable with:

• Providers returning opaque (non-JWT) access tokens
• Providers returning JWTs with non-standard expiry fields
• Custom grant types such as authorization_code with a Permanent Authorization Code (PAC)

RFC 6749 §5.1 designates expires_in in the HTTP response as the authoritative token lifetime — the JWT exp claim is not guaranteed by the OAuth2 spec.

How it works: Vector activates its own RFC 6749 token fetch callback (generate_oauth_token on ClientContext) when sasl.oauthbearer.token.endpoint.url is present in librdkafka_options and sasl.oauthbearer.method is absent or "default". When method=oidc, librdkafka's built-in OIDC handler runs unchanged. No new Vector sasl.* config keys are introduced — everything goes through existing librdkafka_options passthrough.

Extra name=value pairs in sasl.oauthbearer.config are merged into the POST body, allowing grant type overrides and additional parameters. extension_* pairs are silently dropped (SASL broker extensions per RFC 7628, not HTTP params).

Vector configuration

Standard client credentials (opaque token provider):

  librdkafka_options:
    "security.protocol": "SASL_SSL"
    "sasl.mechanism": "OAUTHBEARER"
    "sasl.oauthbearer.token.endpoint.url": "https://auth.example.com/token"
    "sasl.oauthbearer.client.id": "${CLIENT_ID}"
    "sasl.oauthbearer.client.secret": "${CLIENT_SECRET}"

Custom grant type with Permanent Authorization Code:

  librdkafka_options:
    "security.protocol": "SASL_SSL"
    "sasl.mechanism": "OAUTHBEARER"
    "sasl.oauthbearer.token.endpoint.url": "${TOKEN_URL}"
    "sasl.oauthbearer.client.id": "${CLIENT_ID}"
    "sasl.oauthbearer.client.secret": "${CLIENT_SECRET}"
    "sasl.oauthbearer.config": "grant_type=authorization_code code=${PAC}"

Existing method=oidc behavior is completely unchanged.

How did you test this PR?

• 13 unit tests covering extract_oauthbearer_config option combinations, sasl.oauthbearer.config parsing (principal, extension_*, extra params), and missing/malformed inputs
• 4 wiremock integration tests against a live HTTP mock server: standard client credentials flow, authorization_code grant type override, missing expires_in defaulting to 3600s, and unconfigured oauthbearer returning Err
• cargo test --no-default-features --features "sources-kafka,sinks-kafka" -p vector kafka -- kafka unit tests
• make fmt && make check-fmt — clean
• make check-clippy — clean

Change Type

• Bug fix
• New feature
• Dependencies
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Closes: feat(kafka sink/source): support SASL OAUTHBEARER with method=default and external token endpoint URL #25328

[github-actions]

Thank you for your contribution! Before we can merge this PR, please sign our Contributor License Agreement.

To sign, copy and post the phrase below as a new comment on this PR.

Note: If the bot says your username was not found, the email used in your git commit may not be linked to your GitHub account. Fix this at github.com/settings/emails, then comment recheck to retry.

---

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

---

1 out of 2 committers have signed the CLA.
✅ (dvilaverde)[https://github.com/dvilaverde]
❌ @david Vilaverde
David Vilaverde seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You can retrigger this bot by commenting recheck in this Pull Request. Posted by the CLA Assistant Lite bot.}

[dvilaverde]

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

[dvilaverde]

recheck

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 37ae874790

ℹ️ 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]

Gate OAuth token callback to default method only

Setting ENABLE_REFRESH_OAUTH_TOKEN to true unconditionally forces librdkafka to call Vector’s generate_oauth_token for every OAUTHBEARER client, which bypasses librdkafka’s built-in handlers. In this commit, extract_oauthbearer_config returns None for sasl.oauthbearer.method=oidc (and when no token endpoint is configured), so generate_oauth_token immediately errors on self.oauthbearer.as_ref().ok_or(...); this turns previously working OIDC/default flows into authentication failures instead of preserving existing behavior.

Useful? React with 👍 / 👎.