次のプロパティは Kafka コンシューマーでのみ使用可能であり、接頭辞 spring.cloud.stream.kafka.bindings.<channelName>.consumer. を付ける必要があります。

アプリケーションが起動すると、割り当てられた各パーティションの初期位置は、2 つのプロパティ startOffset と resetOffsets に依存します。resetOffsets が false の場合、通常の Kafka コンシューマー auto.offset.reset [Apache] (英語) セマンティクスが適用されます。つまり、バインディングのコンシューマーグループのパーティションにコミットされたオフセットがない場合、位置は earliest または latest です。デフォルトでは、明示的な group を使用したバインディングは earliest を使用し、匿名のバインディング（group を使用しない）は latest を使用します。これらのデフォルトは、startOffset バインディングプロパティを設定することで上書きできます。特定の group で初めてバインディングが開始されたとき、コミットされたオフセットはありません。コミットされたオフセットが存在しないもう 1 つの条件は、オフセットの有効期限が切れているかどうかです。最新のブローカー（2.1 以降）およびデフォルトのブローカープロパティでは、最後のメンバーがグループを離れてから 7 日後にオフセットの有効期限が切れます。詳細については、offsets.retention.minutes [Apache] (英語) ブローカーのプロパティを参照してください。

resetOffsets が true の場合、バインダーは、このバインディングがトピックから消費されたことがないかのように、ブローカーにコミットされたオフセットがない場合に適用されるセマンティクスと同様のセマンティクスを適用します。つまり、現在コミットされているオフセットは無視されます。

以下は、これが使用される可能性のある 2 つのユースケースです。

トピックオフセットの詳細な制御については、KafkaBindingRebalanceListener を使用するを参照してください。リスナーが提供されている場合、resetOffsets を true に設定しないでください。設定しないと、エラーが発生します。

バージョン 3.0 以降、spring.cloud.stream.binding.<name>.consumer.batch-mode が true に設定されている場合、Kafka Consumer のポーリングによって受信されたすべてのレコードは、List<?> としてリスナーメソッドに提示されます。それ以外の場合、メソッドは一度に 1 つのレコードで呼び出されます。バッチのサイズは、Kafka コンシューマープロパティ max.poll.records、fetch.min.bytes、fetch.max.wait.ms によって制御されます。詳細については、Kafka のドキュメントを参照してください。

バッチモードは @StreamListener ではサポートされていないことに注意してください。これは、新しい関数型プログラミングモデルでのみ機能します。

次のプロパティは Kafka プロデューサーのみが使用でき、接頭辞 spring.cloud.stream.kafka.bindings.<channelName>.producer. を付ける必要があります。

このセクションでは、特定のシナリオでの前述のプロパティの使用箇所を示します。

デフォルトでは、レコードは元のレコードと同じパーティションを使用して Dead-Letter トピックに公開されます。つまり、Dead-Letter トピックには、少なくとも元のレコードと同じ数のパーティションが必要です。

この動作を変更するには、DlqPartitionFunction 実装を @Bean としてアプリケーションコンテキストに追加します。そのような Bean は 1 つだけ存在できます。この機能は、コンシューマーグループ、障害が発生した ConsumerRecord、例外とともに提供されます。例: 常にパーティション 0 にルーティングする場合は、次を使用できます。

DLQ トピックのカスタム名を定義することもできます。これを行うには、アプリケーションコンテキストへの @Bean として DlqDestinationResolver の実装を作成します。バインダーがそのような Bean を検出すると、それが優先されます。それ以外の場合は、dlqName プロパティが使用されます。これらのどちらも見つからない場合は、デフォルトで error.<destination>.<group> になります。@Bean としての DlqDestinationResolver の例を次に示します。

DlqDestinationResolver の実装を提供する際に留意すべき重要なことの 1 つは、バインダーのプロビジョナーがアプリケーションのトピックを自動作成しないことです。これは、実装が送信する可能性のあるすべての DLQ トピックの名前をバインダーが推測する方法がないためです。この戦略を使用して DLQ 名を提供する場合、それらのトピックが事前に作成されていることを確認するのはアプリケーションの責任です。

フレームワークは、ユーザーがデッドレターメッセージをどのように処理するかを予測できないため、メッセージを処理するための標準的なメカニズムを提供していません。デッドレタリングの理由が一時的なものである場合は、メッセージを元のトピックにルーティングして戻すことをお勧めします。ただし、課題が永続的な課題である場合は、無限ループが発生する可能性があります。このトピック内のサンプル Spring Boot アプリケーションは、これらのメッセージを元のトピックにルーティングする方法の例ですが、3 回試行すると、メッセージは「駐車場」トピックに移動します。このアプリケーションは、デッドレターのトピックから読み取る別の spring-cloud-stream アプリケーションです。5 秒間メッセージが受信されないと終了します。

例では、元の宛先が so8400out であり、コンシューマーグループが so8400 であると想定しています。

考慮すべき戦略がいくつかあります。

次のコードリストは、サンプルアプリケーションを示しています。

Spring Cloud Stream 3.0.0 以降、Kafka Streams バインダーを使用すると、Java 8 で使用可能な関数型プログラミングスタイルを使用してアプリケーションを設計および開発できます。これは、アプリケーションを java.util.function.Function 型または java.util.function.Consumer 型のラムダ式として簡潔に表すことができることを意味します。

非常に基本的な例を見てみましょう。

シンプルではありますが、これは完全なスタンドアロン Spring Boot アプリケーションであり、ストリーム処理に Kafka ストリームを活用しています。これは、送信バインディングがなく、受信バインディングが 1 つしかないコンシューマーアプリケーションです。アプリケーションはデータを消費し、KStream キーと値からの情報を標準出力に記録するだけです。アプリケーションには、SpringBootApplication アノテーションと Bean としてマークされたメソッドが含まれています。Bean メソッドは、KStream でパラメーター化された型 java.util.function.Consumer です。次に、実装では、本質的にラムダ式であるコンシューマーオブジェクトを返します。ラムダ式の中には、データを処理するためのコードが含まれています。

このアプリケーションでは、型 KStream の単一の入力バインディングがあります。バインダーは、process-in-0 という名前でアプリケーションのこのバインディングを作成します。つまり、関数 Bean 名の後にダッシュ文字（-）が続き、リテラル in の後に別のダッシュが続き、パラメーターの順序位置が続きます。このバインディング名を使用して、宛先などの他のプロパティを設定します。例: spring.cloud.stream.bindings.process-in-0.destination=my-topic。

uber-jar（kstream-consumer-app.jar など）としてビルドしたら、次のように上記の例を実行できます。

