KafkaMessageSource (Spring Integration API)

java.lang.ObjectSE
- org.springframework.integration.util.AbstractExpressionEvaluator
- - org.springframework.integration.endpoint.AbstractMessageSource<ObjectSE>
  - - org.springframework.integration.kafka.inbound.KafkaMessageSource<K,V>

型パラメーター:
K - 鍵の型。
V - 値の型。
実装されているすべてのインターフェース:
Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, Lifecycle, MessageSource<ObjectSE>, Pausable, IntegrationPattern, NamedComponent, IntegrationInboundManagement, IntegrationManagement, ManageableLifecycle
```
public class KafkaMessageSource<K,V>
extends AbstractMessageSource<ObjectSE>
implements Pausable
```
kafka のポーリングされたメッセージソース。一度に 1 つのスレッドのみがデータをポーリング（またはメッセージを確認）できます。
NOTE: アプリケーションがメッセージの順序が正しくないことを確認した場合、オフセットの前のすべてのメッセージが確認応答されるまで、確認応答は延期されます。複数のレコードが取得され、以前のオフセットが再キューイングされた場合、後続のオフセットからのレコードは、正常に処理された場合でも再配信されます。アプリケーションはべき等性を実装する必要があります。
バージョン 3.1.2 以降、このソースは Pausable を実装しており、Consumer を一時停止および再開できます。コンシューマーが一時停止している間、リバランスを防ぐために、max.poll.interval.ms 内で AbstractMessageSource.receive() を呼び出し続ける必要があります。
導入:
5.4
作成者:
Gary Russell, Mark Norkin, Artem Bilan, Anshul Mehra

ネストされたクラスの要約

ネストされたクラス
修飾子と型	クラスと説明
`static class`	`KafkaMessageSource.KafkaAckCallback<K, V>` Kafka の場合は AcknowledgmentCallback。
`static class`	`KafkaMessageSource.KafkaAckCallbackFactory<K, V>` KafkaAckInfo の場合は AcknowledgmentCallbackFactory。
`static interface`	`KafkaMessageSource.KafkaAckInfo<K, V>` KafkaAckCallback を構築するための情報。
`class`	`KafkaMessageSource.KafkaAckInfoImpl` KafkaAckCallback を構築するための情報。

インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたネストクラス / インターフェース
IntegrationManagement.ManagementOverrides

フィールドのサマリー
フィールド
修飾子と型フィールドと説明
static StringSE REMAINING_RECORDS
前回のポーリングから残っているレコードの数。
- クラス org.springframework.integration.util.AbstractExpressionEvaluator から継承されたフィールド
  EXPRESSION_PARSER, logger
- インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたフィールド
  METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

フィールド
修飾子と型	フィールドと説明
`static StringSE`	`REMAINING_RECORDS` 前回のポーリングから残っているレコードの数。

コンストラクターの概要

コンストラクター
コンストラクターと説明
`KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K, V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties)` 指定されたパラメーターを使用してインスタンスを作成します。
`KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K, V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, boolean allowMultiFetch)` 指定されたパラメーターを使用してインスタンスを作成します。
`KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K, V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, KafkaMessageSource.KafkaAckCallbackFactory<K, V> ackCallbackFactory)` 指定されたパラメーターを使用してインスタンスを作成します。
`KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K, V> consumerFactory, org.springframework.kafka.listener.ConsumerProperties consumerProperties, KafkaMessageSource.KafkaAckCallbackFactory<K, V> ackCallbackFactory, boolean allowMultiFetch)` 指定されたパラメーターを使用してインスタンスを作成します。

メソッドのサマリー

