SASL/OAUTHBEARER Fails for GKE Workload Identity to Managed Kafka

**ERROR:** The connection consistently fails during the SASL handshake, resulting in either a client-side `KafkaError{code=_MSG_TIMED_OUT}` or a broker-side `SaslAuthenticationException: Authentication failed due to invalid credentials`.

WIF is enabled on the cluster (`workload_pool` is set) and the node pool (`mode: GKE_METADATA`). A `ServiceAccount` is annotated with `iam.gke.io/return-principal-id-as-email: “true”` and is used by the application pods. A Google Service Account (GSA) has the `roles/managedkafka.client` and `roles/iam.serviceAccountTokenCreator` roles. The Kubernetes Service Account is bound to this GSA with the `roles/iam.workloadIdentityUser` role. The Python client is configured for `SASL_SSL` with `sasl.mechanism: OAUTHBEARER` and uses `google.auth.default()` within an `oauth_cb` callback to provide the authentication token.

I have validated network connectivity (via socket tests), all IAM/GKE/K8s policies and annotations, and Kafka ACLs (including `User:*`)—every component is correctly configured as per the docs.

the authentication fails with both the direct federation (`return-principal-id-as-email` annotation) and the KSA-to-GSA impersonation (`gcp-service-account` annotation) methods of Workload Identity.

Start with the Python client's google.auth.default() oauth_cb path and the SASL/OAUTHBEARER handshake described in the report. Reproduce the timeout or invalid-credentials failure using the stated GKE Workload Identity and Managed Kafka setup; done means authentication succeeds or the supported configuration is documented.

Written by the indexing model from the issue text.