パッケージ org.springframework.integration.kafka.inbound

クラス KafkaInboundGateway<K,V,R>

java.lang.ObjectSE
org.springframework.integration.context.IntegrationObjectSupport
org.springframework.integration.endpoint.AbstractEndpoint
org.springframework.integration.gateway.MessagingGatewaySupport
org.springframework.integration.kafka.inbound.KafkaInboundGateway<K,V,R>
型パラメーター:
K - 鍵の型。
V - リクエスト値型。
R - 応答値の型。
実装されたすべてのインターフェース:
Aware、BeanFactoryAware、BeanNameAware、DisposableBean、InitializingBean、ApplicationContextAware、Lifecycle、Phased、SmartLifecycle、ExpressionCapable、OrderlyShutdownCapable、Pausable、IntegrationPattern、KafkaInboundEndpoint、NamedComponent、IntegrationInboundManagement、IntegrationManagement、ManageableLifecycle、ManageableSmartLifecycle、TrackableComponent
public class KafkaInboundGateway<K,V,R> extends MessagingGatewaySupport implements KafkaInboundEndpoint, Pausable, OrderlyShutdownCapable
受信ゲートウェイ。
導入:
5.4
作成者:
Gary Russell, Artem Bilan, Urs Keller

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

    • KafkaInboundGateway

      public KafkaInboundGateway(org.springframework.kafka.listener.AbstractMessageListenerContainer<K,V> messageListenerContainer, org.springframework.kafka.core.KafkaTemplate<K,R> kafkaTemplate)
      提供されたコンテナーを使用してインスタンスを構築します。
      パラメーター:
      messageListenerContainer - コンテナー。
      kafkaTemplate - kafka テンプレート。

  • メソッドの詳細

    • setMessageConverter

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

    • setPayloadType

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

    • setRetryTemplate

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

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

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

    • setRecoveryCallback

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

    • 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(java.util.Map<org.apache.kafka.common.TopicPartition, java.lang.Long>, org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback)

    • setBindSourceRecord

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

    • onInit

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

    • doStart

      protected void doStart()
      クラスからコピーされた説明: AbstractEndpoint
      サブクラスは、開始動作でこのメソッドを実装する必要があります。このメソッドは、AbstractEndpoint.lifecycleLock を保持しているときに呼び出されます。
      オーバーライド:
      クラス MessagingGatewaySupport の doStart 

    • doStop

      protected void doStop()
      クラスからコピーされた説明: AbstractEndpoint
      サブクラスは、停止動作でこのメソッドを実装する必要があります。このメソッドは、AbstractEndpoint.lifecycleLock を保持しているときに呼び出されます。
      オーバーライド:
      クラス MessagingGatewaySupport の doStop 

    • pause

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

    • resume

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

    • isPaused

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

    • getComponentType

      public StringSE getComponentType()
      クラスからコピーされた説明: IntegrationObjectSupport
      サブクラスはこのメソッドを実装して、コンポーネント型情報を提供できます。
      次で指定:
      インターフェース NamedComponent の getComponentType 
      オーバーライド:
      クラス MessagingGatewaySupport の getComponentType 

    • beforeShutdown

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

    • afterShutdown

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

    • getErrorMessageAttributes

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