'SASL authentication error' and other logs from producer using OAuth

[SnehithReddyp]

The confluent-kafka-python Producer, when using an OAuth token and remaining idle for a period exceeding the token's lifetime, logs transient WARNING and ERROR messages, these messages unnecessarily clutter logs.

This behavior is observed with a producer configured for OAuth authentication (sasl.mechanism=OAUTHBEARER and oauth_cb). The logs occur when an existing, idle broker connection is closed by the broker after the token expires.

ssl://kafka.io:9092/bootstrap: Disconnected (after 43ms in state UP, 1 identical error(s) suppressed)
%6|1757756181.356|FAIL|rdkafka#producer-2| [thrd:sasl_ssl://kafka.io:9093/0]: sasl_ssl://kafka.io:9093/0: Disconnected (after 49ms in state UP, 1 identical error(s) suppressed)
%3|1757756212.977|FAIL|rdkafka#producer-2| [thrd:sasl_ssl://kafka.io:9092/bootstrap]: sasl_ssl://kafka.io:9092/bootstrap: SASL authentication error: {"status":"invalid_token"} (after 150ms in state AUTH_REQ)
%3|1757756213.309|FAIL|rdkafka#producer-2| [thrd:sasl_ssl://kafka.io:9094/1]: sasl_ssl://kafka.io:9094/1: SASL authentication error: {"status":"invalid_token"} (after 160ms in state AUTH_REQ)
%4|1757756213.309|METADATA|rdkafka#producer-2| [thrd:main]: sasl_ssl://kafka.io:9094/1: Metadata request failed: partition leader query: Local: Authentication failure (0ms): Permanent
%3|1757756215.338|FAIL|rdkafka#producer-2| [thrd:sasl_ssl://kafka.io:9095/2]: sasl_ssl://kafka.io:9095/2: SASL authentication error: {"status":"invalid_token"} (after 164ms in state AUTH_REQ)
%4|1757756215.338|METADATA|rdkafka#producer-2| [thrd:main]: sasl_ssl://kafka.io:9095/2: Metadata request failed: partition leader query: Local: Authentication failure (0ms): Permanent
%3|1757756217.706|FAIL|rdkafka#producer-2| [thrd:sasl_ssl://kafka.io:9093/0]: sasl_

When a new message needs to be sent, the producer is trying to reconnect. This attempt fails with the invalid_token error, which triggers the oauth_cb to get a fresh token. The producer then successfully re-authenticates and sends the new message.

The oauth_cb is triggered by a call to poll(). Is there a way to make an idle producer's token refresh more proactive, similar to how an active consumer constantly polling avoids this issue? This could potentially prevent the initial connection failure and corresponding log messages entirely.

Can you provide few suggestions for this issue other than changing my log level / using "log.connection.close": False ??

[cwegener]

Yes. Implementing this requires a lot of searching for the right information. I've compiled the relevant links here:

1. kafka-python is a wrapper for librdkafka link
2. librdkafka is a 100% asynchronous API link
3. MOST IMPORTANT (very poorly documented) The OAuth authentication runs as a callback in librdkafka's asynchronous API very important link: link
4. kafka-python currently lacks native Python asyncio integration link

Therefore, you have to carefully work with Python's primitives in order for your implementation to even function as expected, most importantly you have to ensure that librdkafka's .poll() method runs in a loop as expected, otherwise NONE of the librdkafka callbacks will fire in time.

There is a little bit of an introduction published on the Confluent blog here: link

Hope this helps

[zuerst]

I'm experiencing exactly the same issue. The SASL authentication error: [00c78861-f28b-479e-9a49-8e1db2e93f6c]: Access denied (after 330ms in state AUTH_REQ) error occurs when OAuth tokens expire.

Is there no supported way to keep the producer idle and refresh the token proactively to avoid getting this error? It would be helpful to have built-in token refresh before expiration rather than waiting for the auth failure.

[cwegener]

I'm experiencing exactly the same issue. The SASL authentication error: [00c78861-f28b-479e-9a49-8e1db2e93f6c]: Access denied (after 330ms in state AUTH_REQ) error occurs when OAuth tokens expire.

Is there no supported way to keep the producer idle and refresh the token proactively to avoid getting this error? It would be helpful to have built-in token refresh before expiration rather than waiting for the auth failure.

