Python对接Heroku Kafka 3.8创建生产者触发self.async报错的问题咨询

• 依赖安装存在版本冲突：requirements.txt中同时安装了已停止维护的kafka==1.3.5和通用Kafka客户端kafka-python==2.0.2
• 废弃的kafka==1.3.5包未适配Python3.8版本，代码中使用async作为类属性名，而async是Python3.7+的内置保留关键字，直接触发语法解析错误
• 导入时kafka模块路径被废弃包覆盖，from kafka import KafkaProducer实际加载的是老包代码，而非预期的kafka-python模块

1. 调整依赖清单 #

修改requirements.txt，移除以下冲突、冗余依赖：

• kafka==1.3.5：语法错误的直接来源，无维护价值
• kafka-helper==0.1：第三方无关包，与自定义的kafka_helper.py模块重名，易引发导入路径冲突
• enum34==1.1.10、ipaddress==1.0.23：Python3.8标准库已内置对应能力，属于冗余backport依赖，会增加冲突概率

保留的Kafka相关依赖仅需：

kafka-python==2.0.2
cryptography==3.2.1
six==1.15.0

其余业务依赖（Flask、psycopg2、snowflake-connector-python等）可按原有版本保留，不影响本次连接问题。

2. 修正业务脚本错误 #

原kafka-producer.py存在两处问题，直接修正：

• 删除V_SSL_CONTEXT = kafka_helper.get_kafka_ssl_context().末尾多余的句点
• 移除对未定义变量V_KAFKA_URL、KAFKA_TOPIC的直接引用，统一调用自定义helper中的方法解析参数
• 补充KafkaProducer连接Heroku Kafka必须的security_protocol='SSL'配置项

修正后的参考代码：

import snowflake.connector
import os
import json
import ssl
import kafka_helper
from kafka import KafkaProducer

# 初始化SSL连接上下文
V_SSL_CONTEXT = kafka_helper.get_kafka_ssl_context()
# 从环境变量解析Kafka broker地址
V_KAFKA_URL = kafka_helper.get_kafka_brokers()
# 从Heroku配置变量获取Topic名
KAFKA_TOPIC = os.environ['KAFKA_TOPIC']

def fn_kafka_producer(acks='all',
                      value_serializer=lambda v: json.dumps(v).encode('utf-8')):
    kafkaprod = KafkaProducer(
        bootstrap_servers=V_KAFKA_URL,
        value_serializer=value_serializer,
        ssl_context=V_SSL_CONTEXT,
        security_protocol='SSL',
        acks=acks
    )
    return kafkaprod

if __name__ == '__main__':
    PRODUCER = fn_kafka_producer()
    PRODUCER.send(KAFKA_TOPIC, 'Hello Heroku!!')
    PRODUCER.flush()

注：自定义的kafka_helper.py代码逻辑完全适配Heroku Kafka的SSL连接规则，无需修改。

3. 重新部署应用 #

清空Heroku应用构建缓存后，重新推送代码部署，确保旧的废弃依赖不会被复用。

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