問題タブ [librdkafka]

消費者グループを実行すると、常に次のような統計が得られます。

ご覧のとおり、すべてのオフセットが -1001 に設定されています。

これは私のセットアップです（default_topicも）：

また、トピックに多くのメッセージがあっても、消費者グループがそれらを取得できない場合があります。何か案が？

Kafka クラスターへのメッセージを生成するために rdkafka_simple_producer.c を実行しています。1 つのトピックと 30 のパーティションがあります。デフォルトのラウンドロビン パーティショナーを使用します。プロデューサーが動作して Kafka へのメッセージを生成している間に、さらにパーティションを Kafka に追加します。

プロデューサーがこの変更に気付き、最終的には 40 のトピックすべてに対してプロデュースを開始することを期待しています。しかし、最終的には、元の 30 パーティションにデータが生成されたことしかわかりません。

テストでは、プロデューサーは 2 分間実行されました。

simple_producer に関数呼び出しを追加する必要がありますか、それとも考慮すべき Kafka パラメーターですか?

前もって感謝します！

Confluent dotnet clientを使ってアプリ起動時にトピックを処理したい。次の例を想定します。

Kafka に新しいメッセージがない場合、c.Consume はブロックされます。アプリケーションの起動 (キャッシュのウォームアップなど) に使用したいので、新しいメッセージがないことがわかったときにコードを続行したいと考えています。

タイムアウトを設定するための過負荷があることは知っていc.Consume(timeout)ますが、このアプローチの問題は、トピックにメッセージがあり、メッセージを読む時間がタイムアウトを超えた場合、望ましくない null 出力を受け取ることです。

コンシューマーが 5 分間 (max.poll.interval.ms のデフォルト値は 300000ms) メッセージを受信しない場合、コンシューマーはプログラムを終了せずに停止します。コンシューマ プロセスがハングし、それ以上メッセージを消費しません。

次のエラー メッセージがログに記録されます

ここで confluent-kafka-go でErrMaxPollExceeded定義されていることがわかりますが、それが発生している場所を見つけることができません。

このようなエラーが発生した場合、プログラムが終了しないのはなぜですか?

• カフカ v1.1.0
• librdkafka v1.0.0
• confluent-kafka-go (マスター)

kafka.Consumer に使用される構成

Apache Kafka の学習を始めたばかりです。nodejs と node-rdkafka を使用しています。Kafka の Exactly-Once 機能を使用するために Java コードしか表示されませんが、nodeJs で実装する方法が見つかりません。これを行うことができますか? はいの場合、その方法は?

librdkafka を使用した GoLang Kafka プロデューサーについていくつか質問がありました。これらは、debug: all を設定したときにプロデューサー ログに表示されるログに基づいています。プロデューサーは、バッチしきい値が再キャッシュされるか、linger.ms が渡されると、メッセージ セットの構築にある程度の時間を費やします。ただし、ほとんどの場合、メッセージはパーティション キューから xmit キューに移動されます。私はそれに関するドキュメントを取得しようとしていましたが、あまり見つけることができなかったので、スタックに関するヘルプが得られるかどうかを確認したいと思いました. 私の質問は次のとおりです-

a) アプリケーションは、パーティション固有のキューへの書き込み呼び出しを生成しますか?

b) パーティションごとに 1 つの xmit キューと 1 つのパーティション キューがありますか?

c) パーティション キューから xmit キューへの転送のトリガーは何ですか? また、なぜ 2 つのキューが必要なのですか?

d) Kafka プロデューサーがパーティションのメッセージセットを作成しているとき - パーティションのすべての操作をブロックしますか? (メッセージをパーティション キューから xmit キューに移動するようなもの)? 要するに、メッセージ セットがパーティション用に作成されているときに、新しいメッセージが xmit キューに忍び込む可能性はありますか? ブロックされていますか？

e) メッセージセットを作成するためにいくつのスレッドが機能しますか? プロデューサーごとに 1 つですか、それともパーティションごとに 1 つですか?

アプリケーションの 1 つにnode-rdkafkaライブラリを使用しています。そのため、メッセージを生成する際に、 tenantIdをヘッダーと共に渡すという 1 つの要件があります。ドキュメントを確認しましたが、必要なものが得られなかったか、何か不足している可能性があります。node-rdkafka ライブラリでメッセージを生成する際に、ヘッダー付きのテナント ID を渡す方法はありますか?

アプリケーションの 1 つにnode-rdkafkaを使用しています。グループ ID (グループ IDとトピックのペア) を持つ複数のトピックからのメッセージを消費したいのですが、node-rdkafka のドキュメントに関する詳細を見つけることができません。私の質問は、複数のグループ ID とトピックのペアを渡して、単一のカフカ コンシューマー接続から同時に多くのトピックからのメッセージを消費することは可能ですか?

ありがとう。

security.protocolconf プロパティを持つプロデューサーをプレーンテキストに接続しようとしています。username呼び出す前にオブジェクトに渡す方法passwordは?ConfRdKafka::Producer::create

setコールのどれも合わないようです。インスピレーションを得るためにsaslに関するドキュメントを読みましたが、役に立ちませんでした。

私のdr_cbはこのようなものです。送信に失敗したメッセージの内容を取得する方法はわかっていますが、トピックを取得するにはどうすればよいですか? これを行う最もエレガントな方法は何ですか?