Kafka クライアントのベストプラクティス

プロデューサーの可用性

• ブローカーのフェイルオーバー中に失敗したメッセージの送信を再試行するようにプロデューサーに指示retriesするように を設定します。ほとんどのユースケースでは、整数最大値または同様の最大値の値をお勧めします。これを怠ると、Kafka の高可用性が損なわれます。

• を設定delivery.timeout.msして、メッセージを送信してからブローカーから確認応答を受け取るまでの合計時間の上限を指定します。これは、メッセージの有効期限に関するビジネス要件を反映している必要があります。フェイルオーバーオペレーションを完了するために十分な再試行ができるように、時間制限を十分高く設定します。ほとんどのユースケースでは、60 秒以上の値を使用することをお勧めします。

• 再送信が試行されるまでに 1 つのリクエストが待機する最大数request.timeout.msに設定します。ほとんどのユースケースでは、10 秒以上の値を使用することをお勧めします。

• 再試行間の遅延を設定して、再試行の嵐や可用性への影響を回避するretry.backoff.msように を設定します。ほとんどのユースケースでは、最小値を 200 ミリ秒にすることをお勧めします。

• 高い耐久性を設定するacks=allように を設定します。これは、 RF=3 および のサーバー側の設定と一致しmin.isr=2ている必要があります。これにより、 内のすべてのパーティションが書き込みISRを承認します。オフラインの 1 つのブローカーでは、これは min.isr、つまり です2。

コンシューマーの可用性

• 新しいコンシューマーグループまたは再作成されたコンシューマーグループについては、latest最初に auto.offset.resetを に設定します。これにより、トピック全体を消費してクラスター負荷を追加するリスクを回避できます。

• auto.commit.interval.ms を使用する場合に を設定しますenable.auto.commit。ほとんどのユースケースでは、追加の負荷のリスクを避けるために、最小値を 5 秒にすることをお勧めします。

• 一時的なエラーを処理するために、コンシューマーのメッセージ処理コード内に例外処理を実装します。例えば、サーキットブレーカーや指数バックオフのあるスリープなどです。そうしないと、アプリケーションがクラッシュし、過剰なリバランシングが発生する可能性があります。

• トランザクションメッセージの読み取り方法を制御するisolation.levelように を設定します。

  デフォルトではread_uncommitted、常に暗黙的に を設定することをお勧めします。これは、一部のクライアント実装には欠落しています。

  階層型ストレージread_uncommittedを使用する場合は、 の値を使用することをお勧めします。

• 最も近いレプリカリードを使用するclient.rackように を設定します。ネットワークトラフィックのコストとレイテンシーを最小限に抑えるaz id ために、 を に設定することをお勧めします。「ラック認識 による Amazon MSKコンシューマーのネットワークトラフィックコストの削減」を参照してください。

コンシューマーの再調整

• 実装された起動ジッターを含め、アプリケーションの起動時間より大きい値session.timeout.msに設定します。ほとんどのユースケースでは、60 秒の値をお勧めします。

• グループコーディネーターがコンシューマーを正常と見なす方法を微調整heartbeat.interval.msするように を設定します。ほとんどのユースケースでは、10 秒の値をお勧めします。

• コンシューマーがグループを離れるタイミングを特定するためにセッションタイムアウトに依存するのではなくSIGTERM、 でコンシューマーをクリーンに閉じるように、アプリケーションにシャットダウンフックを設定します。Kstream アプリケーションは の値internal.leave.group.on.closeに設定できますtrue。

• コンシューマーグループ内の個別の値group.instance.idに設定します。ホスト名、タスク ID、または pod-id が理想的です。トラブルシューティング中に、より決定的な動作とクライアントとサーバーのログの相関関係を向上させるために、常にこれを設定することをお勧めします。

• 平均デプロイ時間に沿った値group.initial.rebalance.delay.msに設定します。これにより、デプロイ中の継続的な再調整が停止します。

• スティッキーアサイザーを使用するpartition.assignment.strategyように を設定します。StickyAssignor または をお勧めしますCooperativeStickyAssignor。

プロデューサーのパフォーマンス

