Kafkaのパーティションキー設計：メッセージの順序と負荷分散

1. 概要

Kafkaではpartition keyが、メッセージがどのパーティションに入るかを決定します。それは同時に2つのことも決定します：どのメッセージが正しい順序で保持されるのか、そして負荷がコンシューマー間でどの程度均等に分散されるのかです。

キーを誤って選ぶと、結果は通常以下の2つの方向のいずれかに現れます。1つ目の方向は、同じ注文のイベントが乱雑に処理されることです。2つ目の方向は、1つのパーティション「ホット」がほぼすべてのトラフィックを引き受ける一方で、他のパーティションは使用されていないということです。

この記事を読み終わったら、Kafkaがキーをパーティションにどのようにマッピングするかを理解します。また、順序要件に基づいてキーを選ぶ方法、ホットキーを扱う方法、Spring Boot 3.xでプロデューサー/コンシューマーを設定して誤って順序を破壊しないようにする方法も知ることになります。

2. Kafkaはパーティションをどのように選ぶか

2.1. キーが`null`以外の場合：ハッシュ化して剰余を計算

メッセージがキーを持つ場合、プロデューサーは次の公式に従ってパーティションを計算します：

partition = toPositive(murmur2(keyBytes)) % numPartitions

kafka-clientsに含まれるユーティリティ関数を使ってこのロジックを正確に再現できます：

import org.apache.kafka.common.utils.Utils;
import java.nio.charset.StandardCharsets;

public final class PartitionCalculator {

    private PartitionCalculator() {}

    public static int partitionFor(String key, int numPartitions) {
        byte[] keyBytes = key.getBytes(StandardCharsets.UTF_8);
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

    public static void main(String[] args) {
        for (String orderId : new String[]{"order-1001", "order-1002", "order-1001"}) {
            System.out.println(orderId + " -> partition " + partitionFor(orderId, 6));
        }
    }
}

各部分の説明：

キーはStringSerializerによってバイトに変換されます。つまり、UTF-8です。この動作に合わせるため、例ではUTF-8を使用します。
Utils.murmur2はデフォルトパーティショナーが使用するハッシュ関数です。
toPositiveは符号ビットを削除して、結果が負にならないようにします。その後の%演算がパーティションインデックスを返します。

実行結果は以下のような形になります（具体的な数字はキーに依存します）：

order-1001 -> partition 3
order-1002 -> partition 0
order-1001 -> partition 3

重要なポイント：同じキーと同じパーティション数であれば、常に同じパーティションになります。 Kafkaはパーティション内の順序のみを保証します。したがって、メッセージのグループの順序を保つには、それらのメッセージは共通のキーを使用する必要があります。

2.2. キーが`null`の場合：順序の保証なし

Kafka 3.3以降、キーを持たないメッセージは「uniform sticky」メカニズムに従って分散されます。プロデューサーはバッチを1つのパーティションに集約し、次に別のパーティションに移動します。各ブローカーの速度を考慮します。負荷はかなり均等に分散されますが、連続する2つのメッセージは異なるパーティションに存在する可能性があります。その場合、それらの間に順序はありません。

注意： ハッシュはキーの論理的な値ではなく、キーのバイトに依存します。1つのサービスがLongSerializerを使って1001を送信し、別のサービスがStringSerializerを使って"1001"を送信する場合、2つのメッセージは異なるパーティションに入る可能性があります。システム全体のキーシリアライザーを統一してください。

3. Spring Bootでキー付きメッセージを送信する

依存関係を追加します：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

application.ymlでプロデューサーを設定します：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5

各設定の意味：

acks: allとenable.idempotence: trueはリトライが重複レコードを作成しないようにします。
さらに重要なことに、べき等性が有効な場合、ブローカーは順序が狂ったバッチを拒否します。おかげで、パーティション内の順序は保持されます。これはリトライが発生しても同様で、max.in.flightが5に達している場合でも保持されます。

イベントを送信するサービス：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

public record OrderEvent(String orderId, String customerId, String type) {}

@Service
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publish(OrderEvent event) {
        kafkaTemplate.send("order-events", event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex == null) {
                        var md = result.getRecordMetadata();
                        System.out.printf("%s %s -> p%d@%d%n",
                                event.orderId(), event.type(), md.partition(), md.offset());
                    }
                });
    }
}

