2.12.3　Kafkaインバウンドアダプタの機能

(a)　パーティションの割り当て

パーティションの割り当ては，Consumerグループのグループリーダー（グループ内の最初のConsumer）によって動的に行われます。Consumerに手動でパーティションを割り当てることはできません。

Consumerグループについては，「(5)　KafkaインバウンドアダプタのConsumerグループ」を参照してください。

(b)　サブスクライブ

Consumerが，どのトピックからレコードを受信するか設定することをサブスクライブといいます。Consumerのサブスクライブ対象のトピックを次に示します。

• ユーザが作成したトピック

• Kafkaが作成した内部トピック

サブスクライブ対象のトピックが存在しない場合でも，Broker側ではトピックは自動生成されません。この場合，Consumerの処理の実行中にトピックを作成することで，トピック作成後のポーリング処理でデータが読み込まれます。

(c)　データのポーリング

データのポーリングでは，Consumerに割り当てられたトピック内のパーティションからデータを読み込みます。データの取得単位は，レコードバッチ単位です。ポーリング処理は，この機能が停止しない限り繰り返し行われます。

クラスタの初期接続時のポーリング

Consumerは，Kafkaクラスタの情報を取得するために，データのポーリング実行時にKafkaクラスタの初期接続を実施します。Kafkaクラスタへの初期接続を確立するために使用するホスト／ポートペアのリストのどれかのBrokerと初期接続を行い，接続できたBrokerからKafkaクラスタの情報を取得します。ポーリング処理中に接続できた場合，以降のポーリングでは初期接続は必要ありません。

初期接続ができない場合，ポーリングの待ち時間内に繰り返し初期接続を行います。

ポーリングの待ち時間内に初期接続ができなかった場合，データのポーリング実行時にエラーは発生しません。

Brokerの接続時のポーリング

初期接続が行えた場合は，次に示すBrokerに接続します。

• Consumerグループの，グループコーディネータのBroker

• パーティションのリーダーであるBroker

ポーリングの待ち時間内に接続ができなかった場合は，ポーリング実行時にエラーは発生しません。

レコードのデシリアライズ

Consumerは，Brokerから取得したバイト配列のレコードをデシリアライズします。デシリアライズに使用するデシリアライザをKafka受付定義ファイルで設定できます。レコードのデシリアライズに関するプロパティを次に示します。

表2‒28　レコードのデシリアライズに関するプロパティ

プロパティ	説明
consumer.key.deserializer	レコードのキーのデシリアライザを指定します。
consumer.value.deserializer	レコードの値のデシリアライザを指定します。

取得するメッセージの順序

Consumerが取得するメッセージは，単一のパーティション内のメッセージに対してだけ順序が保証されます。トピックが複数のパーティションで構成される場合，取り出されるメッセージが，送信された順序で取得できるとは限りません。

受信するレコード数の上限

ポーリングごとにBrokerから受信するレコード数の上限を設定できます。レコード数の上限は，Kafka受付定義ファイルのconsumer.max.poll.recordsで設定します。

フェッチ要求のレコードのデータサイズ

Consumerからのデータのフェッチ要求に対して，Brokerが返す必要のあるデータサイズを設定できます。データサイズとは，1回のポーリングでConsumerが送信する，すべてのレコードデータのサイズです。

フェッチ要求のレコードのデータサイズについては，「(7)　レコードのデータサイズ」を参照してください。

(d)　受信できるメッセージの圧縮形式

Consumerでは，次に示す形式で圧縮されたメッセージが受信できます。Consumer はメッセージを伸張して処理を行います。Consumerが受信できない形式で圧縮されたレコードを受信した場合の動作は保証されません。

(e)　コミット

Consumerがどこまでメッセージを読み込んだかをBrokerに通知することをコミットといいます。

コミットについての詳細は，「(6)　コミットのタイミング」を参照してください。

(f)　セキュリティ