すべてのメソッドインスタンスメソッド具象メソッド
修飾子と型	メソッドと説明
`protected void`	`createConsumer()`
`void`	`destroy()`
`protected ObjectSE`	`doReceive()` サブクラスはこのメソッドを実装する必要があります。
`CollectionSE<org.apache.kafka.common.TopicPartition>`	`getAssignedPartitions()` 現在割り当てられているパーティションを返します。
`protected StringSE`	`getClientId()`
`protected java.time.Duration`	`getCommitTimeout()`
`StringSE`	`getComponentType()`
`org.springframework.kafka.listener.ConsumerProperties`	`getConsumerProperties()` 構成されたコンシューマープロパティへの参照を取得します。ソースを開始する前に、プロパティをさらにカスタマイズできます。
`protected StringSE`	`getGroupId()`
`protected org.springframework.kafka.support.converter.RecordMessageConverter`	`getMessageConverter()`
`protected ClassSE<?>`	`getPayloadType()`
`protected long`	`getPollTimeout()`
`protected org.apache.kafka.clients.consumer.ConsumerRebalanceListener`	`getRebalanceListener()`
`boolean`	`isPaused()` エンドポイントが一時停止しているかどうかを確認します。
`protected boolean`	`isRawMessageHeader()`
`boolean`	`isRunning()`
`protected void`	`onInit()`
`void`	`pause()` エンドポイントを一時停止します。
`void`	`resume()` 一時停止した場合は、エンドポイントを再開します。
`void`	`setCloseTimeout(java.time.Duration closeTimeout)` クローズタイムアウトを設定します。デフォルトは 30 秒です。
`void`	`setMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)` デフォルトの `MessagingMessageConverter` を置き換えるようにメッセージコンバーターを設定します。
`void`	`setPayloadType(ClassSE<?> payloadType)` ペイロード型を設定します。
`void`	`setRawMessageHeader(boolean rawMessageHeader)` true に設定すると、生の `ConsumerRecord` がキー `KafkaHeaders.RAW_DATA` および `IntegrationMessageHeaderAccessor.SOURCE_DATA` のヘッダーとして含まれます。
`void`	`start()`
`void`	`stop()`

クラス org.springframework.integration.endpoint.AbstractMessageSource から継承されたメソッド
buildMessage, getBeanName, getComponentName, getManagedName, getManagedType, getOverrides, isLoggingEnabled, receive, registerMetricsCaptor, setBeanName, setHeaderExpressions, setLoggingEnabled, setManagedName, setManagedType

クラス org.springframework.integration.util.AbstractExpressionEvaluator から継承されたメソッド
afterPropertiesSet, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, evaluateExpression, getBeanFactory, getEvaluationContext, getEvaluationContext, getMessageBuilderFactory, setBeanFactory, setConversionService

クラス java.lang.ObjectSE から継承されたメソッド
cloneSE, equalsSE, finalizeSE, getClassSE, hashCodeSE, notifySE, notifyAllSE, toStringSE, waitSE, waitSE, waitSE

インターフェース org.springframework.integration.core.MessageSource から継承されたメソッド
getIntegrationPatternType

インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたメソッド
getThisAs

- フィールドの詳細
  - REMAINING_RECORDS
```
public static final StringSE REMAINING_RECORDS
```
    前回のポーリングから残っているレコードの数。
    導入:
    3.2
    関連事項:
    定数フィールド値
- コンストラクターの詳細
  - KafkaMessageSource
```
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
                          org.springframework.kafka.listener.ConsumerProperties consumerProperties)
```
    指定されたパラメーターを使用してインスタンスを作成します。ポーリングごとに複数のレコードをフェッチすることは無効になります。
    パラメーター:
    consumerFactory - コンシューマーファクトリ。
    consumerProperties - コンシューマーの特性。
    導入:
    3.2
    関連事項:
    KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
  - KafkaMessageSource
```
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
                          org.springframework.kafka.listener.ConsumerProperties consumerProperties,
                          boolean allowMultiFetch)
```
    指定されたパラメーターを使用してインスタンスを構築します。各ポーリングで最大 max.poll.records を取得できるようにするには、"allowMultiFetch" を true に設定します。false (デフォルト) の場合、コンシューマーファクトリが DefaultKafkaConsumerFactory の場合は max.poll.records が 1 に強制変換され、それ以外の場合は IllegalArgumentExceptionSE で拒否されます。重要: true の場合、max.poll.interval.ms 内で受信したレコード数を消費するのに十分な速度で AbstractMessageSource.receive() を呼び出す必要があります。false の場合、max.poll.interval.ms 内で AbstractMessageSource.receive() を呼び出す必要があります。pause() は、前回のポーリングのレコードが消費されるまで有効になりません。
    パラメーター:
    consumerFactory - コンシューマーファクトリ。
    consumerProperties - コンシューマーの特性。
    allowMultiFetch - max.poll.records > 1 を許可する場合は true。
    導入:
    3.2
  - KafkaMessageSource
