【Azure事件中心】使用Python SDK(Confluent)相关方法获取offset或lag时提示SSL相关错误

简介: 【Azure事件中心】使用Python SDK(Confluent)相关方法获取offset或lag时提示SSL相关错误

问题描述

使用Python SDK(Confluent)相关方法获取offset或lag时, 提示SSL相关错误, 是否有更清晰的实例以便参考呢?

 

问题解决

执行代码,因为一直连接不成功,所以检查 confluent_kafka 的连接配置,最后定位是 sasl.password 值设置有误。此处,需要使用Event Hub Namespace级别的连接字符串(Connection String).

 

在Event Hub中,获取方式为: (1: Shared access policies ---> 2: RootManageSharedAccessKey or ..----> 3: Connection String )

 

完整的示例代码:

import confluent_kafka
topics = ["<Your_topic_name>"]
broker = "<Eventhub-namespace-name>.servicebus.chinacloudapi.cn:9093"
group_name = "<Consumer-group-name>"
sasl_password = "<Connection-string>"
# Create consumer.
# This consumer will not join the group, but the group.id is required by
# committed() to know which group to get offsets for.
consumer = confluent_kafka.Consumer({'bootstrap.servers': broker,
                                     'security.protocol': 'SASL_SSL',
                                     'sasl.mechanism': 'PLAIN',
                                     'sasl.username': '$ConnectionString',
                                     'sasl.password': sasl_password,
                                     'group.id': group_name})
print("%-50s  %9s  %9s" % ("Topic [Partition]", "Committed", "Lag"))
print("=" * 72)
for topic in topics:
    # Get the topic's partitions
    metadata = consumer.list_topics(topic, timeout=10)
    if metadata.topics[topic].error is not None:
        raise confluent_kafka.KafkaException(metadata.topics[topic].error)
    # Construct TopicPartition list of partitions to query
    partitions = [confluent_kafka.TopicPartition(topic, p) for p in metadata.topics[topic].partitions]
    # Query committed offsets for this group and the given partitions
    committed = consumer.committed(partitions, timeout=10)
    for partition in committed:
        # Get the partitions low and high watermark offsets.
        (lo, hi) = consumer.get_watermark_offsets(partition, timeout=10, cached=False)
        if partition.offset == confluent_kafka.OFFSET_INVALID:
            offset = "-"
        else:
            offset = "%d" % (partition.offset)
        if hi < 0:
            lag = "no hwmark"  # Unlikely
        elif partition.offset < 0:
            # No committed offset, show total message count as lag.
            # The actual message count may be lower due to compaction
            # and record deletions.
            lag = "%d" % (hi - lo)
        else:
            lag = "%d" % (hi - partition.offset)
        print("%-50s  %9s  %9s" % (
            "{} [{}]".format(partition.topic, partition.partition), offset, lag))
consumer.close()

 

参考文档


confluent-kafka-python : https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/list_offsets.py

相关文章
|
2月前
|
存储 人工智能 开发工具
AI助理化繁为简,速取代码参数——使用python SDK 处理OSS存储的图片
只需要通过向AI助理提问的方式输入您的需求,即可瞬间获得核心流程代码及参数,缩短学习路径、提升开发效率。
1448 4
AI助理化繁为简,速取代码参数——使用python SDK 处理OSS存储的图片
|
5天前
|
弹性计算 安全 开发工具
灵码评测-阿里云提供的ECS python3 sdk做安全组管理
批量变更阿里云ECS安全组策略(批量变更)
|
15天前
|
API Python
【Azure Developer】分享一段Python代码调用Graph API创建用户的示例
分享一段Python代码调用Graph API创建用户的示例
38 11
|
1月前
|
Java 开发工具 Windows
【Azure App Service】在App Service中调用Stroage SDK上传文件时遇见 System.OutOfMemoryException
System.OutOfMemoryException: Exception of type 'System.OutOfMemoryException' was thrown.
|
27天前
|
中间件 Docker Python
【Azure Function】FTP上传了Python Function文件后,无法在门户页面加载函数的问题
通过FTP上传Python Function至Azure云后,出现函数列表无法加载的问题。经排查,发现是由于`requirements.txt`中的依赖包未被正确安装。解决方法为:在本地安装依赖包到`.python_packages/lib/site-packages`目录,再将该目录内容上传至云上的`wwwroot`目录,并重启应用。最终成功加载函数列表。
|
2月前
|
机器人 Shell Linux
【Azure Bot Service】部署Python ChatBot代码到App Service中
本文介绍了使用Python编写的ChatBot在部署到Azure App Service时遇到的问题及解决方案。主要问题是应用启动失败,错误信息为“Failed to find attribute &#39;app&#39; in &#39;app&#39;”。解决步骤包括:1) 修改`app.py`文件,添加`init_func`函数;2) 配置`config.py`,添加与Azure Bot Service认证相关的配置项;3) 设置App Service的启动命令为`python3 -m aiohttp.web -H 0.0.0.0 -P 8000 app:init_func`。
|
2月前
|
Linux Python
【Azure Function】Python Function部署到Azure后报错No module named '_cffi_backend'
ERROR: Error: No module named '_cffi_backend', Cannot find module. Please check the requirements.txt file for the missing module.
|
2月前
|
JavaScript 前端开发 开发工具
【Azure Developer】使用JavaScript通过SDK进行monitor-query的client认证报错问题
AADSTS90002: Tenant 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx' not found. Check to make sure you have the correct tenant ID and are signing into the correct cloud. Check with your subscription administrator, this may happen if there are no active subscriptions for the tenant.
|
2月前
|
安全 网络安全 数据安全/隐私保护
【Azure Developer】System.Net.WebException: The request was aborted: Could not create SSL/TLS secure channel.
System.Net.WebException: The request was aborted: Could not create SSL/TLS secure channel.
|
3月前
|
Kubernetes API 开发工具
【Azure Developer】通过SDK(for python)获取Azure服务生命周期信息
需要通过Python SDK获取Azure服务的一些通知信息,如:K8S版本需要更新到指定的版本,Azure服务的维护通知,服务处于不健康状态时的通知,及相关的操作建议等内容。
51 18