アプリケーションが Spring の Component アノテーションを使用して機能 Bean を定義することを選択した場合、バインダーはそのモデルもサポートします。上記の関数 Bean は以下のように書き直すことができます。

これは別の例で、入力バインディングと出力バインディングの両方を備えたフルプロセッサーです。これは、アプリケーションがトピックからデータを受信し、各単語の出現回数がタンブリングタイムウィンドウで計算される典型的な単語カウントの例です。

ここでも、これは完全な Spring Boot アプリケーションです。ここでの最初のアプリケーションとの違いは、Bean メソッドの型が java.util.function.Function であるということです。Function の最初のパラメーター化された型は入力 KStream 用で、2 番目の型は出力用です。メソッド本体には、Function 型のラムダ式が提供され、実装として実際のビジネスロジックが提供されます。前述のコンシューマーベースのアプリケーションと同様に、ここでの入力バインディングは、デフォルトで process-in-0 という名前が付けられています。出力の場合、バインディング名も自動的に process-out-0 に設定されます。

uber-jar（wordcount-processor.jar など）としてビルドしたら、次のように上記の例を実行できます。

このアプリケーションは、Kafka トピック words からのメッセージを消費し、計算結果は出力トピック counts に公開されます。

Spring Cloud Stream は、受信トピックと発信トピックの両方からのメッセージが KStream オブジェクトとして自動的にバインドされることを保証します。開発者は、コードのビジネス面、つまりプロセッサーに必要なロジックの記述に専念できます。Kafka Streams インフラストラクチャに必要な Kafka Streams 固有の構成のセットアップは、フレームワークによって自動的に処理されます。

上で見た 2 つの例には、単一の KStream 入力バインディングがあります。どちらの場合も、バインディングは単一のトピックからレコードを受け取りました。複数のトピックを単一の KStream バインディングに多重化する場合は、以下の宛先としてコンマ区切りの Kafka トピックを指定できます。

spring.cloud.stream.bindings.process-in-0.destination=topic-1,topic-2,topic-3

さらに、トピックを通常の表現と照合する場合は、トピックパターンを宛先として提供することもできます。

spring.cloud.stream.bindings.process-in-0.destination=input.*

バインダーの 3.1.0 バージョンから始めて、Kafka Streams バインダーベースのアプリケーションには上記の関数型プログラミングモデルを使用することをお勧めします。StreamListener のサポートは、Spring Cloud Stream の 3.1.0 以降で非推奨になりました。以下に、参照として StreamListener ベースの Kafka ストリームプロセッサーの詳細を示します。

以下は、StreamListener を使用したワードカウントの例に相当します。

ご覧のとおり、完全なアプリケーションにするには、EnableBinding と、StreamListener や SendTo などの他の追加のアノテーションを提供する必要があるため、これはもう少し冗長です。EnableBinding は、バインディングを含むバインディングインターフェースを指定する場所です。この場合、次の契約を持つストック KafkaStreamsProcessor バインディングインターフェースを使用しています。

これらの宣言を含むバインディングインターフェースを使用しているため、バインダーは入力 KStream と出力 KStream のバインディングを作成します。

機能スタイルで提供されるプログラミングモデルの明らかな違いに加えて、ここでメンションする必要がある 1 つの特定のことは、バインディング名がバインディングインターフェースで指定するものであるということです。例: 上記のアプリケーションでは、KafkaStreamsProcessor を使用しているため、バインディング名は input と output です。バインディングプロパティはそれらの名前を使用する必要があります。たとえば、spring.cloud.stream.bindings.input.destination、spring.cloud.stream.bindings.output.destination などです。バインダーがアプリケーションのバインディング名を生成するため、これは機能スタイルとは根本的に異なることに注意してください。これは、アプリケーションが EnableBinding を使用する機能モデルにバインディングインターフェースを提供しないためです。

これは、2 つの入力があるシンクの別の例です。

以下は、上記で見たのと同じ BiFunction ベースのプロセッサーと同等の StreamListener です。

最後に、3 つの入力とカレー関数を備えたアプリケーションに相当する StreamListener を次に示します。

EnableBinding を提供するだけでなく、独自のカスタムバインディングインターフェースも作成する必要があるため、上記の 2 つの例はさらに冗長であることに気付くかもしれません。機能モデルを使用すると、これらすべての儀式の詳細を回避できます。

Kafka Streams バインダーによって提供される一般的なプログラミングモデルの検討に移る前に、複数の出力バインディングの StreamListener バージョンを示します。

要約すると、Kafka Streams バインダーを使用する際のさまざまなプログラミングモデルの選択を確認しました。

バインダーは、入力で KStream、KTable、GlobalKTable のバインディング機能を提供します。KTable および GlobalKTable バインディングは、入力でのみ使用できます。バインダーは、KStream の入力バインディングと出力バインディングの両方をサポートします。

Kafka Streams バインダーのプログラミングモデルの結果は、バインダーが完全に機能するプログラミングモデルを使用するか、StreamListener ベースの命令型アプローチを使用する柔軟性を提供することです。

バインダーを使用すると、単一の Spring Cloud Stream アプリケーション内に複数の Kafka ストリームプロセッサーを含めることができます。以下のような申し込みが可能です。

この場合、バインダーは、異なるアプリケーション ID を持つ 3 つの個別の Kafka Streams オブジェクトを作成します（これについては以下で詳しく説明します）。ただし、アプリケーションに複数のプロセッサーがある場合は、どの機能をアクティブにする必要があるかを Spring Cloud Stream に通知する必要があります。機能を有効にする方法は次のとおりです。

spring.cloud.stream.function.definition: process;anotherProcess;yetAnotherProcess

特定の機能をすぐにアクティブにしたくない場合は、このリストから削除できます。

これは、同じアプリケーションに単一の Kafka Streams プロセッサーと他の型の Function Bean があり、異なるバインダーを介して処理される場合にも当てはまります。(たとえば、通常の Kafka メッセージチャネルバインダーに基づく関数 Bean)

アプリケーション ID は、Kafka Streams アプリケーションに提供する必要のある必須のプロパティです。Spring Cloud Stream Kafka Streams バインダーを使用すると、このアプリケーション ID を複数の方法で構成できます。

アプリケーションにシングルプロセッサーまたは StreamListener が 1 つしかない場合は、次のプロパティを使用して、これをバインダーレベルで設定できます。

spring.cloud.stream.kafka.streams.binder.applicationId.

便宜上、プロセッサーが 1 つしかない場合は、プロパティとして spring.application.name を使用してアプリケーション ID を委譲することもできます。

アプリケーションに複数の Kafka Streams プロセッサーがある場合は、プロセッサーごとにアプリケーション ID を設定する必要があります。機能モデルの場合は、プロパティとして各関数に接続できます。

たとえば次の機能があると想像してください。