```
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
                          org.springframework.kafka.listener.ConsumerProperties consumerProperties,
                          KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory)
```
    指定されたパラメーターを使用してインスタンスを作成します。ポーリングごとに複数のレコードをフェッチすることは無効になります。
    パラメーター:
    consumerFactory - コンシューマーファクトリ。
    consumerProperties - コンシューマーの特性。
    ackCallbackFactory - ack コールバックファクトリ。
    導入:
    3.2
    関連事項:
    KafkaMessageSource(ConsumerFactory, ConsumerProperties, KafkaAckCallbackFactory, boolean)
  - KafkaMessageSource
```
public KafkaMessageSource(org.springframework.kafka.core.ConsumerFactory<K,V> consumerFactory,
                          org.springframework.kafka.listener.ConsumerProperties consumerProperties,
                          KafkaMessageSource.KafkaAckCallbackFactory<K,V> ackCallbackFactory,
                          boolean allowMultiFetch)
```
    指定されたパラメーターを使用してインスタンスを構築します。各ポーリングで最大 max.poll.records を取得できるようにするには、"allowMultiFetch" を true に設定します。false (デフォルト) の場合、コンシューマーファクトリが DefaultKafkaConsumerFactory の場合は max.poll.records が 1 に強制変換され、それ以外の場合は IllegalArgumentExceptionSE で拒否されます。重要: true の場合、max.poll.interval.ms 内で受信したレコード数を消費するのに十分な速度で AbstractMessageSource.receive() を呼び出す必要があります。false の場合、max.poll.interval.ms 内で AbstractMessageSource.receive() を呼び出す必要があります。pause() は、前回のポーリングのレコードが消費されるまで有効になりません。
    パラメーター:
    consumerFactory - コンシューマーファクトリ。
    consumerProperties - コンシューマーの特性。
    ackCallbackFactory - ack コールバックファクトリ。
    allowMultiFetch - max.poll.records > 1 を許可する場合は true。
    導入:
    3.2
- メソッドの詳細
  - getAssignedPartitions
```
public CollectionSE<org.apache.kafka.common.TopicPartition> getAssignedPartitions()
```
    現在割り当てられているパーティションを返します。
    戻り値:
    パーティション。
    導入:
    3.2.2
  - onInit
```
protected void onInit()
```
    オーバーライド:
    クラス AbstractExpressionEvaluator の onInit
  - getConsumerProperties
```
public org.springframework.kafka.listener.ConsumerProperties getConsumerProperties()
```
    構成されたコンシューマープロパティへの参照を取得します。ソースを開始する前に、プロパティをさらにカスタマイズできます。
    戻り値:
    プロパティ。
    導入:
    3.2
  - getGroupId
```
protected StringSE getGroupId()
```
  - getClientId
```
protected StringSE getClientId()
```
  - getPollTimeout
```
protected long getPollTimeout()
```
  - getMessageConverter
```
protected org.springframework.kafka.support.converter.RecordMessageConverter getMessageConverter()
```
  - setMessageConverter
```
public void setMessageConverter(org.springframework.kafka.support.converter.RecordMessageConverter messageConverter)
```
    デフォルトの MessagingMessageConverter を置き換えるようにメッセージコンバーターを設定します。
    パラメーター:
    messageConverter - コンバーター。
  - getPayloadType
```
protected ClassSE<?> getPayloadType()
```
  - setPayloadType
```
public void setPayloadType(ClassSE<?> payloadType)
```
    ペイロード型を設定します。型対応のメッセージコンバーターが提供されている場合にのみ適用されます。
    パラメーター:
    payloadType - 変換する型。
  - getRebalanceListener
```
protected org.apache.kafka.clients.consumer.ConsumerRebalanceListener getRebalanceListener()
```
  - getComponentType
```
public StringSE getComponentType()
```
    次で指定:
    インターフェース NamedComponent の getComponentType
  - isRawMessageHeader
