Apache Kafka モジュールドキュメントリファレンス

設定


コンシューマ設定

パラメータ

名前 説明 デフォルト値 必須

Name (名前)

String (文字列)

この設定の名前。コネクタはこの名前の設定を参照します。

x

Connection (接続)

この設定に指定できる接続型。

x

Expiration Policy (有効期限ポリシー)

ランタイムで有効期限の対象とみなされるまで、動的設定インスタンスがアイドル状態を維持できる最小時間を設定します。これは、対象になった瞬間にプラットフォームでインスタンスの有効期限が切れるという意味ではありません。ランタイムによって適切と判断されたときに、インスタンスが実際にパージされます。

接続種別

Kafka の基本コンシューマ接続
パラメータ
名前 説明 デフォルト値 必須

Consumer Partitions (コンシューマパーティション)

Number (数値)

このコンシューマに使用するパーティションの数。

1

Group Id (グループ ID)

String (文字列)

このコンシューマが属するコンシューマグループを識別する一意の文字列。

x

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Kafka の Kerberos コンシューマ接続
パラメータ
名前 説明 デフォルト値 必須

Consumer Partitions (コンシューマパーティション)

Number (数値)

このコンシューマに使用するパーティションの数。

1

Group Id (グループ ID)

String (文字列)

このコンシューマが属するコンシューマグループを識別する一意の文字列。

x

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Principal (プリンシパル)

String (文字列)

Kerberos プリンシパル。

x

Keytab (キータブ)

String (文字列)

「principal」に関連付けられている keytab ファイルへのパス。

x

Service Name (サービス名)

String (文字列)

Kafka ブローカーが実行時に使用する Kerberos プリンシパル名。

x

Additional JAAS Properties (JAAS の追加プロパティ)

Object (オブジェクト)

キー→値の追加プロパティ。「sasl.jaas.config」で設定する必要があり、通常は JAAS 設定ファイルに格納します。

Kafka の Kerberos SSL コンシューマ接続
パラメータ
名前 説明 デフォルト値 必須

Consumer Partitions (コンシューマパーティション)

Number (数値)

このコンシューマに使用するパーティションの数。

1

Group Id (グループ ID)

String (文字列)

このコンシューマが属するコンシューマグループを識別する一意の文字列。

x

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Key Store Type (キーストア種別)

String (文字列)

キーストアファイルのファイル形式。クライアント側では省略可能です。

JKS

Key Store Password (キーストアのパスワード)

String (文字列)

キーストアファイルのストアのパスワード。クライアント側では省略可能で、「keyStoreLocation」が設定されている場合にのみ必要です。

Key Store Location (キーストアの場所)

String (文字列)

キーストアファイルの場所。クライアント側では省略可能で、クライアントの双方向認証に使用できます。

Trust Store Type (トラストストア種別)

String (文字列)

トラストストアファイルのファイル形式。

JKS

Trust Store Password (トラストストアのパスワード)

String (文字列)

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアへのアクセスは可能ですが、整合性チェックが無効になります。

x

Trust Store Location (トラストストアの場所)

String (文字列)

トラストストアファイルの場所。

x

Principal (プリンシパル)

String (文字列)

Kerberos プリンシパル。

x

Keytab (キータブ)

String (文字列)

「principal」に関連付けられている keytab ファイルへのパス。

x

Service Name (サービス名)

String (文字列)

Kafka ブローカーが実行時に使用する Kerberos プリンシパル名。

x

Additional JAAS Properties (JAAS の追加プロパティ)

Object (オブジェクト)

キー→値の追加プロパティ。「sasl.jaas.config」で設定する必要があり、通常は JAAS 設定ファイルに格納します。

Kafka の SSL コンシューマ接続
パラメータ
名前 説明 デフォルト値 必須

Consumer Partitions (コンシューマパーティション)

Number (数値)

このコンシューマに使用するパーティションの数。

1

Group Id (グループ ID)

String (文字列)

このコンシューマが属するコンシューマグループを識別する一意の文字列。

x

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Key Store Type (キーストア種別)

String (文字列)

キーストアファイルのファイル形式。クライアント側では省略可能です。

JKS

Key Store Password (キーストアのパスワード)

String (文字列)

キーストアファイルのストアのパスワード。クライアント側では省略可能で、「keyStoreLocation」が設定されている場合にのみ必要です。

Key Store Location (キーストアの場所)

String (文字列)

キーストアファイルの場所。クライアント側では省略可能で、クライアントの双方向認証に使用できます。

Trust Store Type (トラストストア種別)

String (文字列)

トラストストアファイルのファイル形式。

JKS

Trust Store Password (トラストストアのパスワード)

String (文字列)

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアへのアクセスは可能ですが、整合性チェックが無効になります。

x

Trust Store Location (トラストストアの場所)

String (文字列)

トラストストアファイルの場所。

x

関連付けられたソース


プロデューサ設定

パラメータ

名前 説明 デフォルト値 必須

Name (名前)

String (文字列)

この設定の名前。コネクタはこの名前の設定を参照します。

x

Connection (接続)

この設定に指定できる接続型。

