次のプロパティは 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 つのユースケースです。

トピックオフセットの詳細な制御については、KafkaRebalanceListener を使用するを参照してください。リスナーが提供されている場合、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. を付ける必要があります。

ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)

失敗した送信は、プロデューサーエラーチャネルに送信されます（構成されている場合）。エラーチャネルを参照してください。デフォルト: null

+

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

デフォルトでは、レコードは元のレコードと同じパーティションを使用して 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 番目のバインディングで logAndSkip を有効にしたい場合は、以下のようにコンシューマーで実行できます。

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

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

デフォルトでは、レコードは元のレコードと同じパーティションを使用して 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. のプレフィックスを付ける必要があります

ストリーム構成に含まれる可能性のあるすべてのプロパティの詳細については、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. を使用して構成できます。

streamPartitionerBeanName: コンシューマーで使用されるカスタム送信パーティショナー Bean 名。アプリケーションは、カスタム StreamPartitioner を Spring Bean として提供でき、この Bean の名前を、デフォルトの名前の代わりに使用するためにプロデューサーに提供できます。

+ デフォルト: 送信パーティションのサポートについては、上記の説明を参照してください。

次のプロパティは 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 によって設定されたデフォルトを使用します。

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 開発プロセスに従っています。些細なことでも貢献したい場合は、遠慮なく以下のガイドラインに従ってください。