および

次に、次のバインダーレベルのプロパティを使用して、それぞれのアプリケーション ID を設定できます。

spring.cloud.stream.kafka.streams.binder.functions.process.applicationId

および

spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.applicationId

StreamListener の場合、プロセッサーの最初の入力バインディングでこれを設定する必要があります。

たとえば次の 2 つの StreamListener ベースのプロセッサーがあると想像してください。

次に、次のバインディングプロパティを使用して、このアプリケーション ID を設定する必要があります。

spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId

および

spring.cloud.stream.kafka.streams.bindings.anotherInput.consumer.applicationId

関数ベースのモデルの場合も、アプリケーション ID をバインディングレベルで設定するこのアプローチは機能します。ただし、機能モデルを使用している場合は、上記のようにバインダーレベルで関数ごとに設定する方がはるかに簡単です。

本番デプロイの場合、構成を通じてアプリケーション ID を明示的に指定することを強くお勧めします。これは、アプリケーションを自動スケーリングする場合に特に重要になります。その場合、同じアプリケーション ID で各インスタンスをデプロイしていることを確認する必要があります。

アプリケーションがアプリケーション ID を提供しない場合、その場合、バインダーは静的アプリケーション ID を自動生成します。これは、アプリケーション ID を明示的に提供する必要がないため、開発シナリオで便利です。この方法で生成されたアプリケーション ID は、アプリケーションの再起動時に静的になります。機能モデルの場合、生成されるアプリケーション ID は、関数 Bean 名の後にリテラル applicationID が続きます。たとえば、関数 Bean 名の場合は process の場合は process-applicationID です。StreamListener の場合、関数 Bean 名を使用する代わりに、生成されたアプリケーション ID は、含まれているクラス名、メソッド名、リテラル applicationId を使用します。

デフォルトでは、バインダーは上記の戦略を使用して、機能スタイルを使用するときにバインディング名を生成します。つまり、<function-bean-name>-<in> | <out>-[0..n]、例: process-in-0、process-out-0 など。これらのバインディング名をオーバーライドする場合は、次のプロパティを指定することでそれを行うことができます。

spring.cloud.stream.function.bindings.<default binding name>。デフォルトのバインディング名は、バインダーによって生成された元のバインディング名です。

たとえばたとえば、この機能があります。

バインダーは、process-in-0、process-in-1、process-out-0 という名前のバインディングを生成します。完全に別のもの、おそらくさらにドメイン固有のバインディング名に変更したい場合は、以下のように行うことができます。

spring.cloud.stream.function.bindings.process-in-0=users

spring.cloud.stream.function.bindings.process-in-0=regions

および

spring.cloud.stream.function.bindings.process-out-0=clicks

その後、これらの新しいバインディング名にすべてのバインディングレベルプロパティを設定する必要があります。

上記の関数型プログラミングモデルでは、ほとんどの状況でデフォルトのバインディング名を順守することが理にかなっていることに注意してください。このオーバーライドを実行する必要がある唯一の理由は、構成プロパティの数が多く、バインディングをよりドメインに適したものにマップする場合です。

Kafka Streams アプリケーションを実行するときは、Kafka ブローカーサーバー情報を提供する必要があります。この情報を提供しない場合、バインダーは、デフォルトの localhost:9092 でブローカーを実行していることを想定しています。そうでない場合は、それをオーバーライドする必要があります。これを行うにはいくつかの方法があります。

バインダーレベルのプロパティに関しては、通常の Kafka バインダーである spring.cloud.stream.kafka.binder.brokers を介して提供されるブローカープロパティを使用するかどうかは関係ありません。Kafka Streams バインダーは、最初に Kafka Streams バインダー固有のブローカープロパティが設定されているかどうかを確認し（spring.cloud.stream.kafka.streams.binder.brokers）、見つからない場合は spring.cloud.stream.kafka.binder.brokers を探します。

キーは常にネイティブ Serdes を使用して逆直列化されます。

値の場合、デフォルトでは、受信での逆直列化は Kafka によってネイティブに実行されます。これは、フレームワークによって逆直列化が行われた以前のバージョンの Kafka Streams バインダーからのデフォルトの動作に対する大きな変更であることに注意してください。

Kafka Streams バインダーは、java.util.function.Function|Consumer または StreamListener の型署名を調べて、一致する Serde 型を推測しようとします。Serdes に一致する順序は次のとおりです。

上記の戦略のいずれも機能しなかった場合、アプリケーションは構成を通じて `Serde` を提供する必要があります。これは、バインディングまたはデフォルトの 2 つの方法で構成できます。

最初に、バインダーは、Serde がバインディングレベルで提供されているかどうかを確認します。たとえば次のプロセッサーを使用している場合

次に、以下を使用してバインディングレベル Serde を提供できます。

デフォルトのキー / 値 Serdes を受信デシリアライズに使用する場合は、バインダーレベルで使用できます。

Kafka が提供するネイティブデコードが必要ない場合は、Spring Cloud Stream が提供するメッセージ変換機能を利用できます。ネイティブデコードがデフォルトであるため、Spring Cloud Stream で受信値オブジェクトを逆直列化するには、ネイティブデコードを明示的に無効にする必要があります。

たとえば上記と同じ BiFunction プロセッサーを使用している場合は、spring.cloud.stream.bindings.process-in-0.consumer.nativeDecoding: false すべての入力のネイティブデコードを個別に無効にする必要があります。それ以外の場合は、無効にしないものにはネイティブデコードが引き続き適用されます。

デフォルトでは、Spring Cloud Stream はコンテンツ型として application/json を使用し、適切な json メッセージコンバーターを使用します。次のプロパティと適切な MessageConverter Bean を使用して、カスタムメッセージコンバーターを使用できます。

送信直列化は、受信デ直列化の上記とほぼ同じルールに従います。受信の逆直列化と同様に、Spring Cloud Stream の以前のバージョンからの大きな変更の 1 つは、送信での直列化が Kafka によってネイティブに処理されることです。バインダーの 3.0 バージョンの前は、これはフレームワーク自体によって行われていました。

送信のキーは、バインダーによって推測される一致する Serde を使用して、Kafka によって常に直列化されます。キーの型を推測できない場合は、構成を使用して指定する必要があります。

値 serdes は、受信の逆直列化に使用されるのと同じルールを使用して推測されます。最初に、送信型がアプリケーションで提供された Bean からのものであるかどうかを確認するために一致します。そうでない場合は、- Integer、Long、Short、Double、Float、byte[]、UUID、String などの Kafka によって公開された Serde と一致するかどうかを確認します。それが機能しない場合は、Spring Kafka プロジェクトによって提供される JsonSerde にフォールバックしますが、最初にデフォルトの Serde 構成を調べて、一致するものがあるかどうかを確認します。これらはすべて、アプリケーションに対して透過的に行われることに注意してください。これらのいずれも機能しない場合、ユーザーは構成で使用する Serde を提供する必要があります。

