非同期 @KafkaListener 戻り型

バージョン 3.2 以降では、@KafkaListener (および @KafkaHandler) メソッドに非同期戻り値の型を指定して、応答を非同期に送信できるようになりました。戻り値の型には、CompletableFuture<?>、Mono<?>、Kotlin suspend 関数が含まれます。

@KafkaListener(id = "myListener", topics = "myTopic")
public CompletableFuture<String> listen(String data) {
    ...
    CompletableFuture<String> future = new CompletableFuture<>();
    future.complete("done");
    return future;
}
@KafkaListener(id = "myListener", topics = "myTopic")
public Mono<Void> listen(String data) {
    ...
    return Mono.empty();
}
AckMode は自動的に MANUAL を設定し、非同期戻り値の型が検出されるとアウトオブオーダーコミットを有効にします。代わりに、非同期操作が完了すると、非同期完了が確認応答されます。非同期結果がエラーで完了した場合、メッセージが回復されるかどうかはコンテナーのエラーハンドラーによって異なります。非同期結果オブジェクトの作成を妨げる何らかの例外がリスナーメソッド内で発生した場合は、その例外をキャッチし、メッセージを確認応答または回復させる適切な戻りオブジェクトを返さなければなりません。

KafkaListenerErrorHandler が非同期戻り値の型 (Kotlin サスペンド関数を含む) を持つリスナー上で構成されている場合、エラーハンドラーは失敗後に呼び出されます。このエラーハンドラーとその目的の詳細については、"例外の処理" を参照してください。

DefaultErrorHandler のような処理後シークを行うエラーハンドラーを使用すると、非同期リスナーは、ブロッキングリスナーと比較して、同じパーティション上の失敗したレコードごとにディスパッチループの配信を 1 回多く受け取ります。ディスパッチループは、非同期の失敗コールバックが実行される前に、ポーリングされたすべてのレコードでリスナーを呼び出すため、ポイズンピルバーストの最初のポーリングでは、パーティションに再試行状態が登録される前に、すべてのレコードが 1 回ずつ呼び出されます。その後、先頭レコードの再試行サイクルでは、その背後にキューイングされたレコードがスキップされるため、呼び出しの総数は、バックログの深さに応じて増加するのではなく、バーストサイズに比例して増加します。

maxAttempts = n を持つ単一パーティション上の常に失敗する N レコードの場合、サスペンドリスナーは合計 N * (n + 2) - 1 回の呼び出しを受け取りますが、ブロッキングリスナーは N * (n + 1) 回の呼び出しを受け取ります。非同期リスナーのポイントであるポーリングごとの並列処理を放棄しない限り、イテレータ 0 のディスパッチ呼び出しを省略することはできません。同期スローによって後続のレコードが呼び出される前にバッチ反復処理が中止されるため、これはブロッキングリスナーには影響しません。

リスナーが配信ごとに非冪等な処理（送信 HTTP 呼び出し、オフセットコミットを伴うトランザクションではない DB 書き込みなど）を実行する場合は、以下のいずれかを選択してください。

  • @RetryableTopic （ノンブロッキング再試行）を使用します。失敗したレコードは別の再試行トピックにルーティングされるため、メインパーティションは処理を継続し、レコードが余分なディスパッチループ配信を受けることはありません。

  • リスナーを冪等にします（Kafka のデフォルトの少なくとも 1 回配信はすでにこれを想定している）。

この選択は、アプリケーション全体ではなく、トピックごとに行うことができます。パーティションごとの厳密な順序付けが正当性を保証するイベント（エンティティ ID をキーとする状態遷移など）は、シークベースのハンドラー上で実行され、スループットはパーティション数とキーの分散に依存します。順序付けを必要としないイベント（検索インデックスの更新、通知、ダウンストリームのエンリッチメントなど）は、@RetryableTopic リスナー上で実行できるため、失敗したレコードに余分なディスパッチループ配信が行われることはありません。

非同期戻り値型では、同じパーティション上のレコードで結果が混在する場合、パーティションごとの処理順序が保持されません。失敗したレコードが再試行 / バックオフサイクル中である間に、同じパーティション上の後段オフセットの成功したレコードが先にリスナー本体を完了する可能性があります。Kafka は引き続き順序通りに配信し、コミットされたオフセットは一貫性を保ちますが、リスナー本体は並行して実行されるため、副作用がパーティションの順序から外れる可能性があります。ブロッキングリスナーは、パーティションを厳密に 1 つのレコードずつ処理するため、この問題を回避します。パーティション上の次のレコードが呼び出される前に、前段オフセットのレコードが完了 (成功または回復) します。コンシューマーがキーごとの処理順序 (エンティティ ID をキーとする冪等更新、順序付き状態遷移) に依存している場合は、どのエラーハンドラーを使用するかに関わらず、ブロッキングリスナーが適切なモデルです。