x

Expiration Policy (有効期限ポリシー)

ランタイムで有効期限の対象とみなされるまで、動的設定インスタンスがアイドル状態を維持できる最小時間を設定します。これは、対象になった瞬間にプラットフォームでインスタンスの有効期限が切れるという意味ではありません。ランタイムによって適切と判断されたときに、インスタンスが実際にパージされます。

接続種別

Kafka の基本プロデューサ接続
パラメータ
名前 説明 デフォルト値 必須

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Kafka の Kerberos プロデューサ接続
パラメータ
名前 説明 デフォルト値 必須

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Principal (プリンシパル)

String (文字列)

Kerberos プリンシパル。

x

Keytab (キータブ)

String (文字列)

「principal」に関連付けられている keytab ファイルへのパス。

x

Service Name (サービス名)

String (文字列)

Kafka ブローカーが実行時に使用する Kerberos プリンシパル名。

x

Additional JAAS Properties (JAAS の追加プロパティ)

Object (オブジェクト)

キー→値の追加プロパティ。「sasl.jaas.config」で設定する必要があり、通常は JAAS 設定ファイルに格納します。

Kafka の Kerberos SSL プロデューサ接続
パラメータ
名前 説明 デフォルト値 必須

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Key Store Type (キーストア種別)

String (文字列)

キーストアファイルのファイル形式。クライアント側では省略可能です。

JKS

Key Store Password (キーストアのパスワード)

String (文字列)

キーストアファイルのストアのパスワード。クライアント側では省略可能で、「keyStoreLocation」が設定されている場合にのみ必要です。

Key Store Location (キーストアの場所)

String (文字列)

キーストアファイルの場所。クライアント側では省略可能で、クライアントの双方向認証に使用できます。

Trust Store Type (トラストストア種別)

String (文字列)

トラストストアファイルのファイル形式。

JKS

Trust Store Password (トラストストアのパスワード)

String (文字列)

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアへのアクセスは可能ですが、整合性チェックが無効になります。

x

Trust Store Location (トラストストアの場所)

String (文字列)

トラストストアファイルの場所。

x

Principal (プリンシパル)

String (文字列)

Kerberos プリンシパル。

x

Keytab (キータブ)

String (文字列)

「principal」に関連付けられている keytab ファイルへのパス。

x

Service Name (サービス名)

String (文字列)

Kafka ブローカーが実行時に使用する Kerberos プリンシパル名。

x

Additional JAAS Properties (JAAS の追加プロパティ)

Object (オブジェクト)

キー→値の追加プロパティ。「sasl.jaas.config」で設定する必要があり、通常は JAAS 設定ファイルに格納します。

Kafka の SSL プロデューサ接続
パラメータ
名前 説明 デフォルト値 必須

Bootstrap Servers (ブートストラップサーバ)

String (文字列)

Kafka クラスタへの初期接続の確立に使用するコンマ区切りのホストとポートのペア。Kafka クライアント (プロデューサ/コンシューマ) に指定する必要のある「bootstrap.servers」値と同じです。プロデューサ/コンシューマプロパティファイルからこのプロパティが指定される場合は、この値が無視され、プロパティファイルのいずれかが使用されます。

x

Additional properties (追加プロパティ)

Object (オブジェクト)

接続に必要なキー→値の追加プロパティ。

Reconnection (再接続)

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Key Store Type (キーストア種別)

String (文字列)

キーストアファイルのファイル形式。クライアント側では省略可能です。

JKS

Key Store Password (キーストアのパスワード)

String (文字列)

キーストアファイルのストアのパスワード。クライアント側では省略可能で、「keyStoreLocation」が設定されている場合にのみ必要です。

Key Store Location (キーストアの場所)

String (文字列)

キーストアファイルの場所。クライアント側では省略可能で、クライアントの双方向認証に使用できます。

Trust Store Type (トラストストア種別)

String (文字列)

トラストストアファイルのファイル形式。

JKS

Trust Store Password (トラストストアのパスワード)

String (文字列)

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアへのアクセスは可能ですが、整合性チェックが無効になります。

x

Trust Store Location (トラストストアの場所)

String (文字列)

トラストストアファイルの場所。

x

関連操作

Producer 操作

<kafka:producer>

指定したトピックへの Kafka メッセージの送信を促進する操作。

パラメータ

名前 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Topic (トピック)

String (文字列)

メッセージの送信先のトピック。

x

Key (キー)

String (文字列)

送信するメッセージに属するキー。

x

Message (メッセージ)

String (文字列)

送信するメッセージ。

#[payload]

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

次の設定の場合

スロー

  • KAFKA:CONNECTIVITY

  • KAFKA:CONNECTIVITY

  • KAFKA:UNKNOWN

  • KAFKA:RETRY_EXHAUSTED

Sources (ソース)

Consumer 操作

<kafka:consumer>

指定したトピックからの Kafka メッセージのコンシュームを促進する操作。

パラメータ

名前 説明 デフォルト値 必須

Configuration (設定)

String (文字列)

使用する設定の名前。

x

Topic (トピック)

String (文字列)

