python - 在 SchemaRegistryClient Confluent Kafka 中禁用证书验证
问题描述
所以,我想从 kafka (Confluent) 中阅读一个主题,其中数据位于 Avro 格式。
由于某些不可避免的原因,我想禁用证书验证。
我正在使用 security.protocol= SASL_SSL 和 sasl.mechanisms= OUTHBEARER
我可以通过禁用 ssl 证书验证来连接到 Kafka
'enable.ssl.certificate.verification': 'false'
现在,我在尝试使用 Schema 注册表反序列化值时遇到了问题。avro 解串器需要模式注册表客户端和模式(可选)。我两个都过去了。我通过使用 verify=False 发出单独的请求来获取第二个参数的值以获取架构,这很好。但是当我尝试从中创建一个反序列化消费者时,问题就出现了。
基本上是代码的骨架(以及问题所在的注释)
topic="mytopic"
registry_configuration="schema registry url"
url = urljoin(registry_configuration, f'/subjects/{topic}-value/versions/latest')
schema_registry_response = requests.get(url, verify=False) #here I intentionally used verify=False to get the schema
schema_registry_response.raise_for_status()
consumption_schema=schema_registry_response.json()['schema']
print(consumption_schema) # This works fine
schema_registry_client = SchemaRegistryClient({'url': registry_configuration})
avro_deserializer = AvroDeserializer(schema_registry_client,consumption_schema
) # This is the problem area which is called when the poll method is called
string_deserializer = StringDeserializer('utf_8')
basic_conf=_get_basic_configuration()
consumer_conf = {'key.deserializer': string_deserializer,
'value.deserializer': avro_deserializer,
'group.id': 'myconsumergroupid',
'auto.offset.reset': 'earliest'}
consumer_conf.update(basic_conf)
# update default config with parameter
additional_consumer_conf={}
consumer_conf.update(additional_consumer_conf)
cn=DeserializingConsumer(consumer_conf)
cn.subscribe(['topicname'])
while True:
msg=cn.poll(10) # This fails because the deserializingconsumer calls the schema registry and the certificate validation fails
您在上面看到的 get_basic_configuration 方法有
'enable.ssl.certificate.verification': 'false'
错误是
Max retries exceeded with url: /schemas/ids/140 (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate in certificate chain
我在这里浏览了 SchemaRegistryClient 的代码, 但我没有看到任何将证书验证设置为 false 的选项。
我还搜索了 SO 帖子和其他 Confluent 文档,看看是否能找到一些东西,但没有任何帮助
希望有人在这里了解更多?我愿意详细说明或澄清任何问题吗?如果可能的话,我试图不使用很多自定义逻辑来反序列化。
解决方案
我已经找到了答案。
基本上这个 SO 帖子在这里。特别是接受答案之后的答案,如果您使用的是 Confluent kafka
和 avro 文档,因为我的架构不是来自文件,而是作为 http 响应,所以我必须使用 avro.schema.parse 解析它
最终骨架代码
topic="mytopic"
registry_configuration="schema registry url"
url = urljoin(registry_configuration, f'/subjects/{topic}-value/versions/latest')
schema_registry_response = requests.get(url, verify=False)
schema_registry_response.raise_for_status()
consumption_schema=schema_registry_response.json()['schema']
consumption_schema = avro.schema.parse(consumption_schema)
schema_registry_client = SchemaRegistryClient({'url': registry_configuration})
basic_conf=_get_basic_configuration()
consumer_conf = {
'group.id': 'myconsumergroupid',
'auto.offset.reset': 'earliest'}
consumer_conf.update(basic_conf)
cn=Consumer(consumer_conf)
cn.subscribe(['mytopic'])
reader = DatumReader(consumption_schema)
while True:
msg=cn.poll(10)
if msg is None:
break
m=msg.value()
message_bytes = io.BytesIO(m)
message_bytes.seek(5)
decoder = BinaryDecoder(message_bytes)
event_dict = reader.read(decoder)
print(event_dict)
推荐阅读
- sql - SQL Server:如何对多个版本进行排序和比较?
- react-native - 如何在 React Native 中使用 Redux Thunk 保存我的 JWT Token Globaly?
- mysql - 无法从 Windows Azure VM 连接 Linux Azure VM
- python - 生成10万个特征和类只有0和1且是线性的数据集
- css - 如何将我的文本从图像的右侧和左侧放置到垂直中心?
- django - 创建 Django“Sub Admin”以访问其公司用户的数据
- java - Spring集成邮件imap入站适配器在集成流程更改后抛出异常
- c++ - 链接:如何强制包含 .obj / .cpp 文件或为什么将其从构建中排除?
- python-2.7 - 如何在大型数据集上优化图形质量
- encryption - 在 Flutter 上解密 AES 输入,在 Web 上使用 cryptoJS AES