ApsaraMQ for Kafka:クライアントパラメーターの設定

acks

送信が成功したと見なす前に、プロデューサーがブローカーから要求する確認応答 (ACK) の数を制御します。

推奨値：厳密な耐久性よりもスループットを優先するほとんどのワークロードでは 1 です。

retries

失敗した送信をプロデューサーがリトライする最大回数。値を大きくすると、プロデューサーはリーダー選出など一時的なブローカーの障害から回復しやすくなります。retry.backoff.ms と組み合わせて、リトライのペースを制御します。

retry.backoff.ms

送信リトライ間の遅延時間。値が低すぎると、ブローカーのフェールオーバー中にリトライストームが発生する可能性があります。

batch.size

パーティションごとの単一バッチの最大サイズ。バッチがこのサイズに達すると、プロデューサーは即座に送信します。

ほとんどのワークロードでは、デフォルト値の 16384 を維持してください。値を小さくすると、ネットワークリクエストの数が増加し、スループットが低下します。batch.size を増やす場合は、buffer.memory がより大きなバッチを収容できる十分な大きさであることを確認してください。

linger.ms

プロデューサーがバッチがいっぱいになるのを待ってから送信するまでの最大時間。これは TCP の Nagle アルゴリズムのように機能します。バッチが batch.size に達すると、linger タイマーに関係なく即座に送信されます。linger.ms が経過した時点でバッチがまだ batch.size 未満の場合、プロデューサーは蓄積されたものを送信します。

linger.ms を高くすると、メッセージごとのレイテンシーを犠牲にして、バッチ処理の効率とスループットが向上します。

buffer.memory

プロデューサーがすべてのパーティションにわたる未送信レコードのバッファリングのために割り当てる合計メモリ。このプールが使い果たされると、max.block.ms に応じて send() はブロックされるか、例外をスローします。バッファーが小さすぎると、メモリ割り当てが遅くなったり、スループットが低下したり、送信がタイムアウトしたりする原因になります。

単位：バイト。デフォルト：33554432 (32 MB)。

サイジングの数式：

buffer.memory >= batch.size x number_of_partitions x 2

たとえば、batch.size=16384 でパーティションが 50 個の場合：

16384 x 50 x 2 = 1,638,400 バイト (最小約 1.6 MB)

スループット向上のために batch.size を増やす場合は、buffer.memory も比例してスケールしてください。

partitioner.class

プロデューサーがレコードをパーティションに割り当てる方法を決定します。スティッキーパーティショニング戦略は、次のパーティションに移動する前に 1 つのパーティションのバッチを埋めることで、不完全なバッチの数を減らします。

ご利用のプロデューサークライアントがバージョン 2.4 より前の場合は、バッチ処理の効率を向上させるために、スティッキーパーティショナーを明示的に設定してください。

fetch.min.bytes

ブローカーがフェッチ応答を返す前に蓄積する最小データ量。値を大きくすると、フェッチ頻度とブローカーの CPU オーバーヘッドが減少し、スループットが向上しますが、エンドツーエンドのメッセージレイテンシーは増加します。単位：バイト。

この値を設定する前に、ご利用のプロデューサーのメッセージレートを評価してください。メッセージの到着が遅い場合、fetch.min.bytes を大きくすると不要な遅延が発生します。

fetch.max.wait.ms

ブローカーが応答を返す前に fetch.min.bytes を蓄積するために待機する最大時間。単位：ミリ秒。

ストレージタイプによって動作が異なります：

• ローカルストレージ：ブローカーは、fetch.min.bytes に達するか fetch.max.wait.ms が経過するかのいずれか早い方まで待機します。

• クラウドストレージ：ブローカーは、fetch.min.bytes に関係なく、新しいデータが到着するとすぐにレスポンスを返します。

max.partition.fetch.bytes

ブローカーが 1 回のフェッチでパーティションごとに返す最大データ量。単位：バイト。

session.timeout.ms

ブローカーがコンシューマーを停止したと見なし、リバランスをトリガーするまでのハートビート間の最大時間。

max.poll.records

1 回の poll() 呼び出しで返されるレコードの最大数。コンシューマーが次の poll() のデッドラインまでにこの数のレコードを処理できない場合、ブローカーはコンシューマーを停止したと見なし、リバランスをトリガーします。

サイジングの数式：

max.poll.records < messages_per_thread_per_second x consumer_threads x session_timeout_seconds

たとえば、スレッドあたり 500 msg/s、4 スレッド、セッションタイムアウト 45 秒の場合：

500 x 4 x 45 = 90,000

セッションがタイムアウトする前にコンシューマーが常に処理を完了できるように、max.poll.records をこの値より低く設定してください。

max.poll.interval.ms

ブローカーがコンシューマーをグループから削除するまでの、連続した poll() 呼び出し間の最大間隔。このパラメーターは、ハートビートが別のスレッドで実行される Java クライアント 0.10.1 以降にのみ適用されます。

サイジングの数式：

max.poll.interval.ms > time_per_record x max.poll.records

ほとんどの場合、デフォルトの 300000 (5 分) で十分です。処理ロジックが非常に遅い場合にのみ、この値を増やしてください。

enable.auto.commit

コンシューマーが一定間隔でオフセットを自動的にコミットするかどうかを制御します。

auto.commit.interval.ms

enable.auto.commit が true の場合の自動オフセットコミットの間隔。

間隔を短くすると、クラッシュ後の重複メッセージのウィンドウは減少しますが、ブローカーへのコミットリクエストの数は増加します。

auto.offset.reset

コンシューマーにコミット済みオフセットがない場合、またはコミット済みオフセットが無効な場合 (例：保持ポリシーによりオフセットが削除された場合) の動作を決定します。

推奨値：latest。earliest を使用すると、無効なオフセットが検出されるたびに、コンシューマーは保持されているすべてのメッセージを再処理します。これにより、重複処理やコンシューマーの遅延の急増につながる可能性があります。

アプリケーションがオフセット管理を手動で行う場合 (例：外部データベースにオフセットを保存する場合) は、これを none に設定してください。