問題タブ [kafka-producer-api]

アプリケーションからテキスト メッセージを kafka に送信できませんが、オブジェクト (イベント) を送信しようとすると例外が発生します。

INFO: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: メッセージハンドラでエラーが発生しました [org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler#0]; ネストされた例外は java.lang.NullPointerException: in org.springframework.data.mongodb.core.mapping.event.AfterSaveEvent null of org.springframework.data.mongodb.core.mapping.event.AfterSaveEvent, headers={id=9068de1b- a1cc-c1b5-463a-107b7be21af4、タイムスタンプ = 1456754038408}] 1000

どんな助けでも大歓迎です...

私は、2 つのブローカーと 1 つの飼育係を持つカフカ環境を持っています。

Kafka にメッセージを生成しようとしているときに、ブローカー 1 (リーダーのブローカー) を停止すると、ブローカー 2 がトピックとパーティションの新しいリーダーとして選出されますが、クライアントはメッセージの生成を停止し、以下のエラーが表示されます。

org.apache.kafka.common.errors.TimeoutException: 60000 ミリ秒後にメタデータを更新できませんでした。

10 分が経過した後、ブローカー 2 は新しいリーダーであるため、プロデューサーがブローカー 2 にデータを送信することを期待していましたが、上記の例外を与えて失敗し続けました。プロデューサーのメタデータExpireMsは300000ですが、lastRefreshMsとlastSuccessfullRefreshMsは同じです。

プロデューサー側でカフカの新しいプロデューサー実装を使用しています。

プロデューサーが開始されると、1 つのブローカーにバインドされ、そのブローカーがダウンした場合、クラスター内の別のブローカーに接続しようとさえしないようです。

しかし、ブローカーがダウンした場合、利用可能な別のブローカーのメタデータを直接チェックし、それらにデータを送信する必要があります。

ところで、私のトピックは 4 パーティションで、レプリケーション ファクターは 2 です。意味がある場合に備えて、この情報を提供します。

構成パラメータ。

使用事例：

1- BR1 と BR2 プロデュース データを開始 (リーダーは BR1)

2- BR2 生成データを停止します (細かい)

3- BR1 を停止し (これは、現時点でアクティブな作業ブローカーがクラスターに存在しないことを意味します)、BR2 を開始してデータを生成します (リーダーは BR2 ですが失敗しました)。

4- BR1のデータ作成開始(リーダーはBR2のままですが、データは綺麗に作成されています)

5- BR2 を停止します (現在は BR1 がリーダーです)

6- BR1 を停止します (BR1 はまだリーダーです)

7- BR1 のデータ生成を開始します (メッセージは再び正常に生成されます)。

プロデューサーが最新の成功したデータを BR1 に送信し、その後すべてのブローカーがダウンした場合、プロデューサーは、BR2 が起動して新しいリーダーであっても、BR1 が再び起動することを期待します。これは予想される動作ですか？

KAFKA と Spring を統合しようとしています。私の Java アプリケーションは KAFKA サーバーと通信しており、HTTP を使用してアプリケーションを実行するとメッセージも受信しています。

ここで、Spring を使用して KAFKA に SSL を追加したいと思います。SSL KAFKAおよびSPRING KAFKAで指定された変更を行いました。

コマンド ラインを使用して (SSL を使用して) プロデューサーとコンシューマーを実行すると、通信は適切に行われますが、Java アプリケーションの構成を変更し、トピックでメッセージを生成および消費しようとすると、コンシューマーは正常に動作し、メッセージを取得しています。指定されたトピック。

しかし、プロデューサーは期待どおりに動作していません。JAVA アプリケーション プロデューサーから送信されたメッセージがコンシューマーで受信されません (コマンド ラインでも Java アプリ コンシューマーでもありません)。

私はそれをグーグルで検索しましたが、それに対する答えが見つかりませんでした。ポインタはありますか？

PS: KAFKA バージョン kafka 0.9.0.0、JAVA 7、および Spring で使用される SSL プロパティは次のとおりです。

Kafka プロデューサーでこのエラーが発生する

開いているファイルが多すぎます

同じプロデューサーを使用してメッセージを送信し、テナント ID によってプロデューサーをキャッシュしています。
接続を閉じていません..

Kafka-0.8.2.21 つのプロデューサー、1 つのコンシューマー、および 3 つのブローカーをすべて異なるマシンに配置して、マルチノード クラスターをセットアップしようとしています。

でトピックを作成中にproducer、エラーが発生しましorg.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /brokers/idsた。完全なコンソール出力は、こちらから入手できます。Kafka Producerのログにエラーはありません。

実行に使用しているコマンドKafkaは次のとおりです。

注: Zookeeper サービスはすべてのサーバーで実行されており、3 つのブローカーすべてで Kafka サーバーが実行されています ( Kafka サーバーが必要なのはブローカーだけですよね? )。

私のProducer.propertiesの構成は次のとおりです。

以下は、私が参照として使用していた多くの記事の一部です。

• Zookeeper と Kafka のインストール: 単一ノードと複数のブローカー クラスター - 2016
• Apache Kafka のインストールと Spark との通信の手順

kafka プロデューサーの数を増やしているときに、エラーを下回っています。ここで何が問題になるのか、誰にもわかりますか？

私のプロデューサー設定を見つけてください: https://gist.github.com/Vibhuti/dbf1c24962b91f2bc217

エラーログ:

参考までに、私のカフカコード：

Kafka プロデューサー クライアントを使用していますが、プロジェクトに log4j 構成がありません。

実行時に、プログラムは本当に必要のない多くの Kafka Debug ログを出力します。

そのため、log4j.properties を追加して、以下のようにログ レベルを ERROR に設定しようとしましたが、うまくいかないようです:

Kafka ログ レベルを変更するにはどうすればよいですか?

Java Rest api で Spring Integration を使用して、メッセージを Kafka に生成します。 https://github.com/spring-projects/spring-integration-kafka

私が言ったら

これはtrueを返します。これは公開を意味します..しかし、トピックログにメッセージを表示できません.. /usr/local/var/lib/kafka-logs/test-0 の下のログを確認しました

コンシューマ コンソールを実行しても、API から発行されたメッセージを表示できません。

しかし、プロデューサーコンソールからメッセージを生成すると、トピックログとコンシューマーコンソールでメッセージを見ることができます..

春の統合レストAPIで同じブローカーリストとトピック名を使用しています..

私の春の設定アウトバウンドxml ..

そしてリスナーの実装..

onError関数にさえ来ないデバッグを試みました。