kafka python - Bluemix MessageHub - ConnectionError：套接字已断开连接

Question

我正在使用 kafka python 客户端将消息推送到 Message Hub，但注意到在运行我的应用程序一段时间后它会停止向 Message Hub 发送消息。

然后我在日志文件中注意到以下内容：

ConnectionError: socket disconnected

我更新了我的代码以添加retries=5：

from kafka import KafkaProducer
from kafka.errors import KafkaError
import ssl

sasl_mechanism = 'PLAIN'
security_protocol = 'SASL_SSL'

# Create a new context using system defaults, disable all but TLS1.2
context = ssl.create_default_context()
context.options &= ssl.OP_NO_TLSv1
context.options &= ssl.OP_NO_TLSv1_1

producer = KafkaProducer(bootstrap_servers = app.config['KAFKA_BROKERS_SASL'],
                         sasl_plain_username = app.config['KAFKA_USERNAME'],
                         sasl_plain_password = app.config['KAFKA_PASSWORD'],
                         security_protocol = security_protocol,
                         ssl_context = context,
                         sasl_mechanism = sasl_mechanism,
                         api_version = (0,10),
                         retries=5)

def send_message(message):

    try:
        producer.send(app.config['KAFKA_TOPIC'], message.encode('utf-8'))

        # FIXME sending seems to be unreliable unless we flush
        producer.flush()
    except:
        print("Unexpected error:", sys.exc_info()[0])
        raise

我的代码从 python 烧瓶应用程序运行。每个进来的特定类型的请求都会调用send_message() 方法。

这是来自 Bluemix 的相关日志行。我可能错过了一两行复制和粘贴，但希望这足以弄清楚发生了什么：

APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxx May 15, 2017 8:50:54 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: socket disconnected May 15, 2017 8:50:54 PM
APP/1 Got error produce response on topic-partition TopicPartition(topic='movie_ratings', partition=0), retrying (4 attempts left). Error: ConnectionError: socket disconnectedMay 15, 2017 8:50:54 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:54 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxxMay 15, 2017 8:50:54 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: socket disconnected May 15, 2017 8:50:54 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:54 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxx May 15, 2017 8:50:54 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: socket disconnected May 15, 2017 8:50:55 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:55 PM
APP/1 Got error produce response on topic-partition TopicPartition(topic='movie_ratings', partition=0), retrying (2 attempts left). Error: ConnectionError: socket disconnected May 15, 2017 8:50:55 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxx May 15, 2017 8:50:55 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: socket disconnected May 15, 2017 8:50:55 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:55 PM
APP/1 Got error produce response on topic-partition TopicPartition(topic='movie_ratings', partition=0), retrying (1 attempts left). Error: ConnectionError: socket disconnected May 15, 2017 8:50:55 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxx May 15, 2017 8:50:55 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: socket disconnected May 15, 2017 8:50:55 PM
APP/1 Got error produce response on topic-partition TopicPartition(topic='movie_ratings', partition=0), retrying (0 attempts left). Error: ConnectionError: socket disconnected May 15, 2017 8:50:55 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:55 PM
APP/1 <BrokerConnection host=kafka01-prod01.messagehub.services.us-south.bluemix.net/23.246.202.51 port=9093>: Authenticated as xxxxx May 15, 2017 8:50:55 PM
APP/1 Node 0 connection failed -- refreshing metadata May 15, 2017 8:50:56 PM
APP/1 ConnectionError: socket disconnected May 15, 2017 8:50:56 PM
APP/1 Unable to import 'sasl'. Fallback to 'puresasl'. May 15, 2017 8:51:00 PM
APP/1 Closing active operation May 15, 2017 8:51:01 PM

我的主题存在。我的完整客户端代码在这里：https://github.com/snowch/movie-recommender-demo/blob/effc981cc9f799c41952719619f693172eebcd6a/web_app/app/messagehub_client.py

任何最受赞赏的指针...

Answer 1

根据Dominic's 的评论，将 kafka python 升级到 1.3.3 为我解决了这个问题。