上記と同じ BiFunction プロセッサーを使用しているとしましょう。次に、送信キー / 値 Serdes を次のように構成できます。

Serde 推論が失敗し、バインディングレベルの Serdes が提供されていない場合、バインダーは JsonSerde にフォールバックしますが、デフォルトの Serdes で一致するものを探します。

デフォルトの serdes は、デシリアライズで説明されている上記と同じ方法で構成されます。

spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde

アプリケーションが分岐機能を使用し、複数の出力バインディングがある場合、これらはバインディングごとに構成する必要があります。繰り返しになりますが、バインダーが Serde 型を推測できる場合は、この構成を行う必要はありません。

Kafka が提供するネイティブエンコーディングは必要ないが、フレームワークが提供するメッセージ変換を使用する場合は、ネイティブエンコーディングがデフォルトであるため、ネイティブエンコーディングを明示的に無効にする必要があります。たとえば上記と同じ BiFunction プロセッサーを使用している場合は、spring.cloud.stream.bindings.process-out-0.producer.nativeEncoding: false 分岐の場合は、すべての出力のネイティブエンコーディングを個別に無効にする必要があります。それ以外の場合は、無効にしないものにはネイティブエンコーディングが引き続き適用されます。

Spring Cloud Stream によって変換が行われる場合、デフォルトでは、コンテンツ型として application/json が使用され、適切な json メッセージコンバーターが使用されます。次のプロパティと対応する MessageConverter Bean を使用して、カスタムメッセージコンバーターを使用できます。

ネイティブエンコーディング / デコーディングが無効になっている場合、バインダーはネイティブ Serdes の場合のように推論を行いません。アプリケーションは、すべての構成オプションを明示的に提供する必要があります。そのため、Spring Cloud Stream Kafka Streams アプリケーションを作成するときは、通常、逆直列化のデフォルトオプションを使用し、Kafka Streams が提供するネイティブの逆直列化を使用することをお勧めします。フレームワークによって提供されるメッセージ変換機能を使用する必要がある 1 つのシナリオは、アップストリームプロデューサーが特定の直列化戦略を使用している場合です。その場合、ネイティブメカニズムが失敗する可能性があるため、一致する逆直列化戦略を使用する必要があります。デフォルトの Serde メカニズムに依存する場合、アプリケーションは、バインダーが適切な Serde で受信と送信を正しくマップする方法を持っていることを確認する必要があります。そうしないと、問題が発生する可能性があります。

上で概説したデータの逆直列化アプローチは、プロセッサーのエッジ、つまり受信と送信にのみ適用可能であることに注意してください。ビジネスロジックでは、Serde オブジェクトを明示的に必要とする Kafka StreamsAPI を呼び出す必要がある場合があります。これらは依然としてアプリケーションの責任であり、開発者がそれに応じて処理する必要があります。

Kafka Streams バインダーでは、次のプロパティを使用して、上記の逆直列化例外ハンドラーを指定できます。

または

上記の 2 つの逆直列化例外ハンドラーに加えて、バインダーは、誤ったレコード（ポイズンピル）を DLQ（デッドレターキュー）トピックに送信するための 3 番目のハンドラーも提供します。この DLQ 例外ハンドラーを有効にする方法は次のとおりです。

上記のプロパティを設定すると、逆直列化エラーのすべてのレコードが自動的に DLQ トピックに送信されます。

DLQ メッセージを公開するトピック名は以下のように設定できます。

関数インターフェースである DlqDestinationResolver の実装を提供できます。DlqDestinationResolver は、ConsumerRecord と例外を入力として受け取り、トピック名を出力として指定できるようにします。Kafka ConsumerRecord にアクセスすることにより、BiFunction の実装でヘッダーレコードをイントロスペクトできます。

これは、DlqDestinationResolver の実装を提供する例です。

DlqDestinationResolver の実装を提供する際に留意すべき重要なことの 1 つは、バインダーのプロビジョナーがアプリケーションのトピックを自動作成しないことです。これは、実装が送信する可能性のあるすべての DLQ トピックの名前をバインダーが推測する方法がないためです。この戦略を使用して DLQ 名を提供する場合、それらのトピックが事前に作成されていることを確認するのはアプリケーションの責任です。

DlqDestinationResolver がアプリケーションに Bean として存在する場合、それが優先されます。このアプローチに従わず、構成を使用して静的 DLQ 名を指定する場合は、次のプロパティを設定できます。

これが設定されている場合、エラーレコードはトピック custom-dlq に送信されます。アプリケーションが上記の戦略のいずれも使用していない場合、error.<input-topic-name>.<application-id> という名前の DLQ トピックが作成されます。たとえば、バインディングの宛先トピックが inputTopic で、アプリケーション ID が process-applicationId の場合、デフォルトの DLQ トピックは error.inputTopic.process-applicationId です。DLQ を有効にする場合は、入力バインディングごとに DLQ トピックを明示的に作成することを常にお勧めします。

プロパティ spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler は、アプリケーション全体に適用できます。これは、同じアプリケーションに複数の関数または StreamListener メソッドがある場合、このプロパティがそれらすべてに適用されることを意味します。ただし、単一のプロセッサー内に複数のプロセッサーまたは複数の入力バインディングがある場合は、バインダーが入力コンシューマーバインディングごとに提供するよりきめ細かい DLQ コントロールを使用できます。

次のプロセッサーを使用している場合

最初の入力バインディングでのみ DLQ を有効にし、2 番目のバインディングで skipAndContinue を有効にしたい場合は、以下のようにコンシューマーで実行できます。

spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.deserializationExceptionHandler: sendToDlq spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.deserializationExceptionHandler: skipAndContinue

この方法で逆直列化例外ハンドラーを設定すると、バインダーレベルで設定するよりも優先されます。

デフォルトでは、レコードは元のレコードと同じパーティションを使用して Dead-Letter トピックに公開されます。つまり、Dead-Letter トピックには、少なくとも元のレコードと同じ数のパーティションが必要です。

この動作を変更するには、DlqPartitionFunction 実装を @Bean としてアプリケーションコンテキストに追加します。そのような Bean は 1 つだけ存在できます。この機能は、コンシューマーグループ（ほとんどの状況でアプリケーション ID と同じ）、失敗した ConsumerRecord、例外とともに提供されます。例: 常にパーティション 0 にルーティングする場合は、次を使用できます。

Kafka Streams バインダーの例外処理機能を使用する際に留意すべき点がいくつかあります。

