問題タブ [kafka-producer-api]

[2016-02-17 09:02:36,876] エラー [0,32] の相関 ID を持つトピック ツイートのリクエストを送信できませんでした

この問題を解決する方法

前もって感謝します

以下の 2 つの依存関係の違いは何ですか? コンシューマー アプリまたはプロデューサー アプリを作成するために最初のアプリが本当に必要ですか?

私の Producer は最初のものだけで正常に動作しますが、消費者には 2 番目のものが必要です。

「kafka-clients」アーティファクトはプロデューサーとコンシューマーの両方で機能すると考えていました。しかし、「kafka.consumer.Consumer」は他の依存関係から来ているようです。なぜ違いがあるのですか？

また、最初のアーティファクトが kafka_2.9.2 と名付けられているのはなぜですか? つまり、名前にバージョン識別子が含まれているのはなぜですか?

私はKafkaを使い始めたばかりで、コマンドプロンプトからデータを生成および消費でき、リモートサーバーからでもJavaコードを介してデータを生成できます。

しかし、私はこの単純なコンシューマ Java コードを試していますが、機能していません。

出力は次のとおりです:-

エラーはありません。プログラムは終了せず、出力もありません!!

飼育係の記録

無限ループで Kafka コンソールでこれを取得しています。説明してください

私は次の方法でトピックを作成しました

コマンドプロンプトで使用できます

何が問題なのか理解できません。

私は趣味のプロジェクトとして Kafka 9 を調査しており、「Hello World」タイプの例をいくつか完成させました。

一般的に要求応答メッセージングに基づいて実世界の Kafka アプリケーションについて考えなければなりません。より具体的には、Kafka 要求メッセージをその応答メッセージにリンクする方法を考えなければなりません。

生成された UUID を要求メッセージ キーとして使用し、この要求 UUID を関連する応答メッセージ キーとして使用する方法を考えていました。WebSphere MQ がメッセージ相関 ID を持っているのとほぼ同じタイプのメカニズムです。

私の最後の2つの終了プロセスは次のようになります。

1)。Kafka クライアントはランダムな UUID を生成し、単一の Kafka 要求メッセージを送信します。2)。サーバーは、この要求メッセージの抽出を消費し、要求の UUID 値を保存します 3)。メッセージ ペイロードを使用してビジネス プロセスを完了します。4)。要求メッセージから格納された UUID 値を応答メッセージ キーとして使用する応答メッセージで応答します。5)。Kafka クライアントは、タイムアウトするか、元の要求 UUID 値を持つメッセージを取得するまで、応答トピックをポーリングします。

私が懸念しているのは、Kafka コンシューマーのポーリングが他のクライアントのメッセージを応答トピックから削除し、オフセットを増やして他のクライアントを失敗させることです。

設計されていないユースケースで Kafka を適用しようとしていますか?

Kafka で要求/応答メッセージングを実装することは可能ですか?

3 つのパーティションと 3 つのレプリケーションで Kafka トピックを作成しました

$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test

APIからデータを取得するためのKafka-Producerプログラムを作成し、正常に動作しました。私のデータは 3 つのパーティションと 3 つのレプリケーションに正常に保存されています。

ここで、パーティション化された Kafka ログにあるデータを処理する必要があります。ストーム トポロジがあります。トポロジがログからのデータを入力として受け入れる方法を教えてください。

最初に、1 つのパーティション、1 つのレプリケーションでトピックを作成しましたが、ストーム トポロジは正常に動作します。それを解決する方法。

私のトポロジコード

この問題について助けてください

前もって感謝します

KeyedMessage data = new KeyedMessage("topic",partition_key, msg); を使用してメッセージを公開しました。

メッセージがどのパーティションに送信されたかを再確認する方法は?