Connection Times out with confluent-kafka client.

[keithrozario]

• aws-msk-iam-sasl-signer-python version:aws-msk-iam-sasl-signer-python==1.0.1
• Python version:3.10.12
• Operating System:Linux
• Method of installation: Pip
• Kafka library name: confluent-kafka
• Kafka library versionc: confluent-kafka==2.5.3
• Provide us a sample code snippet of your producer/consumer

Description

The connection times out when I try to do any operation with confluent-kafka. Turning off SASL works (unauthenticated), or using the python-kafka client instead also works. It just doesn't work when I use confluent-kafka with SASL.

from confluent_kafka.admin import AdminClient, NewTopic
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
import socket

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

provisioned_iam_host = <iam endpoint>
tls_host = <tls endpoint>
a = AdminClient(
    {
    # 'bootstrap.servers': tls_host,
    # 'security.protocol': 'SSL',
    'bootstrap.servers': provisioned_iam_host,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'OAUTHBEARER',
    'oauth_cb': oauth_cb,
    "debug": "all",
    'client.id': 'test'
    }
)

new_topics = [NewTopic('topic3', num_partitions=3, replication_factor=1)]

fs = a.create_topics(new_topics,request_timeout=3)

# Wait for each operation to finish.
for topic, f in fs.items():
    try:
        f.result()  # The result itself is None
        print("Topic {} created".format(topic))
    except Exception as e:
        print("Failed to create topic {}: {}".format(topic, e))

What I Did

The connection times out with the confluent-kafka client. The python-kafka client works from the same machine (not an issue with the network).

confluent-kafka will work with no authentication as well.

But when I turn on SASL_SSL (and connect to the right endpoint) it will time out, and be unable to find new brokers.

[keithrozario]

Broker version is 3.5.1, I tested with both MSK serverless and Provisioned.