上記の逆直列化例外ハンドラーのサポートとは異なり、バインダーは、本番例外を処理するためのそのようなファーストクラスのメカニズムを提供しません。ただし、StreamsBuilderFactoryBean カスタマイザを使用して本番例外ハンドラーを構成することはできます。詳細については、以下の次のセクションを参照してください。

次の API メソッドを使用して、指定されたストアとキーの組み合わせに関連付けられた KeyQueryMetadata オブジェクトを取得します。

次の API メソッドを使用して、指定されたストアとキーの組み合わせに関連付けられた KakfaStreams オブジェクトを取得します。

上記のように、バインダーは、グローバル状態ストアを機能として登録するためのファーストクラスの方法を提供しません。そのためには、カスタマイザーを使用する必要があります。これがその方法です。

繰り返しますが、複数のプロセッサーがある場合は、上記のようにアプリケーション ID を使用して他の StreamsBuilderFactoryBean オブジェクトを除外することにより、グローバル状態ストアを右側の StreamsBuilder に接続する必要があります。

エラー処理のセクションでは、バインダーが本番例外を処理するためのファーストクラスの方法を提供しないことを示しました。その場合でも、StreamsBuilderFacotryBean カスタマイザーを使用して実動例外ハンドラーを登録することができます。下記参照。

繰り返しになりますが、複数のプロセッサーがある場合は、正しい StreamsBuilderFactoryBean に対して適切に設定することをお勧めします。構成プロパティを使用してこのような本番例外ハンドラーを追加することもできますが（詳細については以下を参照）、プログラムによるアプローチを選択する場合はこれがオプションです。

次のプロパティはバインダーレベルで使用でき、spring.cloud.stream.kafka.streams.binder. のプレフィックスを付ける必要があります。Kafka Streams バインダーで再利用されるプロパティを提供する Kafka バインダーには、spring.cloud.stream.kafka.binder ではなく spring.cloud.stream.kafka.streams.binder のプレフィックスを付ける必要があります。このルールの唯一の例外は、Kafka ブートストラップサーバープロパティを定義する場合です。この場合、どちらのプレフィックスも機能します。

ストリーム構成に含まれる可能性のあるすべてのプロパティの詳細については、Apache Kafka ストリームドキュメントの StreamsConfig JavaDocs を参照してください。StreamsConfig から設定できるすべての構成は、これを通じて設定できます。このプロパティを使用する場合、これはバインダーレベルのプロパティであるため、アプリケーション全体に適用されます。アプリケーションにプロセッサーが 個以上ある場合は、すべてのプロセッサーがこれらのプロパティを取得します。application.id などのプロパティの場合、これは問題になるため、このバインダーレベルの configuration プロパティを使用して StreamsConfig のプロパティがどのようにマップされるかを慎重に調べる必要があります。

次のプロパティは Kafka Streams プロデューサーでのみ使用可能であり、接頭辞 spring.cloud.stream.kafka.streams.bindings.<binding name>.producer. を付ける必要があります。便宜上、複数の出力バインディングがあり、それらすべてに共通の値が必要な場合は、接頭辞 spring.cloud.stream.kafka.streams.default.producer. を使用して構成できます。

次のプロパティは Kafka Streams コンシューマーで使用でき、接頭辞 spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer. を付ける必要があります。便宜上、複数の入力バインディングがあり、それらすべてに共通の値が必要な場合は、接頭辞 spring.cloud.stream.kafka.streams.default.consumer. を使用して構成できます。

メモ: コンシューマーで resetOffsets を使用しても、Kafka Streams バインダーには影響しません。メッセージチャネルベースのバインダーとは異なり、Kafka Streams バインダーはオンデマンドで開始または終了しようとはしません。

Kafka ストリームでは、num.stream.threads プロパティを使用して、プロセッサーが作成できるスレッドの数を制御できます。これは、バインダー、関数、プロデューサー、コンシューマーレベルで上記のさまざまな configuration オプションを使用して実行できます。コア Spring Cloud Stream がこの目的のために提供する concurrency プロパティを使用することもできます。これを使用するときは、コンシューマーに使用する必要があります。関数または StreamListener のいずれかに複数の入力バインディングがある場合は、これを最初の入力バインディングに設定します。たとえば spring.cloud.stream.bindings.process-in-0.consumer.concurrency を設定すると、バインダーによって num.stream.threads として変換されます。複数のプロセッサーがあり、1 つのプロセッサーがバインディングレベルの同時実行性を定義し、他のプロセッサーは定義しない場合、バインディングレベルの同時実行性がないプロセッサーは、デフォルトで spring.cloud.stream.kafka.streams.binder.configuration.num.stream.threads で指定されたバインダー全体のプロパティに戻ります。このバインダー構成が使用できない場合、アプリケーションは Kafka Streams によって設定されたデフォルトを使用します。

開発者として、Kafka トピックのレコードを処理するコンシューマーアプリケーションを作成したいと思います。ただし、処理中にエラーが発生した場合は、アプリケーションを完全に停止させたくありません。代わりに、エラーのあるレコードを DLT（Dead-Letter-Topic）に送信してから、新しいレコードの処理を続行したいと思います。

この問題の解決策は、Spring Cloud Stream の DLQ 機能を使用することです。この議論の目的のために、以下が私たちのプロセッサー関数であると仮定しましょう。

これは、処理するすべてのレコードに対して例外をスローする非常に簡単な関数ですが、この関数を使用して、他の同様の状況に拡張することができます。

エラーのあるレコードを DLT に送信するには、次の構成を提供する必要があります。

DLQ をアクティブ化するには、アプリケーションでグループ名を指定する必要があります。匿名のコンシューマーは DLQ 機能を使用できません。また、Kafka コンシューマーバインディングの enableDLQ プロパティを true に設定して、DLQ を有効にする必要があります。最後に、Kafka コンシューマーバインディングで dlqName を提供することにより、オプションで DLT 名を提供できます。それ以外の場合は、この場合はデフォルトで input-topic-dlq.my-group.error になります。

上記のコンシューマーの例では、ペイロードの型は byte[] であることに注意してください。デフォルトでは、Kafka バインダーの DLQ プロデューサーは型 byte[] のペイロードを想定しています。そうでない場合は、適切なシリアライザーの構成を提供する必要があります。例: コンシューマー関数を次のように書き直してみましょう。

次に、Spring Cloud Stream に、DLT に書き込むときにデータをどのように直列化するかを指示する必要があります。このシナリオの変更された構成は次のとおりです。

これは上記のレシピに似ていますが、開発者として、再試行の処理方法を構成したいと思います。

上記のレシピに従った場合、処理でエラーが発生したときに、Kafka バインダーに組み込まれているデフォルトの再試行オプションを取得します。

