パッケージ org.springframework.kafka.listener

クラス DefaultErrorHandler

java.lang.ObjectSE
org.springframework.kafka.listener.KafkaExceptionLogLevelAware
org.springframework.kafka.listener.ExceptionClassifier
org.springframework.kafka.listener.FailedRecordProcessor
org.springframework.kafka.listener.FailedBatchProcessor
org.springframework.kafka.listener.DefaultErrorHandler
実装されたすべてのインターフェース:
CommonErrorHandler, DeliveryAttemptAware
public class DefaultErrorHandler extends FailedBatchProcessor implements CommonErrorHandler
レコードリスナーの場合、残りのレコード内の各トピックの現在のオフセットをシークするエラーハンドラー。メッセージの失敗後に、再生できるようにパーティションを巻き戻すために使用されます。バッチリスナーの場合、レコードのバッチ内の各トピックの現在のオフセットをシークします。バッチを再生できるように、メッセージの失敗後にパーティションを巻き戻すために使用されます。リスナーが失敗したレコードとともに BatchListenerFailedException をスローした場合。レコードの前のレコードのオフセットがコミットされ、残りのレコードのパーティションが再配置されるか、失敗したレコードを回復してスキップできます。他の例外がスローされた場合、または例外で有効なレコードが提供されなかった場合、エラー処理はこのハンドラーの BackOff を使用して FallbackBatchErrorHandler に委譲されます。レコードが復元されると、そのオフセットがコミットされます。これは、従来の SeekToCurrentErrorHandler および SeekToCurrentBatchErrorHandler に代わるものです (ただし、フォールバックでは、無期限に再試行するのではなく、再試行が完了した後にメッセージを回復プログラムに送信できるようになりました)。
導入:
2.8
作成者:
Gary Russell

  • コンストラクターの詳細

    • DefaultErrorHandler

      public DefaultErrorHandler()
      デフォルトのバックオフ (9 回の再試行、遅延なし) で、トピック / パーティション / オフセットに対して 10 (maxFailures) が発生した後にレコードをログに記録するだけのデフォルトのリカバリを使用してインスタンスを構築します。

    • DefaultErrorHandler

      public DefaultErrorHandler(BackOff backOff)
      backOff がトピック / パーティション / オフセットに対して STOP を返した後にレコードをログに記録するだけのデフォルトのリカバリを使用してインスタンスを構築します。
      パラメーター:
      backOff - BackOff

    • DefaultErrorHandler

      public DefaultErrorHandler(ConsumerRecordRecoverer recoverer)
      トピック / パーティション / オフセットに対して 10 (maxFailures) が発生した後に呼び出される、提供されたリカバリを使用してインスタンスを構築します。
      パラメーター:
      recoverer - 回復者。

    • DefaultErrorHandler

      public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff)
      backOff がトピック / パーティション / オフセットに対して STOP を返した後に呼び出される、提供されたリカバリを使用してインスタンスを構築します。
      パラメーター:
      recoverer - 回復者 ; null の場合、デフォルト (ロギング) リカバリが使用されます。
      backOff - BackOff

    • DefaultErrorHandler

      public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff, @Nullable BackOffHandler backOffHandler)
      backOff がトピック / パーティション / オフセットに対して STOP を返した後に呼び出される、提供されたリカバリを使用してインスタンスを構築します。
      パラメーター:
      recoverer - 回復者 ; null の場合、デフォルト (ロギング) リカバリが使用されます。
      backOff - BackOff
      backOffHandler - BackOffHandler
      導入:
      2.9

  • メソッドの詳細

    • setCommitRecovered

      public void setCommitRecovered(boolean commitRecovered)
      復元されたレコードのオフセットをコミットするには、true に設定します。コンテナーは ContainerProperties.AckMode.MANUAL_IMMEDIATE で構成されている必要があります。コミットが同期か非同期かは、コンテナーの syncCommits プロパティによって決まります。
      オーバーライド:
      クラス FailedRecordProcessor の setCommitRecovered 
      パラメーター:
      commitRecovered - コミットする場合は true。

    • isAckAfterHandle

      public boolean isAckAfterHandle()
      インターフェースからコピーされた説明: CommonErrorHandler
      処理されたエラーに対してオフセットをコミットする必要がある場合は true を返します（例外はスローされません）。
      次で指定:
      インターフェース CommonErrorHandler の isAckAfterHandle 
      戻り値:
      コミットする場合は true。

    • setAckAfterHandle

      public void setAckAfterHandle(boolean ackAfterHandle)
      インターフェースからコピーされた説明: CommonErrorHandler
      コンテナーがリカバリされたレコードのオフセットをコミットしないようにするには、false に設定します（エラーハンドラー自体が例外をスローしない場合）。
      次で指定:
      インターフェース CommonErrorHandler の setAckAfterHandle 
      パラメーター:
      ackAfterHandle - コミットしない場合は false。

    • seeksAfterHandling

      public boolean seeksAfterHandling()
      インターフェースからコピーされた説明: CommonErrorHandler
      このエラーハンドラーが失敗したレコードと残りのレコード（または失敗したレコードが回復した後の残りのレコードのみ）に対してシークを実行する場合は true を返します。
      次で指定:
      インターフェース CommonErrorHandler の seeksAfterHandling 
      戻り値:
      次のポーリングでレコードをフェッチする必要がある場合は true。

    • deliveryAttemptHeader

      public boolean deliveryAttemptHeader()
      インターフェースからコピーされた説明: CommonErrorHandler
      このエラーハンドラーが配信試行ヘッダーをサポートしている場合は true を返します。
      次で指定:
      インターフェース CommonErrorHandler の deliveryAttemptHeader 
      戻り値:
      可能であれば true。

    • handleOne

      public boolean handleOne(ExceptionSE thrownException, org.apache.kafka.clients.consumer.ConsumerRecord<?,?> record, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container)
      インターフェースからコピーされた説明: CommonErrorHandler
      CommonErrorHandler.seeksAfterHandling() が false を返した場合のレコードリスナーの例外を処理します。これを使用して、失敗した単一のレコードのみを処理します。
      次で指定:
      インターフェース CommonErrorHandler の handleOne 
      パラメーター:
      thrownException - 例外。
      record - レコード。
      consumer - コンシューマー。
      container - コンテナー。
      戻り値:
      エラーが「処理」された場合は true、そうでない場合は false で、コンテナーはレコードをリスナーに再送信します。
      関連事項:

    • handleRemaining

      public void handleRemaining(ExceptionSE thrownException, ListSE<org.apache.kafka.clients.consumer.ConsumerRecord<?,?>> records, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container)
      インターフェースからコピーされた説明: CommonErrorHandler
      CommonErrorHandler.seeksAfterHandling() が true を返したときに、レコードリスナーの例外を処理します。失敗したレコードとポーリングの残りのすべてのレコードが渡されます。通常、エラーハンドラーがシークを実行して、残りのレコードが次のポーリングで再配信されるようにするときに使用されます。
      次で指定:
      インターフェース CommonErrorHandler の handleRemaining 
      パラメーター:
      thrownException - 例外。
      records - 失敗したものを含む残りのレコード。
      consumer - コンシューマー。
      container - コンテナー。
      関連事項:

    • handleBatch

      public void handleBatch(ExceptionSE thrownException, org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container, RunnableSE invokeListener)
      インターフェースからコピーされた説明: CommonErrorHandler
      バッチリスナーの例外を処理します。ポーリングからの完全な ConsumerRecords が提供されます。バッチ内のレコードを再処理する場合は、エラーハンドラーがシークを実行する必要があります。
      次で指定:
      インターフェース CommonErrorHandler の handleBatch 
      パラメーター:
      thrownException - 例外。
      data - コンシューマーの記録。
      consumer - コンシューマー。
      container - コンテナー。
      invokeListener - リスナーを再起動するためのコールバック。

    • handleBatchAndReturnRemaining

      public <K, V> org.apache.kafka.clients.consumer.ConsumerRecords<K,V> handleBatchAndReturnRemaining(ExceptionSE thrownException, org.apache.kafka.clients.consumer.ConsumerRecords<?,?> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container, RunnableSE invokeListener)
      インターフェースからコピーされた説明: CommonErrorHandler
      バッチリスナーの例外を処理します。ポーリングからの完全な ConsumerRecords が提供されます。リスナーに再送信する必要があるバッチのメンバーを返します。返されるレコードは、元のレコードと同じ順序である必要があります。
      次で指定:
      インターフェース CommonErrorHandler の handleBatchAndReturnRemaining 
      型パラメーター:
      K - 鍵の型。
      V - 値の型。
      パラメーター:
      thrownException - 例外。
      data - コンシューマーの記録。
      consumer - コンシューマー。
      container - コンテナー。
      invokeListener - リスナーを再起動するためのコールバック。
      戻り値:
      コンシューマーレコード、またはサブセット。

    • handleOtherException

      public void handleOtherException(ExceptionSE thrownException, org.apache.kafka.clients.consumer.Consumer<?,?> consumer, MessageListenerContainer container, boolean batchListener)
      インターフェースからコピーされた説明: CommonErrorHandler
      利用可能なレコードがない状態で例外がスローされた場合に呼び出されます。コンシューマー投票が例外をスローした場合。
      次で指定:
      インターフェース CommonErrorHandler の handleOtherException 
      パラメーター:
      thrownException - 例外。
      consumer - コンシューマー。
      container - コンテナー。
      batchListener - リスナーがバッチリスナーの場合は true。

    • onPartitionsAssigned

      public void onPartitionsAssigned(org.apache.kafka.clients.consumer.Consumer<?,?> consumer, CollectionSE<org.apache.kafka.common.TopicPartition> partitions, RunnableSE publishPause)
      インターフェースからコピーされた説明: CommonErrorHandler
      パーティションが割り当てられたときに呼び出されます。
      次で指定:
      インターフェース CommonErrorHandler の onPartitionsAssigned 
      パラメーター:
      consumer - コンシューマー。
      partitions - 新しく割り当てられたパーティション。
      publishPause - コンシューマーの一時停止イベントを発行するために呼び出されます。