@zuerst Did you read my answer above? And click the links to the references? If so, let me know if anything is missing in order to better explain the architecture.

[zuerst]

@cwegener No, not at first. After you called it out, I went over the links and carefully read them over and here are some follow up questions to that. Please bare in mind, I am new to python and especially to asyncio and sync/wait.

1. In your research, it mostly mentions about async function in python but what about synchronous functions?

2. OAuth authentication runs as a callback in librdkafka's asynchronous

   • How does that impact the library refreshing the token? oauth_cb return token and expiration. At the point of the call back, can't it get a new token if it is expired?

For context, I am integrating confluent kafka producer in Celery Tasks to produce some message at the end of the task with some status. There is no async-ness to how I am planning to use. I create a producer in the task base and use it to produce a message at the end of the task.

In this case, when the application runs for an extended period without any Kafka operations (for example, when Celery tasks aren't being invoked), the OAuth token eventually expires. At this point, the Kafka producer begins logging AUTH_REQ errors.
While this behavior is functionally acceptable since the next Celery task execution triggers a flush() operation that internally calls poll(), refreshing the authentication and allowing messages to be sent successfully, it creates unnecessary error noise in the logs.
The issue is that there's no built-in mechanism to proactively refresh tokens for idle producers, leading to these benign but noisy authentication errors.

# Here is a pseudocode of my use case
class TaskBase(Task):
    _kafka_producer = None

    @property
    def kafka_producer(self):
        if self._kafka_producer is None:
            self._kafka_producer = confluent_kafka.Producer(...)
        return self._kafka_producer

    def after_return(self, *args, **kwargs):
        if self.kafka_producer:
            remaining = self.kafka_producer.flush(5)
            if remaining > 0:
                logger.warning(f'Failed to send {remaining} messages to Kafka')

@app.task(bind=True, base=TaskBase)
def my_task(self, task_id: str = None):
    ... # Do something
    self.producer.produce(...)

@app.task(bind=True, base=TaskBase)
def my_task_2(self, task_id: str = None):
    ... # Do something
    self.producer.produce(...)

[cwegener]

In your research, it mostly mentions about async function in python but what about synchronous functions?

My guess is that many real-world apps that build on Kafka are designed with concurrency in mind, so the more traditional blocking / synchronous programming models might not be as prevalent amongst the people who use Kafka. But that's pure speculation on my part.

The other point to remember is the asynchronous nature of the librdkafka API itself. So, naturally the numerous callbacks in librdkafka "impose" themselves on the Python users who use kafka-python. So, even in blocking / synchronous Python programs, you still need to be aware of librdkafkas asynchronous API (i.e. poll() loop and callbacks)

How does that impact the library refreshing the token? oauth_cb return token and expiration. At the point of the call back, can't it get a new token if it is expired?

Yup. That's exactly how the oauth_cb is implemented in librdkafka. It doesn't really matter if the callback runs explicitly via .poll() or implicitly via .flush() (which is just a wrapper around .poll() anyway). I haven't fully investigated which functions in librdkafka are actually causing the OAuth error messages being logged, but my guess is that maybe there are some sort of 'heartbeat' messages going to the Kafka brokers perhaps? And maybe those messages are the ones that are causing the error messages? Again, that's pure speculation and I would have to read through the librdkafka code to check if that is what is happening.

All in all your summary matches 100% with my experience.

What you can do to get rid of the gnarly error messages about the expired OAuth tokens in your application is:

1. Spin up a dedicated thread (using Python's concurrency patterns, i.e, in a Python synchronous program, that would be via concurrent.futures.ThreadPoolExecutor)
2. In that background thread, simply run a .poll() loop so that librdkafka can always run any of the configured callbacks in a timely fashion.

edit
PS: Just saw this excellent article which sheds more light into the situation. link

The application thread is responsible for continuously polling the reply queue for events through the rd_kafka_poll() function. This poll must be explicitly performed, or callbacks will not be executed, and internal memory buffers will overflow.

So, further to the above, it could be very possible that the confluent engineers have the importance of .poll() ingrained in their habits already, because they are probably very familiar with the internals of librdkafka. But the same is of course not true for regular Python users of the kafka-python wrapper on top of librdkafka.