Kafkaインバウンドアダプタでは，TLSを使用した暗号化と認証，ホスト名の検証をサポートしています。TLSで使用するセキュリティ通信用プロトコルについては，Kafkaの公式ドキュメントを参照してください。

TLSで使用できるTLS暗号スイートの制限はありません。ただし，使用する暗号スイートを設定することはできません。

ホスト名の検証は，証明書内のホスト名が接続先のホスト名と一致するかどうかをチェックします。ホスト名の検証を無効にすることはできません。

(g)　レコード情報の作成

Brokerから取得したレコードを基に，Kafka受付に渡すレコード情報を作成します。レコード情報の作成は，1回のポーリングで受信したレコードごとに行います。レコード情報が作成されるのは，Brokerからメッセージが取得できた場合だけです。

取得したメッセージから作成されるレコード情報を次に示します。

(a)　Consumerグループのメンバーへのパーティションの割り当て

パーティション数に応じて，Consumerグループのメンバーへのパーティションは次のように割り当てられます。

パーティション数 < グループ内のメンバーの場合

パーティション数がグループ内のメンバーより少ない場合，パーティションが割り当てられないConsumerが存在します。パーティションが割り当てられないConsumerは，リバランスでパーティションが割り当てられない限りメッセージを受信しません。

パーティション数がグループ内のメンバーより少ない場合の割り当て例を次に示します。

パーティション数 < グループ内のメンバーの場合の割り当て例

Consumer2に割り当てるパーティションがないため，Consumer2はメッセージを受信しません。

パーティション数 = グループ内のメンバーの場合

パーティション数がグループ内のメンバーと等しい場合，グループ内のすべてのConsumerがメッセージを受信できる状態となります。

パーティション数がグループ内のメンバーと等しい場合の割り当て例を次に示します。

パーティション数 = グループ内のメンバーの場合の割り当て例

グループ内のすべてのConsumerにパーティションが割り当たります。

パーティション数 > グループ内のメンバーの場合

パーティション数がグループ内のメンバーより多い場合，グループ内のどれかのConsumerに，多くのパーティションが割り当てられます。

パーティション数がグループ内のメンバーより多い場合の割り当て例を次に示します。

パーティション数 > グループ内のメンバーの場合の割り当て例

グループ内にパーティションを均等に割り当てることができないため，Consumer1に2つのパーティションが割り当てられます。

(b)　パーティションの範囲割り当て

Kafkaでは，Consumerがグループ構成になっている場合，トピック内のパーティションをグループのメンバーに分散して割り当てます。これを範囲割り当てといいます。

範囲割り当ては，次の順序で行われます。

1. トピックごとに使用できるパーティションを番号順に，Consumerを辞書式順序でレイアウトします。

2. パーティションの数をConsumerの総数で割り，各Consumerに割り当てるパーティションの数を決定します。

   均等に分割できない場合，最初のConsumerから順にパーティションを1つ追加で割り当てます。

範囲割り当ての例

2つのConsumerC0とC1，2つのトピックt0とt1があり，各トピックに3つのパーティションp0，p1，p2がある場合，パーティションはt0p0，t0p1，t0p2，t1p0，t1p1，およびt1p2の6つとなります。

トピックt0のパーティションは3つあり，これらのパーティションを2つのConsumerで分割します。均等に分割できないため，C0にパーティションが1つ追加で割り当てられます。トピックt1のパーティションも同様に割り当てが行われます。

そのため，Consumerごとのパーティションの割り当ては次のようになります。

• C0：t0p0，t0p1，t1p0，t1p1

• C1：t0p2，t1p2

(c)　Consumerグループのメンバーシップ

Consumerグループのメンバーシップは動的メンバーシップとなります。静的メンバーシップに変更することはできません。

(d)　Consumerグループのリバランス

ConsumerグループへのConsumerの加入や除外などを契機として，Brokerがグループ内のリバランスを行います。Consumerグループのリバランス時は，ポーリングが停止し，リバランス完了後にポーリングが再開します。

(e)　冗長構成

