私は Apache Kafka プロトコルを低レベルでかなり使いました。公式ガイドだけに従ってこれを始めるのは簡単ではなかったので、コードを何度も読みました。この投稿では、原始的な値から意味のある要求まで段階的にガイドして、有利なスタートを切りたいと思います。

この投稿の內(nèi)容:

1. Kafka プロトコル コードと、Wireshark で動作するプロトコルを調(diào)べます。
2. プリミティブ値の読み取りと書き込み方法を?qū)Wびます。
3. プリミティブを組み合わせて意味のあるリクエストを?qū)g行します。

プログラミング言語として Python を使用します。ただし、コードは依存関係がなく、選択した言語に簡単に移植できます。

イントロ

Apache Kafka には、さまざまなデータ型、オプションのフィールドなどを備えた、バージョン管理されたカスタム バイナリ プロトコルがあります。殘念ながら、Protobuf のようなよく知られたシリアル化形式は使用されません。プロトコルメッセージのスキーマはJSONで記述されます。シリアル化と逆シリアル化を行う実際の Java コードは、この記述から生成されます。

Java の世界では、公式のクライアント ライブラリを使用できます。ただし、別のプラットフォームを使用している場合は、サードパーティの実裝に依存することになります。それらは存在しますが、主にプロデューサーとコンシューマーに焦點を當てており、管理クライアントの一部の側(cè)面にはほとんど焦點を當てていません。他に何かする必要がある場合は、ご自身で行ってください。

この投稿は、Kafka プロトコルのハッキングを開始するのに役立ちます。 (Kafka プロトコル用の準備が整った Python (逆) シリアル化ライブラリを探している場合は、Kio¹ をチェックしてください。Rust については、私が取り組んでいるライブラリをご覧ください。)

この投稿のコードと、Github のこのリポジトリで同様のテストをいくつか見つけることができます。

プロトコルの概要

このページで公式のプロトコルの説明を見つけることができます。少なくとも「準備編」と「プロトコル」のセクションを読んで、よく理解しておくことをお勧めします。

ここにいくつかのハイライトがあります。 Kafka プロトコルは、TCP ベースのバイナリ要求/応答プロトコルです:

• TCP ベース: Kafka ブローカーは TCP スタック上のポートでリッスンします (これにより、順序付けの保証などの利點が得られます)。
• バイナリ: メッセージはバイナリ形式でエンコードされ、事前定義されたスキーマに従って特別なシリアル化と逆シリアル化が必要です。
• リクエスト-レスポンス: 交換はクライアントによって開始され、サーバーは受動的でリクエストにのみ応答します。

各 API メッセージ タイプはリクエストとレスポンスのペアで構(gòu)成され、API キーと呼ばれる數(shù)値によって識別されます。たとえば、最も特徴的な Kafka RPC である Produce と Fetch には、API キー 0 と 1 が対応しています。現(xiàn)在、90 近くの API メッセージ タイプがあります (クライアント ブローカーではなく、ブローカー間メッセージもあります)。

リクエストとレスポンスは、バージョン管理されたスキーマによって記述されます。バージョン管理により、フィールドの追加や削除、データ型の変更など、プロトコルの進化が可能になります。

最初のステップ

Kafka プロトコルの使用を開始するためにできることがいくつかあります。

Kafka プロトコル コードを?qū)W習する

Kafka コードは、プロトコルに関する (実質(zhì)的に) 真実の情報源です。 Github から Kafka コードをチェックアウトし、興味のあるリリース (例: 3.8.0) に切り替えます。

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

API メッセージ定義は、clients/src/main/resources/common/message に JSON であります。各 JSON ファイルには、1 つのメッセージ² タイプの定義とそのすべてのバージョンが含まれています。 client/src/main/resources/common/message/README.md には、スキーマ定義形式の概要が記載されています。デフォルト値、柔軟なバージョン、タグ付きフィールドなどに注意してください。

興味のある具體的な API メッセージ タイプとは別に、clients/src/main/resources/common/message/RequestHeader.json および ResponseHeader.json を見てください。これらは、各リクエストとレスポンスの交換で使用されるヘッダーについて説明しています。 .

コードジェネレーターを?qū)g行しましょう:

./gradlew processMessages

これで、clients/src/generated/java/org/apache/kafka/common/message で生成されたクラスを見つけることができます。