send(topic, key, value)の2番目のパラメータはキーです。ここではorderIdを使用しています。同じ注文に対して順番にCREATED、PAID、SHIPPEDを送信する場合、出力は次のようになります：

order-1001 CREATED -> p3@120
order-1001 PAID -> p3@121
order-1001 SHIPPED -> p3@122

3つのイベント全てがパーティション3に存在し、オフセットが増加しています。コンシューマーはこの正しい順序で読み取ります。

4. キーの選択：順序要件から始める

最初の質問は「どのキーが最も均等に負荷を分散するか」ではありません。正しい質問は：どのメッセージが順序どおりに処理されるべきか？ キーは、その要件を満たす最小グループの識別子である必要があります。

ビジネス要件	適切なキー	備考
注文のイベントは順序どおりに処理する必要がある	`orderId`	多くのキー、優れた負荷分散
顧客のウォレット残高の加算/減算は順序どおり必要	`customerId`または`walletId`	非常に大きな顧客がいる場合、ホットキーが発生しやすい
アクセスログは順序が不要	`null`	最も均等な負荷分散
製品ごとの在庫	`sku`	「ホット」製品により負荷が傾く可能性がある

付随する2つの原則：

キーが細かいほど、負荷分散が良くなります。 orderIdは数百万の値があるため、均等に分布します。countryCodeには数個の値しかないため、数個のパーティションがほぼすべてを占めます。
キーの値の数はパーティション数よりもはるかに大きいべきです。 キーが10個しかなく、トピックが12個のパーティションを持つ場合、確実にいくつかのパーティションが空のままになります。通常、いくつかのパーティションは複数のキーを受け取り、その中にはホットなものもあります。

5. ホットパーティションと対処方法

キーがcustomerIdであり、大規模な企業の顧客が40%のトラフィックを生成する場合を想定してください。この顧客を含むパーティションはラグが発生します。コンシューマーを追加しても役に立ちません。1つのパーティションは、グループ内の1つのコンシューマーのみが同時に読むことができます。

5.1. ソルティング：ホットキーを複数のバケットに分割

概念は、キーのサフィックスを追加してホットキーを複数のパーティションに分散させることです：

import java.util.Set;

public class OrderKeyStrategy {

    private final Set<String> hotCustomers;
    private final int buckets;

    public OrderKeyStrategy(Set<String> hotCustomers, int buckets) {
        this.hotCustomers = hotCustomers;
        this.buckets = buckets;
    }

    public String keyFor(OrderEvent event) {
        if (!hotCustomers.contains(event.customerId())) {
            return event.customerId();
        }
        int bucket = Math.floorMod(event.orderId().hashCode(), buckets);
        return event.customerId() + "#" + bucket;
    }
}

通常の顧客の場合、キーはcustomerIdのままです。ホット顧客の場合、バケットはorderIdから計算されるため、キーはcust-big#0 … cust-big#7のような形式になります。

理解すべき結果：

同じ注文のすべてのイベントは同じキーを持つため、順序は保持されます。
同じ顧客の異なる注文の間の順序は保証されなくなります。

ソルティングは、ビジネスがその取引を受け入れる場合にのみ使用できます。顧客別の絶対的な順序が必要な場合（例えば、ウォレット残高）、ソルティングはデータの不正確性を引き起こします。

注意： バケットに乱数またはタイムスタンプを使用しないでください。そうすると、同じ注文のイベントが複数のパーティションに散らばり、順序が完全に破壊されます。

5.2. キー戦略のテスト

