Kafka プロデューサーが send() を呼び出した後、何が起こるのでしょうか?
シンプルな問いに聞こえます。プロデューサーはレコードを作成し、それを Kafka に送信し、確認応答を受け取ります。
しかし、これらのステップの間には、実にさまざまなことが起こり得ます。
プロデューサーはさらなるレコードを待ってバッチを作成するかもしれません。パーティションを選ばなければなりません。バッチを圧縮するかもしれません。ブローカーが遅ければ、プロデューサーはバッファ領域を使い果たすかもしれません。リクエストが失敗すれば、リトライするかもしれません。最初のリクエストがすでに書き込まれた後にリトライが成功すれば、Kafka は同じバッチを 2 度受け取ることになるかもしれません。
Kafka シミュレータ v1.1 は、こうしたプロデューサーのセマンティクスに焦点を当てています。
今回のリリースでは 10 個の新しいシナリオを追加し、カリキュラム全体は 125 シナリオ中 23 個になりました。
v1.1 には何が含まれているのか?
10 個のシナリオのうち 8 個は、プロデューサーに焦点を当てています。残りの 2 個はログ構造のトラックを拡張するもので、ハイウォーターマークまでの読み取りと、ログ開始オフセットを扱います。
プロデューサーのシナリオは、次の 5 つの領域を扱います。
linger.msとbatch.sizeによるバッチング、- プロデューサーのパーティショナー、
- 圧縮、
buffer.memoryとmax.block.msによるバックプレッシャー、- リトライと冪等性。
それぞれを見ていきましょう。
linger.ms と batch.size によるバッチング
Kafka プロデューサーは、必ずしもすべてのレコードを個別のリクエストで送信するわけではありません。
同じパーティションに割り当てられたレコードは、1 つのバッチにまとめられます。プロデューサーは、バッチがいっぱいになったとき、または設定された待機時間が経過したときに、そのバッチを送信します。
ここで 2 つのプロデューサー設定が重要になります。
batch.sizeはバッチのサイズを制限し、linger.msは送信前に追加のレコードを待つことを許可します。
これはレイテンシとスループットの間のトレードオフです。
大きなバッチは通常、リクエスト数が少なくなり、圧縮も効きやすくなります。しかし、バッチがいっぱいになるのを待つことは、個々のレコードのレイテンシを増やすかもしれません。
新しいシナリオは、この振る舞いを可視化します。レコードがいつバッチに入るのか、何がバッチの送信を引き起こすのか、そして設定の変更がクライアントの生成するリクエストにどう影響するのかを見られます。
プロデューサーのパーティショナー
プロデューサーがパーティションごとのバッチを作成する前に、まずパーティションを選ばなければなりません。
キーを持つレコードでは、キーを使ってパーティションが選ばれます。したがって、同じキーを持つレコードは通常、同じパーティションに割り当てられます。
キーを持たないレコードでは、振る舞いが異なります。プロデューサーはキーからパーティションを計算する代わりに、パーティショニング戦略を使ってレコードをグループ化できます。
シミュレータは、両方のケースを一歩ずつ示します。
- プロデューサーがレコードを受け取り、
- パーティショナーがパーティションを選び、
- レコードが対応するバッチに追加され、
- バッチがそのパーティションのリーダーへ送信されます。
これにより、パーティショニングが順序とバッチングの両方に影響する理由が見えやすくなります。
圧縮はバッチ単位で行われる
Kafka プロデューサーの圧縮は、すべてのレコードに個別に適用されるのではなく、レコードバッチに対して適用されます。
この区別は重要です。
圧縮は通常、バッチに複数の似たレコードが含まれているときにより効果的です。小さなレコードが 1 つだけ含まれるバッチは、圧縮による恩恵がほとんどないかもしれません。反復的なデータを含む大きなバッチは、はるかに効率よく圧縮できるかもしれません。
その結果、圧縮はバッチングの設定と結びついています。batch.size や linger.ms のような設定は、リクエストの頻度だけでなく、結果として生成されるバッチの形や圧縮のしやすさにも影響し得ます。
新しいシナリオは、圧縮前と圧縮後のバッチサイズを示し、この関係を明確にします。
buffer.memory と max.block.ms によるバックプレッシャー
アプリケーションが、Kafka が受け入れられるよりも速くレコードを生成したとき、何が起こるのでしょうか?
プロデューサーは、未送信のレコードを無制限にメモリに保持することはできません。buffer.memory で設定される、上限のあるバッファを使います。
バッファに空き領域がなくなると、send() の呼び出しは、メモリが解放されるのを待つ間ブロックされることがあります。
ただし、いつまでも待つわけではありません。最大のブロック時間は max.block.ms で制御されます。
これにより、2 つの結末が考えられます。
- 十分なデータが送信されて確認応答を受け取り、バッファ領域が解放される、
- タイムアウトまでに空き領域ができず、送信が失敗する。
シミュレータは、プロデューサーのバッファを直接追跡します。レコードが追加されたときにメモリが確保され、リクエストが完了したときに解放される様子を見られます。
これは、遅いブローカーやスループットのクォータと組み合わせたときに特に役立ちます。プロデューサーのバックプレッシャーを抽象的な設定の問題として扱うのではなく、バッファが埋まっていく様子と、どの送信がブロックされるのかを正確に観察できます。
リトライと重複の問題
プロデューサーがバッチをブローカーに送信するとしましょう。
ブローカーはバッチの書き込みに成功しますが、その確認応答がプロデューサーに届く前に失われてしまいます。
プロデューサーの視点からは、リクエストは失敗しました。ブローカーがデータを書き込んだかどうかを知る術はありません。したがって、リトライするのは妥当です。
しかし、最初のリクエストが正常に書き込まれていた場合、リトライは同じバッチをもう一度追記してしまうかもしれません。
これが、曖昧な障害によって引き起こされる重複の問題です。
リトライのシナリオは、両方のリクエストを示します。
- 元のリクエストがブローカーに届き、
- 確認応答が失われ、
- プロデューサーがバッチをリトライし、
- ブローカーが同じレコードを 2 度目に受け取ります。
冪等性がなければ、両方のコピーが追記されてしまう可能性があります。
冪等プロデューサーのセマンティクス
冪等プロデューサーを使うと、Kafka はリトライされたバッチを識別できるようになります。
Kafka はプロデューサーにアイデンティティを割り当て、パーティションごとに独立してシーケンス番号を追跡します。ブローカーがすでに処理済みのシーケンス番号でバッチがリトライされたとき、ブローカーはレコードを再び追記することなく確認応答を返せます。
重要なのは、リトライと本物の重複の区別です。
2 つのレコードがまったく同じキーと値を持っていても、それらはアプリケーションが意図的に送信した別々のレコードであり得ます。Kafka がそのようなレコードを取り除いてはなりません。
冪等性はプロデューサープロトコルのリトライを重複排除します。内容に基づいてアプリケーションのデータを重複排除するわけではありません。
新しいシナリオは、次を示します。
- プロデューサー ID、
- パーティションのシーケンス番号、
- 冪等リトライの成功、
- 本物の重複レコード、
- 順序外れのシーケンスの検出。
なぜ max.in.flight.requests.per.connection が重要なのか?
プロデューサーは、複数のリクエストを同時にレスポンス待ちの状態にできます。
これはスループットを向上させますが、順序に関する潜在的な問題も生み出します。
プロデューサーがバッチ A を送り、続いてバッチ B を送るとしましょう。バッチ A が一時的に失敗し、バッチ B は成功します。もし B がすでに書き込まれた後に A がリトライされると、レコードが異なる順序で追記されてしまうかもしれません。
冪等性は、シーケンス番号を使って不正な順序を検出します。
シミュレータは、これを明示的に示します。ブローカーが期待するシーケンス番号、到着したバッチが持つ番号、そしてブローカーが順序外れのリクエストを拒否する地点を見られます。
プロデューサーの内部状態を調べる
新しいシナリオは、シミュレータのインスペクターを大いに活用しています。インスペクターは、プロデューサーとパーティションの内部状態を公開します。含まれるのは次のとおりです。
- リトライカウンター、
- プロデューサーのシーケンス番号、
- ブローカーのシーケンス番号、
- バッファの確保状況。
通常、これらの値はプロデューサーのメトリクス、ログ、そしてプロトコルの振る舞いの背後に隠れています。シミュレータでは、シナリオを進めたり巻き戻したりしながら、これらを直接調べられます。
プロデューサーモデルの修正
新しいシナリオを追加するにあたって、既存のプロデューサーモデルをより正確なものにする必要もありました。
リトライはバッチ単位でカウントされる
Kafka は、レコードのバッチを含む produce リクエストをリトライします。
したがって、失敗したリクエストは、その中の各レコードごとに 1 回ではなく、バッチ全体に対して 1 回のリトライを消費すべきです。
シミュレータはこのモデルに従うようになりました。失敗したバッチのすべてのレコードが再送される一方で、リトライカウンターは 1 回だけ進みます。
バッファメモリが正確に解放される
プロデューサーのバッファメモリは、確認応答が届くたびに解放されるようになりました。
これにより、確認応答済みのレコードがすでにメモリを解放しているはずなのに、シミュレータがバックプレッシャーを報告してしまうケースが修正されました。
同じことが max.block.ms のタイムアウトにも当てはまります。ブロックされた送信がタイムアウトすると、その確保していたバッファがクリーンアップされます。
acks=all はパーティション単位で追跡される
Kafka の耐久性は、パーティションごとに独立して定義されます。
したがって、複数のパーティションに書き込むプロデューサーは、確認応答の要件を満たすために正しいパーティションを待たなければなりません。別のパーティションでの進捗が、無関係なリクエストの確認応答になってはなりません。
シミュレータは、この状態をパーティションとトピックごとに個別に追跡するようになりました。
プロデューサーのリセットで冪等性の状態がクリアされる
ブローカーは、冪等プロデューサーに紐づく重複排除の状態を保存しています。
シミュレートされたプロデューサーが削除またはリセットされたとき、この状態も取り除かれなければなりません。そうでなければ、新しく作成されたプロデューサーのレコードが、古いリトライとして誤って分類されてしまうかもしれません。
リセットの挙動は、対応するブローカー側の状態をクリアするようになりました。
巻き戻しでリトライの状態が復元される
シナリオは、一歩ずつ進めたり巻き戻したりできます。
巻き戻しは、各シミュレーションラウンドで使われたリトライカウンターを再構築するようになりました。そのため、シナリオを前に進めているか後ろに戻しているかにかかわらず、見えているプロデューサーの状態は一貫します。
プレイグラウンドの改善
今回のリリースには、プロデューサーモデル以外にもいくつかの修正が含まれています。
コンシューマーグループのリバランス
プレイグラウンドのコンシューマーグループには、正確性に関するいくつかの修正が入りました。
- スティッキーアサイナーは、ゼロから再割り当てするのではなく、パーティションの移動を最小限に抑えるようになりました、
session.timeout.msを超えて離脱したままのスタティックメンバーは、他のメンバーと同じように除外されます、- トピックのパーティション数を変更すると、メンバーの状態を乱すことなく適切なリバランスがトリガーされるようになりました、
- すでに生存していないメンバーにパーティションが割り当てられることはありません。
トピック設定とコンパクション
Kafka がトピックのパーティション数を減らすことは決してありません。縮小はデータを削除し、キーに基づくパーティショニングを壊してしまうため、kafka-topics --alter は InvalidPartitionsException でこれを拒否します。
プレイグラウンドも同じルールに従うようになりました。パーティション数の削減は、黙ってパーティションを取り除くのではなく、学習用のエラーとともに拒否されます。
コンパクション対象のトピックの振る舞いも、より正確になりました。
- トゥームストーンは、それ自身の
delete.retention.msの猶予期間が切れるまで残ります、 - キーのないレコードは、コンパクション対象のトピックに生成できません、
compact,deleteでは、時間ベースの保持がコンパクションとは独立して、古くなったレコードを回収するようになりました。
共有リンクがシナリオ全体を保持する
プレイグラウンドは、シナリオを他の人と共有できるように、アクションを URL にシリアライズします。
コーデックは、パーティション数の更新を含むすべての設定変更を含めるようになりました。後のリリースパックのクラスタータイプを要求するリンクは、使用できないクラスターを表示する代わりに、サポートされているトポロジーにフォールバックするようになりました。
したがって、共有リンクを開くと、もともと構築されたのと同じシナリオが再現されるはずです。
操作の修正
いくつかの小さなプレイグラウンドの問題も修正されました。
- エンティティを削除して再作成した後、エンティティの識別子が衝突しなくなりました、
- 幅の広いブローカーグリッドでもプロデューサーの配置が正しく機能します。
セキュリティヘッダー
サイトは Content Security Policy と Permissions Policy を送信するようになりました。
なぜプロデューサーから始めるのか?
Kafka の保証は、しばしばブローカーの視点から議論されます。レプリケーション、ISR メンバーシップ、ハイウォーターマーク、そしてリーダーのフェイルオーバーです。
しかし、アプリケーションが観測する保証は、プロデューサーにも依存します。
どの確認応答モードを使うのか? リトライするのか? 冪等性は有効か? いくつのリクエストを同時に処理できるのか? ローカルのバッファがいっぱいになると何が起こるのか?
これらの設定が、書き込みが速いか、耐久性があるか、順序が保たれるか、重複するか、ブロックされるか、あるいは拒否されるかを決定づけます。
だからこそ、最初の大きな Kafka シミュレータのシナリオパックはプロデューサーに焦点を当てています。
バッチング、確認応答、リトライ、そして冪等性が可視化されれば、後のレプリケーション、耐久性、トランザクションを扱うシナリオも、より明確な土台の上に立てるようになります。
Kafka シミュレータを開き、プロデューサートラックを選んで、障害のシナリオを一歩ずつ試してみてください。