clients/src/generated/java/org/apache/kafka/common/message/ApiMessageType.java を見てください。このユーティリティ:

• 既存の API メッセージ タイプのセット全體とそのスキーマおよびバージョンについて説明します。
• requestHeaderVersion関數(shù)とresponseHeaderVersion関數(shù)で、APIメッセージのバージョンをリクエストヘッダーとレスポンスヘッダーのバージョンにマッピングします。

他のファイルは、対応するスキーマ JSON から 1 対 1 で生成されます (互換性のため、データ接尾辭が付いている場合もあります)。これらのファイルには次のものが含まれます:

1. バージョン管理されたスキーマ定義 SCHEMA_0、SCHEMA_1 など。スキーマがバージョン間で同じままである場合があります。これは正常であり、要求と応答の相手のみが変更されたことを意味します。
2. プロトコルのシリアル化と逆シリアル化のグラウンド トゥルースを見つけることができる読み取りおよび書き込みメソッド。

內(nèi)部クラスはメッセージの複雑な構(gòu)造を表すだけでなく、內(nèi)部クラスにも注意してください。

Docker で Kafka を?qū)g行する

Docker で Kafka を?qū)g行すると、ブローカーを?qū)g行してプロトコルをテストしたり、ネットワーク交換をキャプチャしたりするのに便利な方法です。バージョン 3.7.0 以降、Kafka チームは公式の Docker イメージを構(gòu)築し、次のように実行できます。

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

古いバージョンに興味がある場合は、Docker Hub で他のイメージを検索してください。ただし、Kafka プロトコルには下位互換性と上位互換性があることを考慮すると、これは必要ない可能性があります。新しいブローカーは古いプロトコル バージョンを問題なく認識し、古いクライアントは新しいブローカーと通信できます。

これを読んでいる方は、おそらくお使いのマシンに Kafka コマンド ライン ツールがすでにインストールされていると思いますが、念のため、これらのツールを Docker で実行することもできます。たとえば、これを?qū)g行してトピックを作成します:

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

Wireshark でプロトコルを検査する

Kafka コードについて理解したところで、実際のプロトコルを見てみましょう。 Wireshark は、このような検査に広く使用されているツールです。 Kafka プロトコルを分析できます (バージョンが十分に新しい場合は、最新バージョンもサポートします)。

オペレーティング システム パッケージが古く、新しいバージョンの Kafka プロトコルを分析できないため、バージョン 4.5.0 のソースから Wireshark をビルドしました。 Wireshark 4.5.0 は主に Kafka 3.7 プロトコル バージョンをサポートする必要があります。ただし、利用可能なバージョンを試して、それがどのように機能するかを確認することはできます。

ポート 9092 キャプチャ フィルター (1) と Kafka 表示フィルター (2) を使用して、ループバック インターフェイスで Wireshark を?qū)g行してみましょう:

トピックを作成し、Wireshark が何を表示するかを確認してください:

./gradlew processMessages

表示フィルターは無関係なものをすべて削除し、Kafka のリクエストと応答のみを殘します。 Wireshark はプロトコル內(nèi)のほとんどのメッセージ バージョンを理解するため (もちろん Wireshark のバージョンによって異なります)、各メッセージの構(gòu)造を簡単に調(diào)べることができます。 Wireshark は、対応するバイトも表示します。

Wireshark は、特定のケースでプロトコルがどのように動作するのか、実裝の何が問題なのかを理解するのに役立つ優(yōu)れたデバッグ ツールです。

プリミティブ値の読み取りと書き込み

このプロトコルでは多數(shù)のプリミティブ型が定義されており、詳しい説明はここでご覧いただけます。それらの読み取りおよび書き込みコードを?qū)g裝しましょう。このファイルにはすべての関數(shù)が含まれており、対応するテスト ファイルも確認してください。

固定長整數(shù)値: INT8、INT16、INT32、INT64、および UINT16

これらは既知の固定長の整數(shù)です: 1、2、4、または 8 バイト。當然のことながら、このようなフィールドはプロトコル全體でたくさん見つかります。このクラスでは、それらの読み取りと書き込みが Kafka でどのように (自明に) 実裝されているかを見ることができます。

最初に、バッファから正確なバイト數(shù)を読み取る関數(shù)を定義しましょう³:

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

Python の BinaryIO 型ヒントは、バイトの読み取りと書き込みが可能なオブジェクトを表します。 read、write、tell (現(xiàn)在の位置を取得するため)、seek (位置を変更するため) などのメソッドがあります。

