Version 2.13.0 blocks in SerializingProducer constructor

[IgorMilavec]

def oauth_cb(oauth_config):
    region = os.getenv('AWS_REGION') or os.getenv('AWS_DEFAULT_REGION') or 'eu-west-1'
    auth_token, expiry_ms = MSKAuthTokenProvider.generate_auth_token(region)
    return auth_token, expiry_ms / 1000 - 60

...

    conf = {
        'bootstrap.servers': broker_url,
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'OAUTHBEARER',
        'oauth_cb': oauth_cb,
        "error_cb": error_cb,
        'key.serializer': StringSerializer('utf_8'),
        'value.serializer': json_serializer,
        'socket.keepalive.enable': True,
    }

    return SerializingProducer(conf)

The same code works with 2.12.2.
Possible cause #2130 ?

[ojasvajain]

Hi Igor Milavec (@IgorMilavec), the constructor has been modified to wait for a maximum of 10 secs for the OAuth token to be set. An error is expected to be returned if the token could not be set in that time. In your case, did you receive any error after 10s or the constructor is getting blocked indefinitely?

[IgorMilavec]

In my case the constructor is blocked indefinitely (I waited for more than an hour). I did some additional tracing and actually oauth_cb blocks in the call to MSKAuthTokenProvider.generate_auth_token, so it seems that the wait in the constructor is not time bound?

[ojasvajain]

I tried this snippet and received an error after 10 seconds.

from time import time, sleep

from confluent_kafka import SerializingProducer
from confluent_kafka.serialization import StringSerializer


def oauth_cb(oauth_config):
    print ("Oauth callback fn")
    sleep(11)
    return "123", int(time()) + 10

conf = {
    'bootstrap.servers': 'localhost:50595',
    'security.protocol': 'SASL_PLAINTEXT',
    'sasl.mechanism': 'OAUTHBEARER',
    'oauth_cb': oauth_cb,
    'key.serializer': StringSerializer('utf_8'),
    'socket.keepalive.enable': True,
}


p = SerializingProducer(conf)

Error:

Oauth callback fn
%6|1767787382.401|BGQUEUE|rdkafka#producer-1| [thrd:background]: Purging 1 unserved events from background queue
Traceback (most recent call last):
  File "/Users/ojain/sandbox/issue-2164/oauth_producer.py", line 23, in <module>
    p = SerializingProducer(conf)
  File "/Users/ojain/github.com/confluentinc/confluent-kafka-python/.venv/lib/python3.10/site-packages/confluent_kafka/serializing_producer.py", line 77, in __init__
    super(SerializingProducer, self).__init__(conf_copy)
cimpl.KafkaException: KafkaError{code=SASL_AUTHENTICATION_FAILED,val=58,str="OAuth token not set within 10 seconds timeout"}

Process finished with exit code 1

Will need to dig deeper into this. Which Python and Kafka version are you using?

[IgorMilavec]

I see. I'm using Python 3.13 and Python 3.14, but your test code correctly times out in my env.
Some additional context I did not consider relevant before: the code I posted initially is running inside a custom logging.Handler that is loaded by uvicorn:

python -u -m uvicorn src.main:app --log-config logging.json

I'll try to extract the code into a repro.

[ojasvajain]

Hi Igor Milavec (@IgorMilavec), were you able to reproduce it?