Confluent Kafka SchemaRegistryClient 如何禁用SSL证书验证

问题根因 #

你配置的enable.ssl.certificate.verification: false仅作用于Kafka Broker的连接校验，不会影响Schema Registry客户端发起的HTTP请求，因此反序列化阶段根据Schema ID拉取 schema 时会触发SSL证书校验失败。

解决方案 #

方案1：直接配置Schema Registry客户端关闭证书校验（无自定义逻辑，优先使用） #

你初始化SchemaRegistryClient时仅传入了服务地址，补充SSL校验配置即可：

适用于confluent-kafka-python v1.5.0及以上版本

直接在客户端配置中添加ssl.verify参数：

schema_registry_client = SchemaRegistryClient({
    'url': registry_configuration,
    'ssl.verify': False  # 关闭Schema Registry请求的证书校验
})

适用于5.5.1等不支持ssl.verify参数的老版本

手动传入自定义的requests会话，全局关闭会话的证书校验：

import requests
import urllib3
# 禁用不安全请求警告，避免控制台输出冗余告警
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# 自定义会话，关闭证书校验
custom_session = requests.Session()
custom_session.verify = False

# 初始化SchemaRegistryClient时传入自定义会话
schema_registry_client = SchemaRegistryClient(
    {'url': registry_configuration},
    session=custom_session
)

修改完成后原有消费逻辑无需变动，即可正常运行。

方案2：自定义极简反序列化器，完全避免Schema Registry请求（备选方案） #

你已经提前获取到了目标Schema，可以直接写几行代码实现反序列化，完全不依赖SchemaRegistryClient：

import io
from avro.io import DatumReader, BinaryDecoder
from avro.schema import Parse

# 提前解析你已经获取到的schema
avro_schema = Parse(consumption_schema)
avro_reader = DatumReader(avro_schema)

# 自定义反序列化方法
def custom_avro_deserializer(data, ctx):
    if data is None:
        return None
    # 跳过Avro消息前5个字节的固定头（1位魔术位+4位Schema ID）
    bytes_io = io.BytesIO(data[5:])
    decoder = BinaryDecoder(bytes_io)
    return avro_reader.read(decoder)

然后替换消费者配置中的值反序列化器即可：

consumer_conf = {
    'key.deserializer': string_deserializer,
    'value.deserializer': custom_avro_deserializer,
    'group.id': 'myconsumergroupid',
    'auto.offset.reset': 'earliest'
}

内容的提问来源于stack exchange，提问作者Saugat Mukherjee