```
protected boolean isRawMessageHeader()
```
  - setRawMessageHeader
```
public void setRawMessageHeader(boolean rawMessageHeader)
```
    true に設定すると、生の ConsumerRecord がキー KafkaHeaders.RAW_DATA および IntegrationMessageHeaderAccessor.SOURCE_DATA のヘッダーとして含まれます。呼び出し元がレコードにアクセスしてエラーを処理できるようにします。
    パラメーター:
    rawMessageHeader - ヘッダーを含める場合は true。
  - getCommitTimeout
```
protected java.time.Duration getCommitTimeout()
```
  - setCloseTimeout
```
public void setCloseTimeout(java.time.Duration closeTimeout)
```
    クローズタイムアウトを設定します。デフォルトは 30 秒です。
    パラメーター:
    closeTimeout - クローズタイムアウト。
  - isRunning
```
public boolean isRunning()
```
    次で指定:
    インターフェース Lifecycle の isRunning
    次で指定:
    インターフェース ManageableLifecycle の isRunning
  - start
```
public void start()
```
    次で指定:
    インターフェース Lifecycle の start
    次で指定:
    インターフェース ManageableLifecycle の start
  - stop
```
public void stop()
```
    次で指定:
    インターフェース Lifecycle の stop
    次で指定:
    インターフェース ManageableLifecycle の stop
  - pause
```
public void pause()
```
    インターフェースからコピーされた説明: Pausable
    エンドポイントを一時停止します。
    次で指定:
    インターフェース Pausable の pause
  - resume
```
public void resume()
```
    インターフェースからコピーされた説明: Pausable
    一時停止した場合は、エンドポイントを再開します。
    次で指定:
    インターフェース Pausable の resume
  - isPaused
```
public boolean isPaused()
```
    インターフェースからコピーされた説明: Pausable
    エンドポイントが一時停止しているかどうかを確認します。
    次で指定:
    インターフェース Pausable の isPaused
    戻り値:
    一時停止した場合は true。
  - doReceive
```
protected ObjectSE doReceive()
```
    クラスからコピーされた説明: AbstractMessageSource
    サブクラスはこのメソッドを実装する必要があります。通常、戻り値は T 型の payload ですが、戻り値はペイロードが T 型の Message インスタンスでもかまいません。追加のヘッダーの作成に使用される AbstractIntegrationMessageBuilder にすることもできます。
    次で指定:
    クラス AbstractMessageSource<ObjectSE> の doReceive
    戻り値:
    返された値。
  - createConsumer
```
protected void createConsumer()
```
  - destroy
```
public void destroy()
```
    次で指定:
    インターフェース DisposableBean の destroy
    次で指定:
    インターフェース IntegrationManagement の destroy
    オーバーライド:
    クラス AbstractMessageSource<ObjectSE> の destroy

クラス KafkaMessageSource<K,V>

ネストされたクラスの要約

インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたネストクラス / インターフェース

フィールドのサマリー

クラス org.springframework.integration.util.AbstractExpressionEvaluator から継承されたフィールド

インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたフィールド

コンストラクターの概要

メソッドのサマリー

クラス org.springframework.integration.endpoint.AbstractMessageSource から継承されたメソッド

クラス org.springframework.integration.util.AbstractExpressionEvaluator から継承されたメソッド

クラス java.lang.ObjectSE から継承されたメソッド

インターフェース org.springframework.integration.core.MessageSource から継承されたメソッド

インターフェース org.springframework.integration.support.management.IntegrationManagement から継承されたメソッド

フィールドの詳細

REMAINING_RECORDS

コンストラクターの詳細

KafkaMessageSource

KafkaMessageSource

KafkaMessageSource

KafkaMessageSource

メソッドの詳細

getAssignedPartitions

onInit

getConsumerProperties

getGroupId

getClientId

getPollTimeout

getMessageConverter

setMessageConverter

getPayloadType

setPayloadType

getRebalanceListener

getComponentType

isRawMessageHeader

setRawMessageHeader

getCommitTimeout

setCloseTimeout

isRunning

start

stop

pause

resume

isPaused

doReceive

createConsumer

destroy