AdminClient not working with OAUTHBEARER (MSK IAM) - KafkaError: Local: Broker transport failure

[jittakal]

Description:
I'm trying to use AdminClient with AWS MSK IAM authentication (OAUTHBEARER) in confluent-kafka-python, but I'm getting a transport failure error when calling list_topics. The same configuration works with the regular Kafka consumer/producer, but not with AdminClient.

Code Sample:

from confluent_kafka.admin import AdminClient
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider

def oauth_cb(oauth_config):
    auth_token, expiry_ms = MSKAuthTokenProvider.generate_auth_token("us-east-1")
    return auth_token, expiry_ms/1000

admin = AdminClient({
    "bootstrap.servers": "b-1.cluster.abcd.das.kafka.us-east-1.amazonaws.com:9098,b-2.adsd.sdas.asdsad.kafka.us-east-1.amazonaws.com:9098",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "OAUTHBEARER",
    "oauth_cb": oauth_cb
})

def main():
    print("Hello from confluent-kafka-admin!")
    topics = admin.list_topics(timeout=10)
    for topic in topics.topics.values():
        print(f"Topic: {topic.topic}, Partitions: {len(topic.partitions)}")

if __name__ == "__main__":
    main()

Error:

Traceback (most recent call last):
  File "main.py", line 25, in <module>
    main()
  File "main.py", line 19, in main
    topics = admin.list_topics(timeout=10)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "...confluent_kafka\\admin\\__init__.py", line 639, in list_topics
    return super(AdminClient, self).list_topics(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"}

Question:
Is AdminClient expected to work with OAUTHBEARER/MSK IAM? Are there any additional configuration steps required? Why does this error occur only with AdminClient?