デフォルトでは、バインダーは 1 秒の初期遅延で最大 3 回試行され、2.0 乗数はそれぞれ最大遅延 10 秒でバックオフします。これらの構成はすべて、次のように変更できます。

必要に応じて、ブール値のマップを提供することにより、再試行可能な例外のリストを提供することもできます。元:

デフォルトでは、上記のマップにリストされていない例外は再試行されます。それが望ましくない場合は、次のように指定して無効にすることができます。

独自の RetryTemplate を提供し、バインダーによってスキャンされて使用される @StreamRetryTemplate としてマークすることもできます。これは、より高度な再試行戦略とポリシーが必要な場合に役立ちます。

複数の @StreamRetryTemplate Bean がある場合は、プロパティを使用して、バインディングで必要な Bean を指定できます。

Kafka コンシューマーで deserilzartion 例外が発生したプロセッサーがあります。Spring Cloud Stream DLQ メカニズムがそのシナリオをキャッチすることを期待しますが、そうではありません。どうすればこれを処理できますか？

Spring Cloud Stream が提供する通常の DLQ メカニズムは、Kafka コンシューマーが回復不能な逆直列化の例外をスローした場合には役立ちません。これは、コンシューマーの poll() メソッドが戻る前でもこの例外が発生するためです。Spring for Apache Kafka プロジェクトは、この状況でバインダーを支援するためのいくつかの優れた方法を提供します。調べてみましょう。

これが私たちの機能であると仮定します:

これは、String パラメーターを受け取る簡単な関数です。

Spring Cloud Stream が提供するメッセージコンバーターをバイパスし、代わりにネイティブデシリアライザーを使用したいと考えています。String 型の場合はあまり意味がありませんが、AVRO などのより複雑な型の場合は、外部デシリアライザーに依存する必要があるため、変換を Kafka に委譲する必要があります。

ここで、コンシューマーがデータを受け取ったときに、脱セリル化エラーを引き起こす不良レコードがあると仮定します。たとえば、誰かが String ではなく Integer を渡した可能性があります。その場合、アプリケーションで何もしなければ、例外はチェーンを介して伝播され、アプリケーションは最終的に終了します。

これを処理するために、SeekToCurrentErrorHandler を構成する ListenerContainerCustomizer @Bean を追加できます。この SeekToCurrentErrorHandler は DeadLetterPublishingRecoverer で構成されています。また、コンシューマー用に ErrorHandlingDeserializer を構成する必要があります。それは多くの複雑なことのように聞こえますが、実際には、この場合、これらの 3 つの Bean に要約されます。

それぞれを分析してみましょう。最初のものは、SeekToCurrentErrorHandler を取る ListenerContainerCustomizer Bean です。これで、コンテナーはその特定のエラーハンドラーでカスタマイズされます。コンテナーのカスタマイズについて詳しくは、こちらを参照してください。

2 番目の Bean は、DLT への公開で構成された SeekToCurrentErrorHandler です。SeekToCurrentErrorHandler の詳細については、こちらを参照してください。

3 番目の Bean は、DLT への送信を最終的に担当する DeadLetterPublishingRecoverer です。デフォルトでは、DLT トピックの名前は ORIGINAL_TOPIC_NAME.DLT です。ただし、変更できます。詳細については、ドキュメントを参照してください。

また、アプリケーション構成を介して ErrorHandlingDeserializer を構成する必要があります。

ErrorHandlingDeserializer は実際のデシリアライザーに委譲します。エラーが発生した場合は、レコードのキー / 値を null に設定し、メッセージの生のバイトを含めます。次に、ヘッダーに例外を設定し、このレコードをリスナーに渡します。リスナーは、登録されたエラーハンドラーを呼び出します。

必要な構成は次のとおりです。

バインディングの configuration プロパティを介して ErrorHandlingDeserializer を提供しています。また、委譲する実際のデシリアライザーが StringDeserializer であることも示しています。

上記の dlq プロパティは、このレシピの説明には関係がないことに注意してください。これらは、純粋にアプリケーションレベルのエラーに対処することのみを目的としています。

Spring Cloud Stream Kafka コンシューマーアプリケーションを作成したいのですが、Kafka コンシューマーオフセットをどのように管理するかがわかりません。説明できますか？

これに関するドキュメントセクションを読んで、完全に理解することをお勧めします。

これが gist です。

Kafka は、デフォルトで 2 種類のオフセット（earliest と latest）をサポートしています。それらのセマンティクスは、それらの名前から自明です。

初めてコンシューマーを実行していると仮定します。Spring Cloud Stream アプリケーションで group.id を見逃すと、匿名のコンシューマーになります。匿名のコンシューマーがある場合は常に、その場合、Spring Cloud Stream アプリケーションは、デフォルトで、トピックパーティションで使用可能な latest オフセットから開始されます。一方、group.id を明示的に指定すると、デフォルトでは、Spring Cloud Stream アプリケーションは、トピックパーティションで使用可能な earliest オフセットから開始されます。

上記の両方の場合（明示的なグループと匿名グループを持つコンシューマー）、開始オフセットは、プロパティ spring.cloud.stream.kafka.bindings.<binding-name>.consumer.startOffset を使用し、earliest または latest のいずれかに設定することで切り替えることができます。

ここで、以前にコンシューマーを実行し、再度開始したと仮定します。この場合、コンシューマーがコンシューマーグループに対してすでにコミットされたオフセットを見つけるため、上記の場合の開始オフセットセマンティクスは適用されません（匿名コンシューマーの場合、アプリケーションは group.id を提供しませんが、バインダーは自動生成）。最後にコミットされたオフセット以降をピックアップするだけです。これは、startOffset 値が提供されている場合でも当てはまります。

ただし、resetOffsets プロパティを使用して、コンシューマーが最後にコミットされたオフセットから開始するデフォルトの動作をオーバーライドできます。これを行うには、プロパティ spring.cloud.stream.kafka.bindings.<binding-name>.consumer.resetOffsets を true （デフォルトでは false）に設定します。次に、startOffset 値（earliest または latest のいずれか）を指定していることを確認してください。これを実行してからコンシューマーアプリケーションを起動すると、起動するたびに、これが初めて起動するかのように起動し、パーティションのコミットされたオフセットを無視します。

Kafka バインダーを使用すると、オフセットを earliest または latest のいずれかに設定できることはわかっていますが、オフセットを主要な任意のオフセットにシークする必要があります。Spring Cloud Stream Kafka バイナーを使用してこれを実現する方法はありますか？