これで、INT8 の読み取りを?qū)g裝できます:

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

Kafka はビッグエンディアン (別名ネットワーク) バイト順序を使用するため、byteorder="big" となります。

今書いています:

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

INT16、INT32、および INT64 についてはこれを繰り返しません。唯一の大きな違いは、バイト數(shù) (対応する 2、4、および 8) とチェックされる範囲 ([-(2**15)、2* *15 - 1]、[-(2**31)、2**31 - 1]、[-(2**63)、2**63 - 1] に相當します)。

UINT16 は INT16 に似ています:

./gradlew processMessages

ここで signed=False であることに注意してください。

ブール値

BOOLEAN は本質(zhì)的に INT8 に追加のロジックを加えたものです: == 0 は false を意味し、!= 0 は true を意味します。

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

MetadataRequestData 生成クラスのallowAutoTopicCreation フィールドで BOOLEAN の例を確認できます。

FLOAT64

FLOAT64 は、倍精度 64 ビット IEEE 754 値です。 Python には、int のような to_bytes と from_bytes が float にはありません。そのため、代わりに標準ライブラリの struct モジュールを使用します。

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

>d は、「ビッグエンディアン バイト オーダーの double 値」を意味します。

UNSIGNED_VARINT: 可変長整數(shù)値

可変長整數(shù)は、値が小さい場合に、値ごとに使用できるビット數(shù)を減らすアプローチです。 Kafka は、プロトコル バッファーからの varint アプローチを使用します。アイデアはシンプルです:

バリアント內(nèi)の各バイトには、それに続くバイトがバリアントの一部であるかどうかを示す継続ビットがあります。これはバイトの最上位ビット (MSB) です (符號ビットとも呼ばれます)。下位 7 ビットはペイロードです。結(jié)果の整數(shù)は、その構(gòu)成バイトの 7 ビット ペイロードを追加することによって構(gòu)築されます。

詳細については、Protobuf 仕様と Kafka 実裝 (読み取り、書き込み) を確認できます。

この型はプロトコル フィールド自體では使用されませんが、以下で説明するコンパクトなコレクションに使用されます。

実裝してみましょう。自信を持っていただくために、真実のソースである Kafka の ByteUtils クラスから直接いくつかの例を取得します。

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

これを?qū)g行すると、次の結(jié)果が得られます:

def read_exact(buffer: BinaryIO, num_bytes: int) -> bytes:
    value = buffer.read(num_bytes)
    if len(value) != num_bytes:
        raise ValueError(f"Buffer underflow: expected {num_bytes}, got {len(value)}")
    return value

これを、おそらく最もパフォーマンスが高いわけではありませんが、簡単な方法で実裝してみましょう:

def read_int8(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=True)

UUID

UUID は、エンティティを一意に識別するために使用される 128 ビットの値です。たとえば、CreateTopicsResponse.

でトピック ID を渡すために使用されます。

Kafka コードでこれらがどのように読み取られ、書き込まれるかを確認できます。再現(xiàn)は簡単です:

def write_int8(value: int, buffer: BinaryIO) -> None:
    if -(2**7) <= value <= 2**7 - 1:
        buffer.write(value.to_bytes(1, byteorder="big", signed=True))
    else:
        raise ValueError(f"Value {value} is out of range for INT8")

Kafka は null/None をゼロ UUID として扱うので、ここでも同じことを行っていることに注意してください。

文字列

Kafka プロトコルには 4 種類の文字列があります:

Compactness は、文字列の長さが INT16 でエンコードされるか UNSIGNED_VARINT でエンコードされるかを示します。メッセージのバージョンによって異なります（2017年頃に導(dǎo)入されました）。 Nullability は、値が null になれるかどうかです。それはメッセージの目的とバージョンにも依存します (プロトコルの進化中に文字列フィールドがオプションになる場合があります)。

文字列はプロトコル內(nèi)で広く使用されています。たとえば、生成されたクラス MetadataRequestData.MetadataRequestTopic.

のフィールド名を確認してください。

文字列は非常に単純にエンコードされます。最初に長さが指定され、次に UTF-8 でエンコードされた本體が続きます。許可される最大長は 32767 バイトです。 Null 文字列の長さは -1 であり、明らかに本文はありません。