以下のテストは、2つの主要な動作を確認します：同じ注文は同じパーティション、ホット顧客は複数のパーティションに分散します。

import org.junit.jupiter.api.Test;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.assertj.core.api.Assertions.assertThat;

class OrderKeyStrategyTest {

    private static final int PARTITIONS = 12;
    private final OrderKeyStrategy strategy = new OrderKeyStrategy(Set.of("cust-big"), 8);

    @Test
    void sameOrder_alwaysGoesToSamePartition() {
        var created = new OrderEvent("order-1", "cust-big", "CREATED");
        var paid = new OrderEvent("order-1", "cust-big", "PAID");

        int p1 = PartitionCalculator.partitionFor(strategy.keyFor(created), PARTITIONS);
        int p2 = PartitionCalculator.partitionFor(strategy.keyFor(paid), PARTITIONS);

        assertThat(p1).isEqualTo(p2);
    }

    @Test
    void hotCustomer_isSpreadAcrossPartitions() {
        Set<Integer> partitions = IntStream.range(0, 1_000)
                .mapToObj(i -> new OrderEvent("order-" + i, "cust-big", "CREATED"))
                .map(e -> PartitionCalculator.partitionFor(strategy.keyFor(e), PARTITIONS))
                .collect(Collectors.toSet());

        assertThat(partitions).hasSizeGreaterThan(1);
    }

    @Test
    void normalCustomer_keepsCustomerIdAsKey() {
        var event = new OrderEvent("order-9", "cust-small", "CREATED");
        assertThat(strategy.keyFor(event)).isEqualTo("cust-small");
    }
}

テストはセクション2.1のPartitionCalculatorを使用します。これはプロデューサーの正確な公式を再現します。したがって、ブローカーを設定する必要がなく、メッセージが実際にどのパーティションに入るのかを確認できます。

6. コンシューマー側：順序を自分で破壊しない

プロデューサーが順序を正しく保つだけでは不十分です。コンシューマーも台無しにする可能性があります。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {

    @KafkaListener(topics = "order-events", groupId = "order-service", concurrency = "6")
    public void onEvent(OrderEvent event) {
        // リスナーのスレッド内で順序どおり処理
        System.out.println(Thread.currentThread().getName() + " " + event);
    }
}

concurrency = "6"は6つのコンシューマースレッドを作成します。各パーティションは正確に1つのスレッドに割り当てられるため、パーティション内の順序は保持されます。トピックに6個のパーティションがある場合、各スレッドは1個のパーティションを受け取ります。concurrencyをパーティション数より大きく設定すると、余ったスレッドは何もしません。

コンシューマー側でよく見られる誤り：

メッセージをリスナー内の独立したスレッドプールにプッシュする（executor.submit(...)）。同じ注文の2つのイベントが並行実行される可能性があり、PAIDはCREATEDの前に完了する可能性があります。パーティションのレベルを超えて並列化する必要がある場合は、キーごとに並列化し、各キーは1つの順序付きキューを使用します。各メッセージごとに並列化しないでください。
別のトピックに送信してリトライする（retry topic、non-blocking retry）。失敗したメッセージはメインストリームから分離され、同じキーの後続のメッセージは処理を続けます。厳密な順序が必要なストリームでは、DefaultErrorHandlerとBackOffを使用したブロッキングリトライを使用する方が良いです。この方法はパーティションが一時的に停止することを受け入れます。

7. パーティション数の変更：静かなる落とし穴

公式に% numPartitionsがあるため、パーティションを6から12に増やすと、ほとんどのキーが別のパーティションに移動します：

public class RepartitionDemo {
    public static void main(String[] args) {
        String key = "order-1001";
        System.out.println("6 partitions  -> " + PartitionCalculator.partitionFor(key, 6));
        System.out.println("12 partitions -> " + PartitionCalculator.partitionFor(key, 12));
    }
}

6 partitions  -> 3
12 partitions -> 9