以前、Kafka バインダーを使用して基本的なオフセット管理に取り組む方法を見てきました。デフォルトでは、バインダーは、少なくともその reipce で見たメカニズムを介して、任意のオフセットに巻き戻すことを許可しません。ただし、このユースケースを実現するためにバインダーが提供する低レベルの戦略がいくつかあります。調べてみましょう。

まず、earliest または latest 以外の任意のオフセットにリセットする場合は、resetOffsets 構成をデフォルトの false のままにしてください。次に、型 KafkaBindingRebalanceListener のカスタム Bean を提供する必要があります。これは、すべてのコンシューマーバインディングに注入されます。これはいくつかのデフォルトのメソッドが付属しているインターフェースですが、ここに興味を持っているメソッドがあります:

詳細を見てみましょう。

基本的に、このメソッドは、トピックパーティションの最初の割り当て中、またはリバランス後に毎回呼び出されます。説明をわかりやすくするために、トピックが foo であり、4 つのパーティションがあると仮定します。最初は、グループ内の 1 つのコンシューマーのみを開始し、このコンシューマーはすべてのパーティションから消費します。コンシューマーが初めて起動すると、4 つのパーティションすべてが最初に割り当てられます。ただし、デフォルトで消費するパーティションを開始するのではなく（グループを定義しているため、earliest）、パーティションごとに、任意のオフセットをシークした後に消費するようにします。以下のように、特定のオフセットから消費するビジネスケースがあると想像してください。

これは、上記の方法を以下のように実装することで実現できます。

これは単なる基本的な実装です。実際のユースケースはこれよりもはるかに複雑であり、それに応じて調整する必要がありますが、これは確かに基本的なスケッチを提供します。コンシューマー seek が失敗すると、いくつかのランタイム例外がスローされる可能性があり、その場合の対処方法を決定する必要があります。

2 番目のコンシューマーを追加すると、リバランスが発生し、一部のパーティションが移動します。新しいコンシューマーがパーティション 2 と 3 を取得するとします。この新しい Spring Cloud Stream コンシューマーがこの onPartitionsAssigned メソッドを呼び出すと、これがこのコンシューマーのパーティション 2 および 3 の最初の割り当てであることがわかります。initial 引数の条件付きチェックのためにシーク操作を実行します。最初のコンシューマーの場合、現在は 0 と 1 のパーティションしかありません。ただし、このコンシューマーの場合、これは単なるリバランスイベントであり、最初の割り当てとは見なされません。initial 引数の条件付きチェックのため、指定されたオフセットを再シークしません。

Kafka バインダーを使用して、コンシューマーのメッセージを手動で確認したい。それ、どうやったら出来るの？

デフォルトでは、Kafka バインダーは Spring for Apache Kafka プロジェクトのデフォルトのコミット設定に委譲します。Spring Kafka のデフォルトの ackMode は batch です。詳細については、こちらを参照してください。

このデフォルトのコミット動作を無効にし、手動コミットに依存したい場合があります。次の手順でそれを行うことができます。

プロパティ spring.cloud.stream.kafka.bindings.<binding-name>.consumer.ackMode を MANUAL または MANUAL_IMMEDIATE のいずれかに設定します。このように設定すると、コンシューマーメソッドが受信するメッセージに kafka_acknowledgment （KafkaHeaders.ACKNOWLEDGMENT から）というヘッダーが存在します。

例: これをコンシューマー向けの方法として想像してください。

次に、プロパティ spring.cloud.stream.bindings.myConsumer-in-0.consumer.ackMode を MANUAL または MANUAL_IMMEDIATE に設定します。

Spring Cloud Stream は、関数の定義と署名に基づいてデフォルトのバインディングを作成しますが、これらをよりドメインに適した名前にオーバーライドするにはどうすればよいですか？

以下が関数のシグネチャーであると想定します。

デフォルトでは、Spring Cloud Stream は以下のようにバインディングを作成します。

次のプロパティを使用して、これらのバインディングを何かにオーバーライドできます。

この後、すべてのバインディングプロパティは、新しい名前 my-transformer-in および my-transformer-out で作成する必要があります。

これは、Kafka ストリームと複数の入力を使用した別の例です。

デフォルトでは、Spring Cloud Stream はこの関数に対して 3 つの異なるバインディング名を作成します。

これらのバインディングにいくつかの構成を設定するたびに、これらのバインディング名を使用する必要があります。それが好きではなく、たとえば、のような、よりドメインフレンドリーで読みやすいバインディング名を使用したいと考えています。

これらの 3 つのプロパティを設定するだけで、簡単にそれを行うことができます

これを行うと、デフォルトのバインディング名が上書きされ、それらに設定するプロパティは、これらの新しいバインディング名にある必要があります。

レコードのペイロードと一緒にキーを送信する必要がありますが、Spring Cloud Stream でそれを行う方法はありますか？

多くの場合、キーと値を含むレコードとして、マップのような連想データ構造を送信する必要があります。Spring Cloud Stream を使用すると、簡単な方法でそれを行うことができます。以下は、これを行うための基本的な青写真ですが、特定のユースケースに適合させることをお勧めします。

これがサンプルプロデューサーメソッド（別名 Supplier）です。

これは、String ペイロードだけでなく、キーも使用してメッセージを送信する簡単な関数です。KafkaHeaders.MESSAGE_KEY を使用してキーをメッセージヘッダーとして設定していることに注意してください。

キーをデフォルトの kafka_messageKey から変更する場合は、構成で次のプロパティを指定する必要があります。

バインディング名 supplier-out-0 を使用していることに注意してください。これは関数名であるため、それに応じて更新してください。

次に、メッセージを生成するときにこの新しいキーを使用します。

Spring Cloud Stream でメッセージコンバーターを使用する代わりに、Kafka でネイティブシリアライザーとデシリアライザーを使用したいと思います。デフォルトでは、Spring Cloud Stream は、内部の組み込みメッセージコンバーターを使用してこの変換を処理します。これをバイパスして、Kafka に責任を委譲するにはどうすればよいですか？

これは本当に簡単です。

ネイティブ直列化を有効にするには、次のプロパティを指定するだけです。

次に、serailzer も設定する必要があります。これを行うにはいくつかの方法があります。

またはバインダー構成を使用します。

バインダー方式を使用する場合、すべてのバインディングに対して適用されますが、バインディングでの設定はバインディングごとに行われます。

デシリアライズ側では、デシリアライザーを構成として提供する必要があります。

以下に例を示します。

バインダーレベルで設定することもできます。

ネイティブデコードを強制するために設定できるオプションのプロパティがあります。

ただし、Kafka バインダーの場合、これは不要です。バインダーに到達するまでに、Kafka は構成済みのデシリアライザーを使用してすでにデシリアライズしているためです。

