問題タブ [pykafka]

大きなファイルを Kafka に保存し、レコードに関するメタデータを使用して将来それらを取得したいと考えています。

そこで、トピック、partition_id、offset を含むメッセージを送信してから、次の方法でファイルを取得しようとします。

ただし、機能せず、次のように出力されます。

このエラーは非常にわかりにくく、reset_offsets で発生します。消費しようとすると、プロセスは rebalancing_lock を待ってスタックします。私は何を間違っていますか？

Python で kafka トピックを消費し、prometheus クライアントを使用して http 経由で提供しようとしていますが、トピックの消費がブロックされているようです。単純にメトリックを追加するためにいくつかのプレースホルダーを配置しましたが、その部分がブロックされているようです。

コードを実行すると、トピック データが期待どおりにコンソールにストリーミングされていることがわかります。ただし、メトリック エンドポイントにはデータが入力されず、Web サーバーへのリクエストはアプリを強制終了するまでハングします。アプリはライブラリからの標準メトリックで応答します。

複数のマシンで共通のコンシューマー グループを使用して、kafka キューを管理しています。ここで、キューの現在の内容も表示する必要があります。グループ内で未読のメッセージのみを読み取り、実際にそれらのメッセージを処理するグループ内の他のコンシューマーがそれらのメッセージを再度読み取れるようにするにはどうすればよいですか。どんな助けでも大歓迎です。