パーティションはConsumerごとに割り当てられるため，冗長構成にすることができます。Consumerグループがクラスタ内の同一のKafka受付を使用した冗長構成の例を次の図に示します。

ただし，Consumerごとにパーティションが割り当てられるため，冗長構成にしても特定のHCSCサーバの負荷を分散することはできません。

(f)　注意事項

Consumerグループについての注意事項を次に示します。

• Consumerグループに対する操作は，Consumerグループの情報表示に関する操作，およびオフセットのリセットだけ許可されています。それ以外の操作を行った場合の動作は保証されません。

• サービスプラットフォームで構成したConsumerグループに参加できるのは，サービスプラットフォームのConsumerだけです。その他の製品のConsumerがグループに参加した場合の動作は保証されません。

(a)　メッセージ受信後のコミット

メッセージの処理順序を考慮する必要がない場合に設定します。この場合，ビジネスプロセスのメッセージ処理が失敗したときに，該当するメッセージ処理の回復処理を行わなくても，該当するメッセージよりあとのオフセットのメッセージが処理できます。

Kafka受付からの応答が正常か，異常かに関わらず，「現在のオフセット」および「コミットされたオフセット」が更新されるため，次回のポーリングでは，前回受信したメッセージの次のオフセットのメッセージが受信されます。

図中の1.〜3.について説明します。

1. ポーリング処理でメッセージを受信します。

   受信するメッセージ数は1つです。現在のオフセットは「1」であるため，オフセット「1」のメッセージを受信します。メッセージ受信後に，現在のオフセットは「2」になります。

2. メッセージをコミットします。

   コミット実行後のオフセット（コミットされたオフセット）は「2」となります。

3. オフセット「1」のメッセージをKafka受付に送信します。

   Kafka受付からの応答が正常か，異常かに関わらず，現在のオフセットは「2」のため，次回のポーリングではオフセット「2」のメッセージを受信します。

   また，コミット後にリバランスが発生して，該当するパーティションがグループ内のConsumerに引き継がれた場合は，パーティションを引き継いだConsumerは2.でコミットされたオフセットからメッセージを受信します。

(b)　メッセージ処理後のコミット

ビジネスプロセスのメッセージ処理が失敗した場合，該当するメッセージ処理の回復処理を行ってから，次のオフセットのメッセージを処理する必要があります。この場合，該当するメッセージは再送されます。

メッセージ処理後のコミットでは，Kafka受付からの応答が正常な場合だけコミットを行います。メッセージ受信後の処理が失敗した場合や，Kafka受付からエラー応答が返ってきた場合はコミットしません。また，次回のポーリングでは前回受信したメッセージを再度受信します。

図中の1.〜2.について説明します。

1. ポーリング処理を行い，メッセージを受信します。

   現在のオフセットは「1」のため，オフセット「1」のメッセージを受信します。メッセージ受信後に現在のオフセットは「2」となります。

2. オフセット「1」のメッセージをKafka受付に送信します。

   Kafka受付からエラー応答があった場合，コミット処理が省略されるため，コミットされたオフセットは更新されません。

   また，次回のポーリングで同一のメッセージを受信するために，現在のオフセットをエラーが発生したメッセージのオフセットに戻します。

(c)　オフセットの自動リセット

オフセットが初期化されていない場合，または「現在のオフセット」がサーバ上にない場合は，オフセットの自動リセットを行います。自動リセットは，Kafka受付定義ファイルのconsumer.auto.offset.resetで設定します。

自動リセットの方法に応じて取得されるデータを次に示します。

自動リセットの設定に応じた自動リセットで取得できるメッセージの例を次に示します。

earliest（最も早いオフセットにリセット）を設定した場合，オフセットが0にリセットされるため，オフセット0以降のメッセージを受信できます。

latest（最新のオフセットにリセット）を設定した場合，サブスクライブのあとに送信されたメッセージであるオフセット4以降のメッセージが受信できます。サブスクライブ前にパーティションに存在していたオフセット0〜3のメッセージは受信しません。