搭建MQTT与Kafka桥接时遇pykafka SocketDisconnectedError求助

相关环境说明 #

• 树莓派4
• 运行Ubuntu 20.04（ARM64架构）
• 通过ZeroTier+SSH远程连接
• 已运行nodered、mosquitto、portainer、zookeeper、Kafka容器

Docker-Compose配置 #

zookeeper:
  image: ubuntu/zookeeper
  container_name: zookeeper
  ports:
    - "2181:2181"
kafka:
  image: ubuntu/kafka
  container_name: kafka
  ports:
    - "9092:9092"
  environment:
    - KAFKA_ADVERTISED_HOST_NAME=kafka
    - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
    - KAFKA_CREATE_TOPICS= "mqtt-sensor-1:1:1"
    - KAFKA_DELETE_TOPIC_ENABLE=true
  depends_on:
    - zookeeper
  restart: on-failure

现状与问题 #

MQTT发布订阅功能正常，可用的MQTT发布端代码：

import paho.mqtt.client as mqtt
from random import randrange, uniform
import time
import sys

client= mqtt.Client("GreenForges v1")

if client.connect("localhost",1883,60) !=0:
                print("could not connect to MQTT broker!")
                sys.exit(-1)
while True:
    randNumber = uniform(20.0, 21.0)
    client.publish("mqtt-sensor-1", randNumber)
    print("Just published " + str(randNumber) + " to Topic mqtt-sensor-1")
    time.sleep(10)

尝试用Python+pykafka编写MQTT-Kafka桥接代码时，创建kafka_producer环节持续失败，桥接代码：

import paho.mqtt.client as mqtt
from pykafka import KafkaClient
import time

mqtt_client = mqtt.Client("BridgeMQTT3Kafka")
mqtt_client.connect("localhost",1883,60)

kafka_client = KafkaClient(hosts='localhost:9092')
kafka_topic = kafka_client.topics['mqtt-sensor-1']
kafka_producer = kafka_topic.get_sync_producer()

def on_message(client, userdata, message):
    msg_payload = str(message.payload)
    print("Received MQTT message: ", msg_payload)
    kafka_producer.produce(str(msg_payload).encode('ascii'))
    print("KAFKA: Just published " + str(msg_payload) + " to topic mqtt-sensor-1")

mqtt_client.loop_start()
mqtt_client.subscribe("mqtt-sensor-1")
mqtt_client.on_message = on_message
time.sleep(300)
mqtt_client.loop_end()

报错信息 #

File "/home/green/.local/lib/python3.10/site-packages/pykafka/broker.py", line 44, in wrapped
    raise SocketDisconnectedError
pykafka.exceptions.SocketDisconnectedError

Kafka日志信息 #

Controller.log #

[2022-09-09 15:45:16,117] INFO [Controller id=0] New partition creation callback for mqtt-sensor-1-0,mqtt-sensor-1-1 (kafka.controller.KafkaController)
[2022-09-09 15:47:59,406] INFO [Controller id=0] Processing automatic preferred replica leader election (kafka.controller.KafkaController)
[2022-09-09 15:47:59,406] TRACE [Controller id=0] Checking need to trigger auto leader balancing (kafka.controller.KafkaController)
[2022-09-09 15:47:59,417] DEBUG [Controller id=0] Topics not in preferred replica for broker 0 Map() (kafka.controller.KafkaController)
[2022-09-09 15:47:59,423] TRACE [Controller id=0] Leader imbalance ratio for broker 0 is 0.0 (kafka.controller.KafkaController)

server.log #

[2022-09-09 15:45:16,001] INFO Creating topic mqtt-sensor-1 with configuration {} and initial partition assignment HashMap(0 -> ArrayBuffer(0), 1 -> ArrayBuffer(0)) (kafka.zk.AdminZkClient)
[2022-09-09 15:45:16,344] INFO [ReplicaFetcherManager on broker 0] Removed fetcher for partitions Set(mqtt-sensor-1-1, mqtt-sensor-1-0) (kafka.server.ReplicaFetcherManager)
[2022-09-09 15:45:16,596] INFO [LogLoader partition=mqtt-sensor-1-1, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.UnifiedLog$)
[2022-09-09 15:45:16,649] INFO Created log for partition mqtt-sensor-1-1 in /tmp/kafka-logs/mqtt-sensor-1-1 with properties {} (kafka.log.LogManager)
[2022-09-09 15:45:16,658] INFO [Partition mqtt-sensor-1-1 broker=0] No checkpointed highwatermark is found for partition mqtt-sensor-1-1 (kafka.cluster.Partition) 
[2022-09-09 15:45:16,662] INFO [Partition mqtt-sensor-1-1 broker=0] Log loaded for partition mqtt-sensor-1-1 with initial high watermark 0 (kafka.cluster.Partition)
[2022-09-09 15:45:16,722] INFO [LogLoader partition=mqtt-sensor-1-0, dir=/tmp/kafka-logs] Loading producer state till offset 0 with message format version 2 (kafka.log.UnifiedLog$)
[2022-09-09 15:45:16,731] INFO Created log for partition mqtt-sensor-1-0 in /tmp/kafka-logs/mqtt-sensor-1-0 with properties {} (kafka.log.LogManager)
[2022-09-09 15:45:16,732] INFO [Partition mqtt-sensor-1-0 broker=0] No checkpointed highwatermark is found for partition mqtt-sensor-1-0 (kafka.cluster.Partition) 
[2022-09-09 15:45:16,734] INFO [Partition mqtt-sensor-1-0 broker=0] Log loaded for partition mqtt-sensor-1-0 with initial high watermark 0 (kafka.cluster.Partition)