コンパクトと非コンパクトの唯一の違いは文字列の長さがエンコードされる方法であるため、両方のモードに対して 1 つの関數(shù)を使用できます。

NULL 許容文字列の読み取りと書き込みから始めましょう:

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

null 非許容の文字列関數(shù)は、次のものに基づいて構(gòu)築できます。

./gradlew processMessages

バイト配列

バイト配列は文字列と非常によく似ています。これらは同じ潛在的な null 可能性とコンパクト性を持っています:

これらも同じ方法でエンコードされます: length body。當然のことながら、本文は UTF-8 文字列としてではなく、不透明なバイト配列として扱われます。バイト配列の最大長は 2147483647;

生成されたクラス JoinGroupRequestData.JoinGroupRequestProtocol のフィールド メタデータ內(nèi)のバイトの例を見つけることができます。

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

ご覧のとおり、これらの関數(shù)と文字列に対応する関數(shù)との違いはわずかです。

その他のアレイ

このプロトコルは、バイト以外の型の配列、文字列、數(shù)値、構(gòu)造體 (ただし、ネストされた配列は除く)、ARRAY および COMPACT_ARRAY をサポートします。 コンパクトさはバイト配列や文字列と同じです。

Null 可能性は、何らかの理由でプロトコル仕様に明示的に記載されていません。ただし、配列は null 値を許容することができます。これは、ここにあるように、スキーマ定義の nullableVersions によって制御されます。

read_array_length と write_array_length がすでに実裝されていることを考慮して、リーダー関數(shù)とライター関數(shù)を?qū)g裝しましょう。

./gradlew processMessages

記録

RECORDS は Kafka レコードをエンコードします。この構(gòu)造はかなり複雑なので、このガイドでは説明しません (ただし、ご希望の場合はコメントでお知らせください??)。簡単にするために、レコードを NULLABLE_BYTES または COMPACT_NULLABLE_BYTES として扱うことができます (メッセージのバージョンによって異なります)。

タグ付きフィールド

タグ付きフィールドは、オプションのデータをメッセージに添付できるようにする Kafka プロトコルの拡張機能です。このアイデアには 2 つの要素があります:

1. サービスのクライアントがタグ付きフィールドを理解できない場合、それを不明として保存し、無視します。
2. フィールドがほとんど使用されない場合、そのデフォルト値は転送からスキップできます。

たとえば、このフィールドを見てください。これには、このフィールドがどのバージョン以降にタグ付けされているかを示す taggedVersions があります (ほとんどの場合、フィールドが追加されたときと同じバージョンです)。

タグ付きフィールドは次のもので構(gòu)成されます:

1. UNSIGNED_VARINT 型のタグ。
2. COMPACT_BYTES 型のデータ。

タグ付きフィールドの詳細については、KIP-482 を參照してください。

実裝しましょう:

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

ここでは「不明」というタイトルが付けられています。既知のフィールドはその構(gòu)造內(nèi)に作成する必要があります。

メッセージ構(gòu)造

高レベルのメッセージ構(gòu)造は非常に単純です。仕様によると:

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

つまり、メッセージ自體の前にバイト単位のサイズが付加されます。要求メッセージと応答メッセージはどちらも、ヘッダーとその直後に本文が続きます。何らかの理由で、これは明示的に文書化されていません⁴が、信じていただけますか?またはコードを確認してください。

リクエストヘッダーとレスポンスヘッダー

リクエスト ヘッダーには 0、1、2 の 3 つのバージョンが存在します。これらはプロトコルで次のように指定されます。

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

TAG_BUFFER は、前述したタグ付きフィールドです。

これらを Python データ クラスとして実裝しましょう:

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

ご覧のとおり、バージョン 2 にはタグ付きフィールドがいくつかありますが、予期される既知のフィールドはありません。タグ付きフィールドが誤ってブローカーに送信された場合、それは無視されます。

応答ヘッダーには 0 と 1 の 2 つのバージョンが存在します。これらはプロトコルで次のように指定されます。

./gradlew processMessages

これらも実裝しましょう:

docker run --rm -ti -p 9092:9092 apache/kafka:3.8.0

リクエストヘッダーの読み取りとレスポンスヘッダーの書き込みは実裝しません。これは簡潔にするためです。サーバー側(cè)をプログラミングしていないため、この例では応答ヘッダーの送信と要求ヘッダーの受信は行いません。ただし、サーバー側(cè)にも興味がある場合は、両方の関數(shù)を?qū)g裝する必要があります (これは簡単なはずです)。