order-1001の古いイベントはパーティション3にあり、処理が完了していない可能性があります。新しいイベントはパーティション9に入ります。異なる2つのコンシューマーが同時に処理する可能性があり、順序が狂います。

リスクを軽減する方法：

最初からパーティション数を過剰に見積もります。 スループット目標を1コンシューマーのスループットで割ったものに、バッファ係数を掛けて計算します。
パーティションを増やす必要がある場合は、プロデューサーを停止し、コンシューマーがラグを処理し終わるまで待ち、パーティションを増やしてから、プロデューサーを再び有効にします。
または、パーティション数の多い新しいトピックを作成し、流れを徐々に新しいトピックに移します。

注意： Kafkaはパーティション数の削減をサポートしていません。過度に多くのパーティションを選択するにも費用がかかります：ブローカーのリソースの消費、リバランスが遅い、およびより多くのオープンファイル。適切なバッファを選択し、「確実性」のために1000パーティションを設定しないでください。

8. カスタムパーティショナーが必要な場合

自分でPartitionerを書く必要があることはほとんどありません。ほとんどの場合、正しいキーを選択すれば十分です。カスタムパーティショナーは、例えばいくつかのパーティションをVIP顧客専用にするなど、明示的なマッピングが必要な場合にのみ適切です。

逆に、カスタムパーティショナーには欠点があります：

トピックに書き込むすべてのプロデューサー（別の言語で書かれたサービスを含む）は、同じロジックを使用する必要があります。そうしないと、順序が破壊されます。
Kafka StreamsやKafka Connectなどのツールはデフォルトでmurmur2を使用します。それらはあなたのパーティション分散方法と合致しない可能性があります。

したがって、パーティショナーを検討する前に、キー設計で問題を解決することを優先してください。

9. まとめ

パーティションキーは設計上の決定であり、send()の単なるパラメータではありません。覚えるべき主要なポイント：

Kafkaはパーティション内の順序のみを保証します。同じキーと同じパーティション数の場合、同じパーティションになります。murmur2ハッシュを剰余計算した結果に従います。
キーは、順序を保つ必要がある最小グループの識別子を選択します。キーが多くの値を持つほど、負荷分散が均等になります。
ソルティングでホットキーを処理しますが、ビジネスがバケット間の順序の喪失を受け入れる場合に限ります。
プロデューサーでべき等性を有効にします。コンシューマーでは、メッセージを気分で任意のスレッドプールにプッシュしないでください。non-blocking retryを使用する前に注意深く検討してください。
順序が必要なトピックで実行中にパーティション数を増やすことを避けてください。最初から適切に見積もってください。

これらのポイントをしっかり理解することで、トラフィックが増加する際に、順序を保ちながらスケーリング可能なトピックを設計できます。

Kafkaのパーティションキー設計：メッセージの順序と負荷分散

1. 概要

2. Kafkaはパーティションをどのように選ぶか

2.1. キーがnull以外の場合：ハッシュ化して剰余を計算

2.2. キーがnullの場合：順序の保証なし

3. Spring Bootでキー付きメッセージを送信する

4. キーの選択：順序要件から始める

5. ホットパーティションと対処方法

5.1. ソルティング：ホットキーを複数のバケットに分割

5.2. キー戦略のテスト

6. コンシューマー側：順序を自分で破壊しない

7. パーティション数の変更：静かなる落とし穴

8. カスタムパーティショナーが必要な場合

9. まとめ

同じトピックの記事

Kafka の Consumer group と Rebalance (Spring Boot)

Change Data CaptureとDebeziumとKafka

KafkaでのSagaパターンによる分散トランザクション：ツアー予約の例

KafkaとのOutboxパターン：データベース書き込みでイベントを失わない

Kafkaの監視: consumer lagと監視対象のメトリクス

2.1. キーが`null`以外の場合：ハッシュ化して剰余を計算

2.2. キーが`null`の場合：順序の保証なし