使用Confluent-Kafka+OAuth连接Kafka Broker遇超时及SSL握手错误

我在使用Confluent-Kafka结合OAuth认证连接Kafka Broker时失败，日志中显示SSL握手错误，但回调函数返回的错误信息却是"KafkaError{code=_MSG_TIMED_OUT,val=-192,str="Local: Message timed out"}"。以下是我的实现代码：

from confluent_kafka import Producer
import time

bootstrap_servers="IP Address"
topic = "topic name"

oauth_access_token = 'access token'
def delivery_callback(err, msg):
    if err:
        print('ERROR: Message failed delivery: {}'.format(err), msg)
    else:
        print("message produced in topic", msg)

def kafka_producer_with_oauth():
    conf = {
        'bootstrap.servers': bootstrap_servers,
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'OAUTHBEARER',
        'sasl.oauthbearer.config': f'oauthbearer_token={oauth_access_token}',
        'message.timeout.ms': 10000,
    }

    p = Producer(conf)
    print(p)
    for i in range(2):
        message = f"Message {i}"
        p.produce(topic, message.encode('utf-8'), callback=delivery_callback)
        print("message produced.. ", i)

    p.poll(0)
    p.flush()
kafka_producer_with_oauth()

排查方向 #

• SSL证书信任问题：SASL_SSL模式下默认会验证Broker的SSL证书，若未配置信任根证书，会直接导致SSL握手失败，最终表现为消息超时（握手未完成，发送操作等待至超时）。需添加ssl.ca.location参数指定CA证书路径。
• OAuth令牌有效性：确认oauth_access_token是否过期、是否拥有目标Topic的写入权限。令牌无效或权限不足会导致SASL认证失败，进而引发后续操作超时。
• 网络连通性：检查Producer所在机器能否访问Broker的SSL端口（默认9093），可通过nc -zv <Broker-IP> 9093测试，网络不通会直接导致握手超时。
• 超时参数设置：当前message.timeout.ms设为10秒，若网络延迟高或Broker负载大，需适当调大该值。

代码优化建议 #

修改配置添加SSL证书配置、开启调试日志，并调整超时相关逻辑：

from confluent_kafka import Producer
import time

bootstrap_servers="IP Address"
topic = "topic name"

oauth_access_token = 'access token'
def delivery_callback(err, msg):
    if err:
        print('ERROR: Message failed delivery: {}'.format(err), msg)
    else:
        print("message produced in topic", msg.topic())

def kafka_producer_with_oauth():
    conf = {
        'bootstrap.servers': bootstrap_servers,
        'security.protocol': 'SASL_SSL',
        'sasl.mechanism': 'OAUTHBEARER',
        'sasl.oauthbearer.config': f'oauthbearer_token={oauth_access_token}',
        'message.timeout.ms': 30000,  # 调大超时时间至30秒
        'ssl.ca.location': '/path/to/your/ca-cert.pem',  # 指定CA证书路径
        'debug': 'security,ssl'  # 开启安全/SSL调试日志，便于定位问题
    }

    p = Producer(conf)
    print(p)
    for i in range(2):
        message = f"Message {i}"
        try:
            p.produce(topic, message.encode('utf-8'), callback=delivery_callback)
            print("message queued for delivery.. ", i)
            p.poll(1)  # 每次produce后调用poll处理回调，避免事件堆积
        except Exception as e:
            print(f"Failed to produce message {i}: {e}")

    p.flush(30)  # flush时指定超时时间，确保所有消息处理完成
kafka_producer_with_oauth()

修改说明 #

1. 添加ssl.ca.location解决SSL握手的证书信任问题；
2. 开启debug参数，输出详细的安全及SSL调试日志，可直接看到握手失败的具体原因；
3. 调大message.timeout.ms为30秒，给网络和认证流程留足时间；
4. 在produce循环中增加p.poll(1)，及时处理内部事件和回调；
5. flush时指定超时时间，避免无限等待。

内容的提问来源于stack exchange，提问作者senthil rajendran