メッセージのコンシューム元の Kafka トピックの名前。

x

Partition offsets (パーティションオフセット)

Array of オフセット (の配列)

パーティションオフセットの設定を表すマップ。[{"partition_number": "partition_offset"}] (例: [{"0":"2"}、[{"1":"5"}]]) の形式にする必要があります。

Redelivery Policy (再配信ポリシー)

同じメッセージの再配信を処理するためのポリシーを定義します。

Streaming Strategy (ストリーミング戦略)

反復可能なストリームを使用するかどうかと、それらの動作を設定します。

Reconnection Strategy (再接続戦略)

接続エラーが発生した場合の再試行戦略。

出力

Type (型)

String (文字列)

Attributes Type (属性型)

Any (任意)

次の設定の場合

Reconnection (再接続)

項目 説明 デフォルト値 必須

Fails Deployment (デプロイに失敗)

Boolean

アプリケーションがデプロイされたときに、すべてのコネクタで接続テストが実行されます。true に設定されている場合、関連する再接続戦略をすべて実行した後にテストに合格しないと、デプロイが失敗します。

Reconnection Strategy (再接続戦略)

使用する再接続戦略。

Reconnect (再接続)

項目 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

Count (カウント)

Number (数値)

再接続の試行回数。

Reconnect Forever (繰り返し再接続)

項目 説明 デフォルト値 必須

Frequency (頻度)

Number (数値)

再接続する頻度 (ミリ秒)。

Expiration Policy (有効期限ポリシー)

項目 説明 デフォルト値 必須

Max Idle Time (最大アイドル時間)

Number (数値)

有効期限の対象とみなされるまで、動的設定インスタンスがアイドル状態を維持できる最大時間のスカラー時間値。

Time Unit (時間単位)

Enumeration (列挙)。次のいずれか:

  • NANOSECONDS (ナノ秒)

  • MICROSECONDS (マイクロ秒)

  • MILLISECONDS (ミリ秒)

  • SECONDS (秒)

  • MINUTES (分)

  • HOURS (時)

  • DAYS (日)

maxIdleTime 属性に使用できる時間単位。

オフセット

項目 説明 デフォルト値 必須

Partition Number (パーティション番号)

String (文字列)

Partition Offset (パーティションオフセット)

String (文字列)

Redelivery Policy (再配信ポリシー)

項目 説明 デフォルト値 必須

Max Redelivery Count (最大再配信数)

Number (数値)

正常に処理されずにプロセス失敗メッセージがトリガされるまでにメッセージを再配信できる最大回数。

Use Secure Hash (セキュアハッシュを使用)

Boolean (ブール)

再配信されたメッセージの識別にセキュアハッシュアルゴリズムを使用するかどうか。

Message Digest Algorithm (メッセージダイジェストアルゴリズム)

String (文字列)

使用するセキュアハッシュアルゴリズム。設定しない場合、デフォルトの SHA-256 になります。

Id Expression (ID 式)

String (文字列)

メッセージがいつ再配信されたのかを判断するために使用する 1 つ以上の式を定義します。このプロパティは、useSecureHash が false の場合にのみ設定できます。

Object Store (オブジェクトストア)

各メッセージの再配信カウンタが保存されるオブジェクトストア。

Repeatable In Memory Stream (反復可能なメモリ内ストリーム)

項目 説明 デフォルト値 必須

Initial Buffer Size (初期バッファサイズ)

Number (数値)

これは、ストリームを消費し、ストリームへのランダムアクセスを提供する目的で割り当てられるメモリの容量です。ストリームに含まれているデータがこのバッファに適合できる量を超えている場合、bufferSizeIncrement 属性に従って拡張され、上限は maxInMemorySize となります。

Buffer Size Increment (バッファサイズ増分)

Number (数値)

バッファサイズが初期サイズを超えた場合に、どれくらい拡張されるか。0 以下の値を設定すると、バッファは拡張されません。つまり、バッファがいっぱいになると STREAM_MAXIMUM_SIZE_EXCEEDED エラーが発生します。

Max Buffer Size (最大バッファサイズ)

Number (数値)

使用するメモリの最大量。これより大きい値が使用されると、STREAM_MAXIMUM_SIZE_EXCEEDED エラーが発生します。0 以下の値は無制限を意味します。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれか:

  • BYTE (バイト)

  • KB (キロバイト)

  • MB (メガバイト)

  • GB (ギガバイト)

これらのすべての属性の単位。

Repeatable File Store Stream (反復可能なファイルストアストリーム)

項目 説明 デフォルト値 必須

Max In Memory Size (最大メモリ内サイズ)

Number (数値)

データをメモリ内に保持するためにストリームで使用する最大メモリを定義します。これを超えると、ディスクへのコンテンツのバッファが開始されます。

Buffer Unit (バッファ単位)

Enumeration (列挙)。次のいずれか:

  • BYTE (バイト)

  • KB (キロバイト)

  • MB (メガバイト)

  • GB (ギガバイト)

maxInMemorySize で表される単位。

Was this article helpful?

💙 Thanks for your feedback!

Edit on GitHub