• プロデューサーがバッチがいっぱいになるまで待機する時間を制御するlinger.msように を設定します。バッチが小さいほど、一度により多くのスレッドと I/O オペレーションに変換されるため、Kafka には計算コストがかかります。以下の値をお勧めします。

  低レイテンシーを含むすべてのユースケースの最小値は 5 ミリ秒です。

  ほとんどのユースケースでは、25 ミリ秒の値を大きくすることをお勧めします。

  低レイテンシーのユースケースでは、ゼロの値を使用することはお勧めしません。（値が 0 の場合、通常、IO オーバーヘッドが原因でレイテンシーが発生します）。

• クラスターに送信されるバッチサイズを制御するbatch.sizeように を設定します。これを 64KB または 128KB の値に増やすことをお勧めします。

• より大きなバッチサイズを使用する場合buffer.memoryに を設定します。ほとんどのユースケースでは、664MB の値をお勧めします。

• バイトの受信に使用されるTCPバッファを制御するsend.buffer.bytesように を設定します。高レイテンシーネットワークでプロデューサーを実行するときに OS がこのバッファを管理できるようにするには、-1 の値をお勧めします。

• バッチの圧縮を制御するには、 compression.type を設定します。lz4 または zstd のいずれかで、高レイテンシーネットワークでプロデューサーを実行することをお勧めします。

コンシューマーのパフォーマンス

• fetch.min.bytes を設定して、フェッチ数とクラスター負荷を減らすために有効になる最小フェッチサイズを制御します。

  すべてのユースケースでは、32 バイトの最小値を使用することをお勧めします。

  ほとんどのユースケースでは、128 バイトの値を大きくすることをお勧めします。

• fetch.max.wait.ms を設定して、fetch.min.bytes が無視されるまでコンシューマーが待機する時間を決定します。ほとんどのユースケースでは、1000 ミリ秒の値をお勧めします。

• コンシューマーの数は、パーティションの数と少なくとも同じにすることをお勧めします。

• バイトの受信に使用されるTCPバッファを制御するreceive.buffer.bytesように を設定します。高レイテンシーネットワークでコンシューマーを実行するときに、OS がこのバッファを管理できるようにするには、-1 の値をお勧めします。

クライアント接続

接続ライフサイクルには、Kafka クラスターの計算コストとメモリコストがかかります。一度に作成された接続が多すぎると、Kafka クラスターの可用性に影響を与える可能性のある負荷が発生します。この可用性への影響により、アプリケーションがさらに多くの接続を作成することがよくあり、カスケード障害が発生し、完全な停止が発生する可能性があります。合理的なレートで作成された場合、多数の接続を実現できます。

高い接続作成率を管理するには、次の緩和策をお勧めします。

• アプリケーションデプロイメカニズムがすべてのプロデューサー/コンシューマーを一度に再起動するのではなく、できれば小さなバッチで再起動していることを確認します。

• アプリケーションレイヤーでは、デベロッパーは、管理者クライアント、プロデューサークライアント、またはコンシューマークライアントを作成する前に、ランダムジッター (ランダムスリープ) が実行されていることを確認する必要があります。

• ではSIGTERM、接続を閉じるときにランダムなスリープを実行して、すべての Kafka クライアントが同時に閉じられないようにする必要があります。ランダムなスリープは、 SIGKILLが発生する前にタイムアウト内である必要があります。

  例 A (Java)

  sleepInSeconds(randomNumberBetweenOneAndX);
                          this.kafkaProducer = new KafkaProducer<>(this.props);

  例 B (Java)

  Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      sleepInSeconds(randomNumberBetweenOneAndTwentyFive);
      kafkaProducer.close(Duration.ofSeconds(5));
  });

• アプリケーションレイヤーでは、デベロッパーは、クライアントがアプリケーションごとに 1 回だけシングルトンパターンで作成されることを確認する必要があります。例えば、lambda を使用する場合、クライアントはメソッドハンドラーではなく、グローバルスコープで作成する必要があります。

• 接続数は、安定することを目標にモニタリングすることをお勧めします。デプロイ時とブローカーのフェイルオーバー中は、接続creation/close/shiftは正常です。

プロデューサーメトリクス

• バイトレート

• record-send-rate

• records-per-request-avg

• acks-latency-avg

• request-latency-avg

• request-latency-max

• record-error-rate

• record-retry-rate

• エラーレート

コンシューマーメトリクス

• records-consumed-rate

• bytes-consumed-rate

• フェッチレート

• records-lag-max

• record-error-rate

• fetch-error-rate

• ポーリングレート

• rebalance-latency-avg

• コミットレート

一般的なメトリクス

• connection-close-rate

• connection-creation-rate

• 接続数