クラス KafkaMessageDrivenChannelAdapter<K,V>

java.lang.ObjectSE
org.springframework.integration.context.IntegrationObjectSupport
org.springframework.integration.endpoint.AbstractEndpoint
org.springframework.integration.endpoint.MessageProducerSupport
org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter<K,V>
型パラメーター:
K - 鍵の型。
V - 値の型。
実装済みのインターフェース一覧:
Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, SmartInitializingSingleton, ApplicationContextAware, Lifecycle, Phased, SmartLifecycle, ComponentSourceAware, ExpressionCapable, OrderlyShutdownCapable, MessageProducer, Pausable, IntegrationPattern, KafkaInboundEndpoint, NamedComponent, IntegrationInboundManagement, IntegrationManagement, ManageableLifecycle, ManageableSmartLifecycle, TrackableComponent
public class KafkaMessageDrivenChannelAdapter<K,V> extends MessageProducerSupport implements KafkaInboundEndpoint, OrderlyShutdownCapable, Pausable
メッセージ駆動型チャネルアダプター。
導入:
5.4
作成者:
Marius Bogoevici, Gary Russell, Artem Bilan, Urs Keller, Jooyoung Pyoung, Oleksii Zernov

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

    • KafkaMessageDrivenChannelAdapter

      public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer)
      モード KafkaMessageDrivenChannelAdapter.ListenerMode.record でインスタンスを作成します。
      パラメーター:
      messageListenerContainer - コンテナー。

    • KafkaMessageDrivenChannelAdapter

      public KafkaMessageDrivenChannelAdapter(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode mode)
      提供されたモードでインスタンスを構築します。
      パラメーター:
      messageListenerContainer - コンテナー。
      mode - モード。

  • 方法の詳細

    • setMessageConverter

      public void setMessageConverter(org.springframework.kafka.support.converter.MessageConverter messageConverter)
      メッセージコンバーターを設定します。モードに応じて、RecordMessageConverter または BatchMessageConverter でなければなりません。
      パラメーター:
      messageConverter - コンバーター。

    • setRecordMessageConverter

      public void setRecordMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
      レコードコンバーターを使用するようにメッセージコンバーターを設定します。
      パラメーター:
      messageConverter - コンバーター。

    • setBatchMessageConverter

      public void setBatchMessageConverter(org.springframework.kafka.support.converter.BatchMessageConverter messageConverter)
      メッセージコンバーターを設定して、バッチベースのコンシューマーで使用します。
      パラメーター:
      messageConverter - コンバーター。

    • setRecordFilterStrategy

      public void setRecordFilterStrategy(org.springframework.kafka.listener.adapter.RecordFilterStrategy<K,V> recordFilterStrategy)
      KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener を FilteringMessageListenerAdapter にラップする RecordFilterStrategy を指定します。
      パラメーター:
      recordFilterStrategy - 使用する RecordFilterStrategy。

    • setAckDiscarded

      public void setAckDiscarded(boolean ackDiscarded)
      FilteringMessageListenerAdapter が破棄されたレコードを確認する必要があるかどうかを示す boolean フラグ。setRecordFilterStrategy(RecordFilterStrategy) が指定されていない場合は意味がありません。
      パラメーター:
      ackDiscarded - 破棄されたメッセージに確認応答（コミットオフセット）する場合は true。

    • setRetryTemplate

      public void setRetryTemplate(RetryTemplate retryTemplate)
      配信の再試行に使用する RetryTemplate インスタンスを指定します。

      IMPORTANT: この形式の再試行はブロックされており、ポーリングされたすべてのレコードの合計再試行遅延が max.poll.interval.ms を超える可能性がある場合、再調整が発生する可能性があります。代わりに、KafkaErrorSendingMessageRecoverer で構成されたリスナーコンテナーに DefaultErrorHandler を追加することを検討してください。

      パラメーター:
      retryTemplate - 使用する RetryTemplate。

    • setRecoveryCallback

      public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback)
      再試行操作用の RecoveryCallback インスタンス。null の場合、再試行回数の上限に達した後にコンテナーに例外がスローされます（エラーチャネルが設定されていない場合）。setRetryTemplate(RetryTemplate) が指定されている場合にのみ使用されます。エラーチャネルが指定されている場合、デフォルトは ErrorMessageSendingRecoverer です。再試行回数の上限に達した後にコンテナーに例外をスローする場合は、null に設定してください。
      パラメーター:
      recoveryCallback - リカバリコールバック。

    • setFilterInRetry

      public void setFilterInRetry(boolean filterInRetry)
      boolean フラグは、フィルター処理と再試行処理の実行順序を指定します。RetryTemplate または RecordFilterStrategy のいずれか一方のみ、またはどちらも指定されていない場合は、このフラグは意味を持ちません。true の場合、フィルターは再試行ごとに呼び出されます。false の場合、フィルターはコンテナーからの配信ごとに 1 回だけ呼び出されます。
      パラメーター:
      filterInRetry - 再試行ごとにフィルター処理する場合は true。デフォルトは false です。

    • setPayloadType

      public void setPayloadType(ClassSE<?> payloadType)
      StringJsonMessageConverter などの型対応メッセージコンバーターを使用する場合は、コンバーターが作成するペイロード型を設定します。デフォルトは ObjectSE です。
      パラメーター:
      payloadType - 型。

    • setOnPartitionsAssignedSeekCallback

      public void setOnPartitionsAssignedSeekCallback(BiConsumerSE<MapSE<org.apache.kafka.common.TopicPartition, LongSE>, org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback)
      KafkaMessageListenerContainer からの ConsumerSeekAware.onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback) 呼び出し中のシーク管理用に BiConsumerSE を指定します。これは、内部 MessagingMessageListenerAdapter 実装から呼び出されます。
      パラメーター:
      onPartitionsAssignedCallback - 使用する BiConsumerSE
      関連事項:
      • ConsumerSeekAware.onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)

    • setBindSourceRecord

      public void setBindSourceRecord(boolean bindSourceRecord)
      true に設定すると、IntegrationMessageHeaderAccessor.SOURCE_DATA という名前のヘッダーにソースコンシューマーレコードがバインドされます。バッチリスナーには適用されません。
      パラメーター:
      bindSourceRecord - バインドする場合は true。

    • getComponentType

      public StringSE getComponentType()
      次で指定:
      インターフェース NamedComponent 内の getComponentType 
      オーバーライド:
      クラス MessageProducerSupport の getComponentType 

    • onInit

      protected void onInit()
      クラスからコピーされた説明: IntegrationObjectSupport
      サブクラスは、初期化ロジック用にこれを実装できます。
      オーバーライド:
      クラス MessageProducerSupport の onInit 

    • doStart

      protected void doStart()
      クラスからコピーされた説明: MessageProducerSupport
      デフォルトでは何も実行されません。ライフサイクル管理された動作が必要な場合、サブクラスはこれをオーバーライドできます。'lifecycleLock' によって保護されています。
      オーバーライド:
      クラス MessageProducerSupport の doStart 

    • doStop

      protected void doStop()
      クラスからコピーされた説明: MessageProducerSupport
      デフォルトでは何もしません。サブクラスは、ライフサイクル管理の動作が必要な場合、これをオーバーライドできます。
      オーバーライド:
      クラス MessageProducerSupport の doStop 

    • pause

      public void pause()
      インターフェースからコピーされた説明: Pausable
      エンドポイントを一時停止します。
      次で指定:
      インターフェース Pausable 内の pause 

    • resume

      public void resume()
      インターフェースからコピーされた説明: Pausable
      一時停止した場合は、エンドポイントを再開します。
      次で指定:
      インターフェース Pausable 内の resume 

    • isPaused

      public boolean isPaused()
      インターフェースからコピーされた説明: Pausable
      エンドポイントが一時停止しているかどうかを確認します。
      次で指定:
      インターフェース Pausable 内の isPaused 
      戻り値:
      一時停止した場合は true。

    • beforeShutdown

      public int beforeShutdown()
      インターフェースからコピーされた説明: OrderlyShutdownCapable
      シャットダウンが始まる前に呼び出されます。実装は、新しいメッセージの受け入れを停止する必要があります。オプションで、処理中のアクティブなメッセージの数を返すことができます。
      次で指定:
      インターフェース OrderlyShutdownCapable 内の beforeShutdown 
      戻り値:
      使用可能な場合、アクティブなメッセージの数。

    • afterShutdown

      public int afterShutdown()
      インターフェースからコピーされた説明: OrderlyShutdownCapable
      スケジューラー、エグゼキューターなどの通常のシャットダウンの後、シャットダウン遅延が経過した後、残りのアクティブなスケジューラー / エグゼキュータースレッドの強制シャットダウンの前に呼び出されます。オプションで、まだ処理中のアクティブなメッセージの数を返すことができます。
      次で指定:
      インターフェース OrderlyShutdownCapable 内の afterShutdown 
      戻り値:
      使用可能な場合、アクティブなメッセージの数。

    • getErrorMessageAttributes

      protected AttributeAccessor getErrorMessageAttributes(@Nullable Message<?> message)
      クラスからコピーされた説明: MessageProducerSupport
      errorMessageStrategy でエラーメッセージを作成するときに使用する AttributeAccessor を設定します。
      オーバーライド:
      クラス MessageProducerSupport の getErrorMessageAttributes 
      パラメーター:
      message - メッセージ。
      戻り値:
      属性。