Kafka メッセージキー設計: パーティション配分と順序保証の実務指針

Kafka の順序保証はトピック全体ではなくパーティション単位で成立する。したがって、メッセージキーの設計は配分だけでなく順序保証のスコープを決める中核となる。

本稿では、公式ドキュメントの挙動を前提に、デフォルトパーティショナーの動作、ホットパーティション回避、シリアライズの安定性、パーティション数変更の影響を試験観点と実務観点の両面から整理する。

順序保証とパーティションの基本

Kafka が提供する順序保証は「同一パーティション内の書き込み順序」に限られる。したがって、同じキーを常に同じパーティションに送る設計であれば、そのキーのメッセージは順序を維持できる。一方で、異なるパーティションを横断した全順序は保証されない。

プロデューサは1パーティションに対して送信順序を維持する。ただしリトライや同時送信数の設定次第で乱れうる。コンシューマグループでは1パーティションは同時に1コンシューマスレッドにだけ割り当てられるため、並列度の最小単位もパーティション数に一致する。

• 順序保証の単位は「パーティション」= 並列処理の単位でもある
• 同一キーが同一パーティションに行く設計で、キー単位の順序を維持できる
• トピック全体の順序は設計上保証されない。集計やジョインではタイムスタンプやウィンドウを併用する

メッセージキーとデフォルトパーティショナーの動作

Kafka の標準的なプロデューサは、キーが設定されている場合、キーをシリアライズしたバイト列にハッシュを取り、その結果をパーティション数で割った余りで配分する。同じキーのバイト列であれば常に同じパーティションに送られる。

キーが null の場合は、ラウンドロビンに近い戦略（バッチ効率を考慮した粘着的な配分）でパーティションに分散される。null キーではキー単位の順序保証は得られないため、順序が必要な場合は必ずキーを付与する。

• ハッシュの入力は「キーのシリアライズ後のバイト列」。同じ論理キーでもシリアライザを変えると配分が変わりうる
• パーティション数を変えるとハッシュ結果の剰余が変わり、既存のキー→パーティション対応は崩れる
• キーが必要な要件例: ログコンパクション、ユーザー単位の順序保証、同一エンティティの集約

キーによる配分と順序保証の流れ

アンチパターンとホットパーティション対策

偏ったキー分布は単一パーティションへの集中を招き、遅延やスループット低下につながる。キー設計時は順序のスコープとスケーリングの両立を明確にする必要がある。

完全なキー順序を維持しつつスループットを上げたい場合は、パーティション数を増やしつつプロデューサ・コンシューマの並列度を合わせるのが基本路線。順序を一部緩和できる場合はソルトや合成キーでスケールを取りに行く。

• ホットキーが存在する場合は、順序要件を再確認（ユーザー単位か、セッション単位か）
• ソルトは順序を壊す。必要な範囲だけに限定し、必要ならコンシューマ側で再整列を行う
• 合成キーは順序スコープを明示する手段（例: userId#sessionId ならセッション内順序）

シリアライザとキーの安定性

デフォルトパーティショナーは、キーオブジェクトそのものではなく、シリアライズ後のバイト列をハッシュする。つまり、論理的に同じキーでも、シリアライザやエンコーディングが変わればパーティション配分が変わる可能性がある。

複合キーを文字列連結で表現する場合は、区切り文字や正規化ルールを固定し、将来的な変更でバイト列が変わらないようにする。Avro/Protobuf のバイナリ表現を使う場合も、フィールド順やデフォルト値、可変長エンコーディングの影響を理解して安定化させる。

• キーの正規化ルール（大文字小文字、ゼロ埋め、区切り）を明文化
• シリアライザの変更は配分変更を招くため、移行時はローリングで二重送信やシャドウ検証を用意
• Schema Registry を使う場合でも、キー側スキーマの変更は慎重に（ハッシュ対象はあくまでバイト列）

パーティション数の増減と再パーティショニングの影響

Kafka の標準ハッシュは単純な剰余演算を使うため、パーティション数を変更するとキー→パーティションの対応関係が広範に変わる。これにより、同一キーが別パーティションに移動し、コンシューマ側の状態や順序前提に影響が出る。

Kafka Streams や ksqlDB では、キー変更やシャッフルが必要な演算で自動的に再パーティション用トピックが使われる。外部のプロデューサ設計でも、パーティション数変更を前提にした可観測性（遅延・スループット・ホットパーティション検知）を準備しておくと安全。

• パーティションを増やすとキーの散らばりが変化し、キャッシュやステートストアのヒット率に影響
• 安定ハッシュを使うカスタムパーティショナーで移動割合を抑制する設計もあるが、公式のデフォルトは剰余ベース
• ログコンパクションの観点では、キーが変わらない限り機能は維持されるが、パーティション境界変更で読み出しの並列度やレイテンシが変化しうる

実装パターンと運用チェックリスト

順序と重複制御を両立する場合、プロデューサ側では冪等送信を有効化し、acks=all と組み合わせる。in-flight リクエスト数は冪等時の上限内に収め、リトライ時の順序崩れを回避する。コンシューマ側は1パーティション1スレッド処理を守り、必要なら整列用のバッファとタイムアウトを設ける。

ホットパーティション検知のために、キー分布のメトリクス（partitionごとのレコード数・レイテンシ）を可視化する。スキューが閾値を超えたら、要件に応じて合成キー化、限定的ソルト、あるいはパーティション数の増加を検討する。

• プロデューサ: enable.idempotence=true, acks=all, 適切な delivery.timeout.ms
• 順序重視: max.in.flight.requests.per.connection を小さめに（冪等有効時の安全範囲内）
• コンシューマ: パーティション専有スレッド設計。外部I/Oは非同期化して待ち時間を短縮

Java: カスタムパーティショナーの雛形（安定ハッシュの例）

package example;

import java.util.Map;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.utils.Utils;

public class StableHashPartitioner implements Partitioner {
  private int virtualNodes = 64; // 調整可能。増やすと分散が滑らかに

  @Override
  public void configure(Map<String, ?> configs) {
    Object v = configs.get("stable.partitioner.vnodes");
    if (v instanceof String) {
      try { virtualNodes = Integer.parseInt((String) v); } catch (NumberFormatException ignore) {}
    }
  }

  @Override
  public int partition(String topic, Object keyObj, byte[] keyBytes,
                       Object value, byte[] valueBytes, Cluster cluster) {
    int partitionCount = cluster.partitionsForTopic(topic).size();
    if (keyBytes == null || partitionCount <= 0) {
      // フォールバック: デフォルトの粘着分配に近い動きはしない。単純剰余にする
      return 0;
    }
    // 簡易な HRW/Rendezvous 風: 複数の仮想ノードに対してスコアを計算し、最大のものを選ぶ
    int selected = 0;
    long bestScore = Long.MIN_VALUE;
    for (int p = 0; p < partitionCount; p++) {
      long score = 0L;
      for (int v = 0; v < virtualNodes; v++) {
        // Utils.murmur2 は Kafka が内部で使う 32bit murmur2
        int h1 = Utils.murmur2(keyBytes);
        int h2 = Utils.murmur2((topic + ":" + p + ":" + v).getBytes());
        long combined = ((long)h1 << 32) ^ (h2 & 0xffffffffL);
        score = Math.max(score, combined);
      }
      if (score > bestScore) { bestScore = score; selected = p; }
    }
    return selected;
  }

  @Override
  public void close() {}
}

問題で確認

よくある質問