問題タブ [spring-kafka]

spring-integration-kafka を 1.0.0.M2 から 2.1.0.RELEASE に、クライアントを 0.9.0 から 0.10.0 にアップグレードしています。

以下のような現在のxml構成

これを 2.1.0.RELEASE に変更するにはどうすればよいですか?

~~~~~~~~~~~~~~~

ここで編集：

参照を使用して、私の要件に従ってxmlを変更しました。Consumer Record の読み取り中に小さな問題が発生しました。次のようにペイロードを取得しました

コンシューマーでさらに使用するには、値 (NotificationVo) が必要です。ペイロードの一部として取得する方法は?

~~~~~~~~~~~~~~~

ここで編集：

これは変更された xml 構成ファイルです

~~~~~~~~~~~~~~~

ここで編集：

消費者クラス:

O/P:

~~~~~~~~~~~~~~~

ここで編集：

Consumer クラスのメソッド パラメータを変更した後、予期されたペイロードを受け取りました。

O/P:

最後に期待どおりに動作します。

ご支援ありがとうございます。

Kafka を 0.9.0 から 0.10.0 にアップグレードする際、異なるプロデューサーを異なるトピックに設定する際に問題に直面します。以下に示す XML 構成

2 つのトピックを個別に公開中にエラーが発生しました。スタックトレースは以下の通り

2 つの個別の Serializer クラスと Deserializer クラスを定義しました。しかし、どのように内部的に他のクラスを参照しているのでしょうか? 構成を見逃していませんか?

テスト 1 とテスト 2 の 2 つのトピックを読む必要がある spring-kafka アプリを作成しています。

私の設定は次のようになります：

「test2」の最初からすべてのメッセージを読み取ることができる一方で、「test1」からは最新のメッセージのみを読み取ることができる必要があります。アプリの起動時に「test2」メッセージのみに関心がありますが、「test1」メッセージは、アプリが実行されている限り継続的に読み取る必要があります。

そのような機能を構成する方法はありますか?

Javaのまったく新しい。

次の Kafka リスナー メソッドを持つ spring-kafka アプリを作成します。

アプリのこの部分を個別のスタンドアロン パッケージにリファクタリングするように依頼されたので、「receiveMessage」を呼び出して、「handleMessage」の代わりにメソッド/関数を渡して、それぞれを処理できます。着信メッセージ。

この「receiveMessage」メソッドは、Kafka トピックをリッスンし続けるため、戻りません。

このメソッドを変更/追加するための適切な構文は何でしょうか。これにより、パッケージ/ライブラリとして呼び出すことができ、呼び出し元のアプリによってメッセージ処理メソッドが渡されます。

Spring Kafka ドキュメントのバッチ パターンを使用して、Spring Boot アプリで Kafka コンシューマーをセットアップしました。トピックに負荷をかけるまでは機能しているように見えました。その後、オフセットは更新されず、コンシューマー メッセージの同じバッチが何度も処理されます。

デフォルトでは、バッチ サイズは、フレームワークのアルゴリズムに基づいて動的に表示されます。そのため、トピックがますます遅れるにつれて、メッセージのバッチ処理は数百から数千になりました。バッチ サイズ (max.poll.records) を 25 に設定することで問題を回避できましたが、根本的な原因を理解するまでは Kafka/Spring-Kafka を信頼していません。

バッチが完了するまでに 3000 ミリ秒から 15000 ミリ秒かかります。

ファクトリのプロパティ:

工場ビルド:

リスナー サービスは、次の注釈を使用します。

Spring-Kafka では、最初から Kafka トピックを再利用したいと考えています。group.idを Kafka にとって未知のものに変更することでこれを行うと、もちろん機能します。

ただし、オフセットを 0 に設定してやり直すと失敗します。

私が得るエラー：

これに詳しい人はいますか？

私は使用Kafka 0.10.1.0しています

ここで言及した、Kafka Metrics を Spring Boot Actuator と統合するという私の考えに戻ることにしました。

https://github.com/spring-projects/spring-boot/issues/6227

今のところ、Spring Boot にマージしたい作業コードを含む別の「サンドボックス」プロジェクトがあります。そして今、私は少し混乱しています。私のテストの一部では、Mock Kafka の「超セキュア」クラスに powermock が必要です。

ただし、Powermock は Spring Boot では使用されません。

私は何をすべきか？

1. spring-boot-dependencies と spring-boot-actuator の両方に、最新の安定した powermock を追加します。
2. spring-boot-actuator のみに最新の安定した powermock を追加します。これを隠れた秘密のヘルパーにします。
3. パワーモックを必要とするテストを除外します。
4. Kafka Metrics のことは忘れてください。それらは巨大で恐ろしく、私たちの親切でフレンドリーな Spring Boot Actuator では誰もそれらを望んでいません。