相関ID

リクエストヘッダーとレスポンスヘッダーの correlation_id フィールドに特に注意してください。このプロトコルはパイプライン処理をサポートしています。クライアントは接続ごとに複數(shù)の未処理のリクエストを持つことができます。相関 ID を使用すると、リクエストに対する応答を照合できます。

ヘッダーのバージョンの選択

どのバージョンを使用する必要があるかは、API キーとメッセージのバージョンによって決まります?，F(xiàn)在、プロトコル ガイドには明示的に記載されていません⁵.
生成されたクラス ApiMessageType の requestHeaderVersion 関數(shù)と responseHeaderVersion 関數(shù)を參照として使用します。

リクエストの送信とレスポンスの受信

ここまでの知識とコードをすべて理解したので、最後に ApiVersions リクエストを送信し、レスポンスを受信して??読み取りましょう。 ApiVersions は通常、クライアントが送信する最初のリクエストです。その目的は、ブローカーによってサポートされている API バージョンと機能を見つけることです。最新バージョン 3 を?qū)g裝します。

プロトコル仕様では、次のように定義されています。

docker run --rm -ti --net=host apache/kafka:3.8.0 \
  /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

データクラスを作成しましょう:

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --create \
  --topic test-topic1 --partitions 2

そして応答:

def read_exact(buffer: BinaryIO, num_bytes: int) -> bytes:
    value = buffer.read(num_bytes)
    if len(value) != num_bytes:
        raise ValueError(f"Buffer underflow: expected {num_bytes}, got {len(value)}")
    return value

[api_keys] は「api_keys の配列」を意味します。ここで、api_keys は以下の 2 行で定義された構(gòu)造です。

これを Python データ クラスに変換します:

def read_int8(buffer: BinaryIO) -> int:
    return int.from_bytes(read_exact(buffer, 1), byteorder="big", signed=True)

配列について話すとき、コンパクトな配列が必要か、それとも非コンパクトな配列が必要かを知る必要があります。これを確認するために、ApiVersionsRequest.json のスキーマ定義を見てみましょう。 "flexibleVersions": "3 " と表示されます。これは、コンパクト配列がバージョン 3 から使用されることを意味します (これについては、スキーマ ディレクトリの README.md を參照してください)。ここではバージョン 3 を使用しているため、コンパクトな配列を使用します。

リクエスト クラスとレスポンス クラスを?qū)g裝すると、これらのリクエストを送受信できるようになります。この ApiVersions v3 の場合、v2 要求ヘッダーと v0 応答ヘッダーが必要です (生成された ApiMessageType.java を確認してください)。 API キー (18) は、ApiVersionsRequest.json またはプロトコル仕様で見つかります。

git clone git@github.com:apache/kafka.git
git checkout 3.8.0

このコードを?qū)g行すると、コンソールに応答ヘッダーとメッセージが表示されます。おめでとうございます。Kafka ブローカーとの正しいネットワーク交換が実行されました!

_unknownTaggedFields に 3 つのタグ付きフィールドが配置されていることがわかります。生成された ApiVersionsResponseData クラスの読み取りメソッドと書き込みメソッド、および ApiVersionsResponse.json 內(nèi)のメッセージ定義は、それらを解釈するのに役立ちます。この宿題を検討してみませんか?

---

1. 私の本業(yè)では、オープンソース ライブラリ Kio を開発しました。これにより、Python から任意の Kafka API 呼び出しを簡単に実行できるようになります。シリアル化/逆シリアル化コードは、Kafka 自體と同様に、JSON プロトコル定義から生成されます。生成されたコードは、実際の Java Kafka コードに対するプロパティ テストなど、厳密にテストされます。??

2. または、必要に応じて「メッセージ」: 一部のスキーマは API 用ではありませんが、例:ディスク上のデータの場合。??

3. read_exact 関數(shù)には、基礎(chǔ)となるバッファがすでにメモリ內(nèi)にあるときにデータが複製されるという欠點があります。ただし、教育目的にはその方が便利です。??

4. これを修正するために PR を作成しました。??

5. 繰り返しになりますが、これを修正するために PR を作成しました。??

以上がKafka プロトコル実踐ガイドの詳細內(nèi)容です。詳細については、PHP 中國語 Web サイトの他の関連記事を參照してください。