デフォルトでは、Kafka Streams バインダーは、常に新しいコンシューマーの最も早いオフセットから開始します。場合によっては、最新のオフセットから開始することが有益であるか、アプリケーションによって要求されます。Kafka Streams バインダーを使用するとそれが可能になります。

ソリューションを検討する前に、次のシナリオを見てみましょう。

2 つの入力バインディングを必要とする BiConsumer Bean があります。この場合、最初のバインディングは KStream 用で、2 番目のバインディングは KTable 用です。このアプリケーションを初めて実行するとき、デフォルトでは、両方のバインディングは earliest オフセットから始まります。いくつかの要件のために latest オフセットから始めたいのはどうですか？ これを行うには、次のプロパティを有効にします。

1 つのバインディングのみを latest オフセットから開始し、もう 1 つをデフォルトの earliest からコンシューマーに開始する場合は、後者のバインディングを構成から除外します。

コミットオフセットが存在したら、これらの設定は光栄とコミットオフセットが優先されていない、ということを覚えておいてください。

Kafka プロデューサーアプリケーションを持っており、成功したすべてのセディングを追跡したいと思っています。

アプリケーションに次のサプライヤーがあると仮定します。

次に、成功したすべての送信情報をキャプチャーするために、新しい MessageChannel Bean を定義する必要があります。

次に、アプリケーション構成でこのプロパティを定義して、recordMetadataChannel の Bean 名を指定します。

この時点で、正常に送信された情報が fooRecordChannel に送信されます。

以下のように IntegrationFlow を記述して、情報を確認できます。

handle メソッドでは、ペイロードは Kafka に送信されたものであり、メッセージヘッダーには kafka_recordMetadata と呼ばれる特別なキーが含まれています。その値は、トピックパーティション、現在のオフセットなどに関する情報を含む RecordMetadata です。

いくつかのヘッダーを設定する Kafka プロデューサーアプリケーションがありますが、コンシューマーアプリケーションにありません。何故ですか？

通常の状況では、これで問題ありません。

次のプロデューサーがいると想像してみてください。

コンシューマー側では、ヘッダー "foo" が表示されるはずですが、以下では課題は発生しません。

アプリケーションでカスタムヘッダーマッパーを提供する場合、これは機能しません。アプリケーションに空の KafkaHeaderMapper があるとしましょう。

それが実装である場合、コンシューマーの foo ヘッダーを見逃すことになります。おそらく、これらの KafkaHeaderMapper メソッド内にいくつかのロジックがある可能性があります。foo ヘッダーにデータを入力するには、次のものが必要です。

これにより、プロデューサーからコンシューマーに foo ヘッダーが適切に入力されます。

Spring Cloud Stream では、id ヘッダーは特別なヘッダーですが、一部のアプリケーションでは、custom-id、ID、Id などの特別なカスタム ID ヘッダーが必要になる場合があります。最初のもの（custom-id）は、カスタムヘッダーマッパーなしでプロデューサーからコンシューマーに伝播します。ただし、フレームワークで予約されている id ヘッダーのバリアント（ID、Id、iD など）を使用して作成すると、フレームワークの内部で問題が発生します。このユースケースの詳細については、この StackOverflow スレッド (英語) を参照してください。その場合、大文字と小文字を区別する ID ヘッダーをマップするためにカスタム KafkaHeaderMapper を使用する必要があります。例: 次のプロデューサーがいるとします。

上記のヘッダー Id は、フレームワーク id ヘッダーと衝突するため、消費側から削除されます。この課題を解決するために、カスタム KafkaHeaderMapper を提供できます。

これにより、id ヘッダーと Id ヘッダーの両方がプロデューサー側からコンシューマー側で使用できるようになります。

複数の Kafka トピックへのトランザクションメッセージを生成するにはどうすればよいですか？

詳細については、この StackOverflow の質問 (英語) を参照してください。

トランザクションに Kafka バインダーのトランザクションサポートを使用してから、AfterRollbackProcessor を提供します。複数のトピックを作成するには、StreamBridge API を使用します。

以下は、このためのコードスニペットです。

テストするには、次を使用できます。

いくつかの重要な注意事項:

DLT を手動で構成するため、アプリケーション構成に DLQ 設定がないことを確認してください（デフォルトでは、初期のコンシューマー機能に基づいて input.DLT という名前のトピックに公開されます）。また、バインダーによる再試行を回避するために、1 へのコンシューマーバインディングで maxAttempts をリセットします。上記の例では、最大で合計 3 回試行されます（最初の試行 + FixedBackoff での 2 回の試行）。

このコードをテストする方法の詳細については、StackOverflow スレッド (英語) を参照してください。Spring Cloud Stream を使用して、コンシューマー関数を追加してテストする場合は、コンシューマーバインディングの isolation-level を read-committed に設定してください。

この StackOverflow スレッド (英語) もこの議論に関連しています。

ポーリング可能なコンシューマーの複数のインスタンスを実行し、インスタンスごとに一意の client.id を生成するにはどうすればよいですか？

次の定義を持っていると仮定します:

アプリケーションを実行すると、Kafka コンシューマーは client.id（consumer-my-group-1 など）を生成します。実行中のアプリケーションのインスタンスごとに、この client.id は同じになり、予期しない問題が発生します。

これを修正するために、アプリケーションの各インスタンスに次のプロパティを追加できます。

詳細については、この GitHub の課題 (英語) を参照してください。

Eclipse を使用する場合は、m2eclipeeclipse (英語) プラグインをお勧めします。m2eclipse をまだインストールしていない場合は、"eclipsemarketplace" から入手できます。

残念ながら、m2e はまだ Maven 3.3 をサポートしていないため、プロジェクトを Eclipse にインポートしたら、プロジェクトに .settings.xml ファイルを使用するように m2eclipse に指示する必要もあります。これを行わないと、プロジェクトの POM に関連するさまざまなエラーが表示される場合があります。Eclipse 設定を開き、Maven 設定を展開して、ユーザー設定を選択します。ユーザー設定フィールドで、参照をクリックし、インポートした Spring Cloud プロジェクトに移動して、そのプロジェクトの .settings.xml ファイルを選択します。適用、OK の順にクリックして、設定の変更を保存します。

m2eclipse を使用したくない場合は、次のコマンドを使用して Eclipse プロジェクトのメタデータを生成できます。

生成された日食プロジェクトは、file メニューから import existing projects を選択することでインポートできます。

[[ 貢献 ] == 貢献

Spring Cloud は、制限のない Apache 2.0 ライセンスでリリースされており、課題には Github トラッカーを使用し、プルリクエストをマスターにマージする、非常に標準的な Github 開発プロセスに従っています。些細なことでも貢献したい場合は、遠慮なく以下のガイドラインに従ってください。