問題タブ [kafka-python]

メッセージ ハブにメッセージをプッシュするために kafka python クライアントを使用していますが、しばらくアプリを実行すると、メッセージ ハブへのメッセージの送信が停止することに気付きました。

その後、ログ ファイルで次のことに気付きました。

コードを更新して追加しましたretries=5：

私のコードは python フラスコ アプリから実行されます。着信する特定のタイプのすべてのリクエストがsend_message()メソッドを呼び出します。

以下は、Bluemix からの関連するログ行です。1 行か 2 行のコピーと貼り付けを見逃している可能性がありますが、何が起こっているのかを理解するのに十分であることを願っています。

私のトピックは存在します。私の完全なクライアント コードはこちら: https://github.com/snowch/movie-recommender-demo/blob/effc981cc9f799c41952719619f693172eebcd6a/web_app/app/messagehub_client.py

最も高く評価されているポインタ...

私は Kafka を初めて使用し、その上でメッセージング プラットフォームにサービスを提供するサービスを構築しようとしています。これが私のセットアップです：

Kafka 0.9.0.1
Zookeeper 3.4.8 kafka
-python 1.3.3

私のアプリケーションは、KafkaProducerメッセージのストリームを 6 つのパーティションを持つ単一のトピックに送信する を作成します。また、7 つの s を作成しますKafkaConsumer(単一の の下で、group_idそのうちの 6 つが 6 つのパーティションに割り当てられ、1 つがアイドル状態のままになります (これは予想されることです)。プロデューサーがストリーミングしている間、期待して、パーティション数を 7 に増やします)。ストリームは 7 つのパーティションに分散されず、アイドル状態のコンシューマーを起動します.ただし、アプリケーションを再起動して再初期化するまで、プロデューサーは新しく追加されたパーティションを取得しないようです.パーティション数をスケーリングしますこれを実行して：

kafka-topics --alter --zookeeper localhost:2181 --topic test --partitions 7

プロデューサーがパーティション数の変更を再初期化せずに取得する方法はありますか?

関連するコード スニペットは次のとおりです。

プロデューサー

消費者

を使用してsslでkafkaに接続しようとするとkafka-python

次のエラーが表示されます。

I have python v2.7.3running on (Debian 7 wheezy)
ssl.SSLContex属性が追加されましたpython v2.7.9

sslサポートはkafka-pythonv1.1.0から導入されたようですが、使用上、pythonバージョン>= v2.7.9が必要ssl.SSLContextです。

下位バージョンの python を使用して ssl との接続を開始する方法はありkafka-pythonますか?
またはどういうわけかこれをバイパスしますか？

おそらく誰かが、kafka への ssl 接続をサポートし、python バージョン =< 2.7.3 をサポートする別のパッケージを提案できます。