新着情報

4.0 以降の 4.1 の新機能

このセクションでは、バージョン 4.0 からバージョン 4.1 への変更点について説明します。以前のバージョンの変更点については、変更履歴を参照してください。

@KafkaListener の変更

@KafkaListener アノテーションは、ackMode 属性をサポートするようになりました。これにより、個別のリスナーは、個別のコンテナーファクトリ Bean を作成することなく、コンテナーファクトリのデフォルトの確認モードをオーバーライドできます。この属性は、SpEL 式とプロパティプレースホルダーもサポートしています。詳細については、@KafkaListener アノテーションを参照してください。

コンシューマーの認知度を高める方法

ContainerProperties のブール型プロパティ setExplicitShareAcknowledgment(boolean) は、ShareAckMode 列挙型 (EXPLICIT、MANUAL、IMPLICIT) に置き換えられました。非推奨のメソッドは引き続き機能します。true は MANUAL に、false は EXPLICIT にマッピングされます。デフォルトの動作は変更されていません。デフォルトを使用するアプリケーションは移行する必要はありません。モードの説明については記録確認を、アップグレードの詳細については 4.0 からの移行を参照してください。

コンシューマーの非同期コミットを共有する

ContainerProperties の新しいプロパティ syncShareCommits は、共有コンテナーが確認応答コミットに commitSync() (デフォルト) または commitAsync() のどちらを使用するかを制御します。確認応答の耐久性に若干の遅延が許容される場合は、スループットを向上させるために false に設定してください。詳細は同期コミットと非同期コミットを参照してください。

コンシューマーエラー処理を共有する

共有コンシューマーコンテナーに、設定可能なエラー処理機能が追加されました。

  • 世論調査レベル : poll() からの RecordDeserializationException と CorruptRecordException が捕捉されるため、コンシューマースレッドは続行されます。逆直列化できないレコードは拒否され、次のポーリングが実行されます。

  • リスナーレベル : ShareConsumerRecordRecoverer インターフェースは、リスナーが例外をスローした際に、ACCEPT、RELEASE、REJECT のいずれを実行するかを決定します。デフォルトは ShareConsumerRecordRecoverer.REJECTING ですが、RELEASING も利用可能です。ファクトリまたはコンテナーにカスタムリカバリを設定することもできます。

  • Kafka キューのドキュメントにあるコンシューマーのエラー処理を共有するを参照してください。

コンシューマーライフサイクルイベントを共有する

ShareKafkaMessageListenerContainer は、ConsumerStartingEvent および ConsumerStartedEvent に加えて、停止および障害ライフサイクルイベントを公開するようになりました。

  • ShareConsumerStoppingEvent (新規): 基盤となる ShareConsumer が閉じられる前に公開されたため、新しいイベントクラスが導入されました。ShareConsumer は Consumer とは別のクライアント API であるため、既存の ConsumerStoppingEvent は再利用できません。

  • ConsumerStoppedEvent: 株式コンシューマーが閉鎖された後に公開され、reason は NORMAL、ABNORMAL、ERROR です。

  • ConsumerFailedToStartEvent: 共有コンシューマーの起動に失敗した場合に公開されます。同じ起動試行ですでに構築されたコンシューマーは、静かに閉じられます。

詳細はライフサイクルイベントを参照してください。

ShareAcknowledgment.renew()

ShareAcknowledgment は、処理時間がブローカーのロック期間を超えた場合に、renew() が取得ロックを延長する機能をサポートするようになりました（KIP-1222、Kafka 4.2）。詳細は、Kafka キューのドキュメントにある ShareAcknowledgment API を参照してください。

RetryTopicConfigurationBuilder デフォルト戦略変更

RetryTopicConfigurationBuilder における sameIntervalTopicReuseStrategy のデフォルト値が、@RetryableTopic アノテーションのデフォルト値に合わせるため、MULTIPLE_TOPICS から SINGLE_TOPIC に変更されました。詳細については、トピックの命名を参照してください。

Kafka ストリームのネイティブ DLQ サポート

Spring for Apache Kafka は、Kafka Streams DLQ (KIP-1034、Kafka 4.2) をサポートする例外ハンドラーを提供するようになりました。

提供される例外ハンドラーは、複数の Kafka Streams DLQ 有効化戦略とデッドレタートピック解決オプションをサポートしています。

新しいハンドラープロパティとの整合性を図るため、RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER は非推奨となり、RecoveringDeserializationExceptionHandler.RECOVERER が推奨されるようになりました。

Kafka ストリームグループプロトコル構成

StreamsBuilderFactoryBean は、groupProtocol プロパティの設定をサポートするようになりました。これにより、基盤となる Kafka Streams コンシューマーが classic プロトコルに依存するか、より新しい streams グループプロトコル (KIP-1071: Streams のサーバー側リバランス) を選択するかを明示的に管理できます。

詳細については、Kafka Streams のドキュメントにある構成を参照してください。

FailedBatchProcessor における BackOffFunction のサポート

FailedBatchProcessor は、バッチメッセージ処理において setBackOffFunction および setResetStateOnExceptionChange をサポートするようになりました。これは、FailedRecordProcessor がすでにシングルメッセージ処理でサポートしているのと同様です。これにより、DefaultErrorHandler を使用する際に、シングル処理とバッチ処理の両方で一貫した例外分類が可能になります。

後方互換性を維持するため、FailedBatchProcessor では resetStateOnExceptionChange のデフォルト値は false ですが、FailedRecordProcessor では true がデフォルト値となっています。両方のインスタンスで一貫した動作が必要な場合は、DefaultErrorHandler インスタンスで setResetStateOnExceptionChange(true) を呼び出してください。