进一步排查结果 #

• 执行./bin/zookeeper-shell.sh localhost:2181 ls /brokers/ids返回连接拒绝：

Connecting to localhost:2181
[2022-09-09 16:18:36,326] WARN Session 0x0 for sever localhost/127.0.0.1:2181, Closing socket connection. Attempting reconnect except it is a SessionExpiredException. (org.apache.zookeeper.ClientCnxn)
java.net.ConnectException: Connection refused
        at java.base/sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
        at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:777)
        at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:344)
        at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1290)
KeeperErrorCode = ConnectionLoss for /brokers/ids
[2022-09-09 16:18:36,482] ERROR Exiting JVM with code 1 (org.apache.zookeeper.util.ServiceUtils)

• 执行./bin/zookeeper-shell.sh localhost:9092 ls /brokers/ids返回会话超时：

Connecting to localhost:9092
[2022-09-09 16:19:44,858] WARN Client session timed out, have not heard from server in 30006ms for session id 0x0 (org.apache.zookeeper.ClientCnxn)
[2022-09-09 16:19:44,867] WARN Session 0x0 for sever localhost/127.0.0.1:9092, Closing socket connection. Attempting reconnect except it is a SessionExpiredException. (org.apache.zookeeper.ClientCnxn)
org.apache.zookeeper.ClientCnxn$SessionTimeoutException: Client session timed out, have not heard from server in 30006ms for session id 0x0
        at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1258)
KeeperErrorCode = ConnectionLoss for /brokers/ids
[2022-09-09 16:19:45,018] ERROR Exiting JVM with code 1 (org.apache.zookeeper.util.ServiceUtils)

• Kafka命令行工具生产消费正常：

bin/kafka-console-consumer.sh --topic mqtt-sensor-1 --from-beginning --bootstrap-server localhost:9092
bin/kafka-console-producer.sh --topic mqtt-sensor-1 --bootstrap-server localhost:9092

1. 调整Kafka的广告地址配置 #

当前Docker-Compose中KAFKA_ADVERTISED_HOST_NAME=kafka是容器内部主机名，宿主机（运行Python代码的环境）无法解析该名称。pykafka会从Kafka broker获取其广告地址，尝试连接时会因无法解析导致失败。

修改Docker-Compose中Kafka的环境变量：

environment:
  - KAFKA_ADVERTISED_HOST_NAME=localhost  # 替换为树莓派本地IP或localhost，确保宿主机可访问
  - KAFKA_ADVERTISED_PORT=9092
  - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
  - KAFKA_CREATE_TOPICS="mqtt-sensor-1:1:1"
  - KAFKA_DELETE_TOPIC_ENABLE=true

修改后重启Kafka容器：docker-compose restart kafka

2. 强制pykafka使用指定的Broker地址 #

若不想修改Kafka配置，可在创建KafkaClient时指定broker_version参数，强制使用传入的hosts地址，避免从Broker获取广告地址：

kafka_client = KafkaClient(hosts='localhost:9092', broker_version='2.8.0')  # 替换为你的Kafka实际版本

注意：pykafka维护活跃度低，对新版本Kafka兼容性有限，需确保版本匹配。

3. 检查容器网络连通性 #

确认宿主机能正常访问Kafka容器的9092端口：

nc -zv localhost 9092

若不通，检查Docker端口映射及容器运行状态：

docker ps | grep kafka

4. 替换pykafka为confluent-kafka #

pykafka对新版本Kafka支持不足，改用维护更活跃的confluent-kafka库，示例代码：

import paho.mqtt.client as mqtt
from confluent_kafka import Producer
import time

# Kafka Producer配置
conf = {'bootstrap.servers': 'localhost:9092'}
kafka_producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

def on_message(client, userdata, message):
    msg_payload = message.payload.decode('utf-8')
    print("Received MQTT message: ", msg_payload)
    kafka_producer.produce('mqtt-sensor-1', value=msg_payload.encode('ascii'), callback=delivery_report)
    kafka_producer.poll(0)
    print("KAFKA: Just published " + msg_payload + " to topic mqtt-sensor-1")

# MQTT客户端配置
mqtt_client = mqtt.Client("BridgeMQTT3Kafka")
mqtt_client.connect("localhost", 1883, 60)
mqtt_client.subscribe("mqtt-sensor-1")
mqtt_client.on_message = on_message

mqtt_client.loop_start()
time.sleep(300)
mqtt_client.loop_end()

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