TimurRakhmatullin86 opened a new pull request, #69427: URL: https://github.com/apache/airflow/pull/69427
## Problem The Apache Kafka provider already injects an `oauth_cb` token callback for Google Managed Kafka, detected from the bootstrap servers hostname (`managedkafka…cloud.goog`). Amazon MSK offers the same OAUTHBEARER-based IAM authentication, and **MSK Serverless supports IAM only** — but there was no equivalent branch. Users had to wire the MSK IAM token signer by hand for every connection, with no supported extra to pull the dependency. ## Solution Add a symmetric AWS branch to `KafkaBaseHook.get_conn`. When the bootstrap servers point at an Amazon MSK endpoint (`*.kafka.<region>.amazonaws.com` or `*.kafka-serverless.<region>.amazonaws.com`, including `.cn` regions) **and** `sasl.mechanism` is `OAUTHBEARER`, the hook: - derives the AWS region from the hostname (normalised to lower case for the SigV4 scope); - installs an `oauth_cb` backed by `aws_msk_iam_sasl_signer.MSKAuthTokenProvider.generate_auth_token(region)`, converting the signer's millisecond expiry to the seconds expected by confluent-kafka; - never overwrites an explicit user-provided `oauth_cb`; - raises `AirflowOptionalProviderFeatureException` with an install hint when the signer library is missing. A new `msk` optional dependency (`aws-msk-iam-sasl-signer-python>=1.0.1`) is added, mirroring the existing `google` extra, and `uv.lock` is regenerated accordingly. This mirrors the pattern established for Google Managed Kafka in #47056 and #48926. ## Tests + Docs - Unit tests cover provisioned/serverless detection, region parsing/normalisation, the no-injection paths (non-MSK host, non-OAUTHBEARER mechanism), preservation of a user `oauth_cb`, the millisecond→second conversion in the callback, the missing-library error, and a parametrised check of the host-matching regex (including a look-alike host that must not match). - New "Amazon MSK with IAM authentication" section in the Kafka connection docs. ## Notes for reviewers - The MSK host pattern is anchored to a host boundary so a look-alike hostname (e.g. `...amazonaws.com.example.com`) cannot cause the hook to sign an IAM token for an untrusted broker. - Token generation uses the standard AWS credential chain (env / files / instance role), mirroring how the GCP branch relies on ambient credentials. Profile / role-ARN variants were intentionally left out to keep the change minimal and symmetric; happy to add them if preferred. - `KafkaBaseHook.test_connection()` builds an `AdminClient` directly from the connection extras and bypasses both the GCP and MSK injections. This is pre-existing behaviour and is left as a possible follow-up rather than expanded here. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
