[feature] Support for kafka consumer through SASL/OAUTHBEARER

[mmalhado]

According to the Centrifugo documentation, the following authentication mechanisms are currently supported for server API kafka consumers:
"plain", "scram-sha-256", "scram-sha-512", "aws-msk-iam"

Should I assume that SASL/OAUTHBEARER authentication is not supported?

If that is the case, are there plans to get support for this authentication method?

In my particular scenario, I'm working in setting up centrifugo to connect to a GCP managed kafka instance, which recommends OAUTHBEARER :
https://cloud.google.com/managed-service-for-apache-kafka/docs/authentication-kafka

[FZambia]

Hello @mmalhado

I don't have a quick way to test it out, but I tried to prototype support for it in #1016

It loads credentials from Application Default Credentials and then extracts access token using them.

Do you have a chance to try it out? To build Centrifugo from branch - https://centrifugal.dev/docs/getting-started/installation#build-from-source (just checkout feature branch before go build).

[mmalhado]

Thanks FZambia!

I'm a bit busy the next few weeks but I'll get back to you when I get the chance to try it out 👍

[mmalhado]

I've deployed the custom Centrifugo build to test this branch, but I'm encountering a critical authentication issue with the Kafka consumer that prevents the service from starting.

Error Details:

{"level":"fatal","error":"error initializing kafka consumer (centrifugo_kafka_consumer): error init Kafka client: error ping Kafka: SASL_AUTHENTICATION_FAILED: SASL Authentication failed.: Authentication failed during authentication due to invalid credentials with SASL mechanism OAUTHBEARER","time":"2025-08-07T11:24:03Z","message":"error initializing consumers"}"

Configuration:

• Using Google OAuth Bearer ADC for SASL authentication (sasl_mechanism: "google-oauth-bearer-adc")
• TLS is enabled
• The pod is using workload identity with appropriate GCP service account annotations

Could you please look into this?

Used Config
"
...
values:
replicaCount: 1
serviceAccount:
create: false
name: centrifugo
annotations:
iam.gke.io/gcp-service-account: centrifugo@project.iam.gserviceaccount.com
...
consumers:
enabled: true
name: "centrifugo_kafka_consumer"
type: kafka
kafka:
brokers: ["kafka-broker.example.com:9092"]
topics: ["topic-name"]
consumer_group: "consumer-group-name"
sasl_mechanism: "google-oauth-bearer-adc"
tls:
enabled: true
...
"

[FZambia]

Hi, unfortunate, will try to take a look once I have time for this.

Meanwhile, maybe try some other software which supports OAUTHBEARER with ADC to make sure it's not a credentials configuration issue. Looked at the implementation - and it seems correct in general, no quick ideas what may be wrong with it.