Kafka アーキテクチャ

Kafka はイベントを再生可能なパーティションログとして保存します。これを理解するには、2つの独立した問題から始めることができます。⁠レコードはどこに保存されているか、読者はどこまで読んだかです。前者はブローカーのログとレプリケーション機構が担当し、後者はコンシューマーの位置とコンシューマーグループが担当します。

本稿では、Kafka 4.1 ドキュメントの通常トピック、レプリケーションログ、コンシューマーグループを基準とします。具体的なクライアントのデフォルト値やクラスター機能は、デプロイバージョンに応じて確認してください。Share group は、本稿のパーティション排他的消費モデルには含まれません。

ログが複数の独立した読者に適している理由

注文サービスが順に3つのイベント、すなわち「作成」「支払い」「出荷」を生成すると仮定します。在庫サービスは在庫に関心を持ち、集計サービスは売上高に関心を持ち、監査サービスは完全な履歴を保持する必要があります。これらは速度が異なり、それぞれ再起動する可能性があります。

「ある読者が読んだら」イベントを削除すると、他の読者は独立して進捗できなくなります。ログモデルは保持ポリシーと消費進捗を分離します。イベントはログに残り続け、各読者は独自の位置を維持します。コンシューマーは復元時にカーソルを移動するだけでよく、プロデューサーに履歴全体を再送信させる必要はありません。

オフセットはパーティション内の位置識別子であり、グローバルなメッセージIDでも、「残りメッセージ数」でもありません。ログ圧縮やトランザクション制御レコードなどの要因により、可視なビジネスレコードのオフセットは連続しない場合があります。

Topic、Partition、Replica の役割分担

パーティション数は、データがいくつの書き込みおよび消費ユニットに分散されるかを決定します。レプリカ数は、同じデータがいくつのコピーとして保存されるかを決定します。両方ともリソースを消費しますが、解決する問題は異なります。KRaft コントロールプレーンはメタデータ管理を担当し、データパーティションは独自のレプリケーションパスを持ちます。Kafka KRaft を参照してください。

1つのレコードの書き込みパス

クライアントのバッチ処理はネットワークとプロトコルのオーバーヘッドを薄めることができますが、バッチ処理を待つことでレイテンシが増加する可能性があります。圧縮は通常バッチに対して適用されます。パーティション数が増えると、トラフィックがより小さなバッチに分割される可能性があります。したがって、「パーティション数が多いほど速い」とは限りません。ホットスポット、バッチサイズ、ファイルディスクリプタ、レプリケーション接続、リカバリコストも考慮する必要があります。

同じビジネスエンティティを順序立てて処理する必要があるストリームでは、order_id をキーとしてよく使用します。順序の保証は、パーティショナー、キーのシリアライズ、パーティション数が一貫していることに依存します。パーティション数を変更するとマッピングが変わり、同じキーの新旧レコードが異なるパーティションに属する可能性があります。移行時には順序を個別に考慮する必要があります。

キーなしのレコードの割り当てはクライアントの実装に依存し、一律に1件ごとのラウンドロビンとして記述することはできません。Java クライアントのバッチ処理とパーティショニング戦略については、プロデューサー設定 を参照してください。同じキーは、順序付けが行われるパーティションを特定するだけです。複数の独立したプロデューサー間のビジネス上の前後関係は、アプリケーション側で構築する必要があります。

レプリケーション、ISR、および確認条件

リーダーは書き込みを受け付け、フォロワーは継続的にプルします。ISR（In-Sync Replicas）は、追いついていると見なされるレプリカの集合であり、リーダーを含みます。レプリカが集合に残るかどうかは、レプリケーションの進捗状況と関連する設定によります。

min.insync.replicas は acks=all と組み合わせて、書き込みを受け入れるための最小 ISR 条件を設定します。これは「2 に設定すると任意の2つのレプリカだけを待つ」という意味ではありません。例えば、レプリケーション係数が3で最小 ISR が2の場合：ISR が1つのレプリカだけになったとき、システムはこの種の書き込みを拒否し、単一レプリカでの成功を十分な冗長性の成功として偽装しません。詳細なセマンティクスについては Kafka Replication を参照してください。

「複数のノードにレプリケートする」と「各メッセージが各ディスク上で fsync を実行済みである」を区別する必要があります。確認ポリシーは、ログのフラッシュ、フェイルオーバードメイン、電源、リカバリポリシーを離れて、絶対的な永続性として単独で解釈することはできません。遅延レプリカをリーダーとして昇格させるリカバリポリシーは、確認済みレコードの損失を許容して可用性を得ることもあります。

3つの位置：末尾、レプリケーションコミット、トランザクション可視性

Kafka の読み書き位置は同じ数字ではありません。以下では「次の位置」を境界とし、概念的に簡略化します。

• LEO（Log End Offset）：あるレプリカのログ末尾の次の位置。レプリカごとに異なる場合があります。
• HW（High Watermark）：レプリケーションコミット境界。通常の読み取りでは、コミットされていないレプリケーション末尾を超えて読み取ることはできません。
• LSO（Last Stable Offset）：トランザクション安定境界。read_committed での読み取りはこれによって制限され、中止されたトランザクションがスキップされます。

この区別は、以下の現象を説明できます。ディスク上のログはすでに成長していますが、コンシューマーは後続のレコードを一時的に見ることができません。その理由は、レプリケーションの進捗状況や、前面の未完了のトランザクションのいずれかです。「ファイルにこのメッセージがあるかどうか」だけを見ても、それがコンシューマーに対して可視であるかどうかを判断することはできません。

同じ HW なのに、なぜ2つの分離レベルで異なるレコードが読めるか

図の位置を引き続き使用します。10、11 は通常のレコード、トランザクション T は 12 から始まり、13 はまた通常のレコードです。HW=14 は、これらの位置がすでにレプリケーションコミット範囲に入ったことを示していますが、T はまだ完了していません。read_uncommitted は 10、11、12、13 を返すことができますが、read_committed は 10、11 しか返すことができません。13 は T に属していませんが、前面のトランザクション安定境界を超えることはできません。

以下では、境界を入力として可視性フィルタリングをシミュレートします。トランザクションの状態 committed または aborted は、完了フラグが安定条件に達したことを示しており、プロデューサーがコミットリクエストをただ送信しただけではありません。モデルはブローカーのレプリケーションとトランザクション調整プロセスを省略しています。

records = [(10, None), (11, None), (12, "T"), (13, None)]

def visible(records, hw, transactions, committed_only):
    # transactions: トランザクションID -> (最初のオフセット, 安定状態)
    first_open = [first for first, status in transactions.values()
                  if status == "open"]
    lso = min([hw] + first_open)
    boundary = lso if committed_only else hw
    return [offset for offset, tx in records
            if offset < boundary
            and (not committed_only or tx is None
                 or transactions[tx][1] == "committed")]

assert visible(records, 14, {"T": (12, "open")}, False) == [10, 11, 12, 13]
assert visible(records, 14, {"T": (12, "open")}, True) == [10, 11]
assert visible(records, 15, {"T": (12, "committed")}, True) == [10, 11, 12, 13]
assert visible(records, 15, {"T": (12, "aborted")}, True) == [10, 11, 13]
assert visible(records, 11, {"T": (12, "open")}, True) == [10]
assert visible([], 0, {}, True) == []

最後の2つの完了シナリオでは、オフセット 14 にトランザクション完了フラグが格納され、HW がすでに 15 に到達していると仮定しています。フラグはビジネスレコードとしてモデルに含めません。T が中止になると、12 はフィルタリングされ、13 は可視性を回復するため、読み取り結果にギャップが生じます。このギャップは、コンシューマーが処理すべきレコードを見落としたことを意味するものではありません。

ここでは LSO を一貫して排他的な上限として扱う。読めるのは offset が LSO 未満のレコードであり、未完了のトランザクションが 12 から始まるなら境界は 12 である。read_committed での endOffsets() は min(HW, 最も早い未完了トランザクションの開始 offset) を返す。KafkaConsumer の公式仕様を参照。したがって、LSO を基準とする lag が 0 でも、未完了のトランザクションによって、すでに複製済みの末尾まで読めない場合がある。

トランザクション安定境界（LSO）の定義

LSO は排他的上限として統一されています。オフセットが LSO より小さいレコードを読み取ります。未完了のトランザクションが 12 から始まる場合、境界は 12 です。read_committed モードでの endOffsets() は min(HW, 最も早い未完了トランザクションの起点) を返します。詳細は KafkaConsumer のトランザクション読み取りと末尾位置の定義 を参照してください。これはまた、LSO に対するラグがゼロであることだけを見ても、レプリケーションログの末尾にトランザクションによってブロックされた蓄積がないことを証明できない理由も説明しています。

レコードはすでにレプリケートされているのに、なぜまだ読めないのか？

「トランザクションのコミットを要求する」と「コミットフラグがすでに安定している」を区別する必要があります。以下では3つの独立した履歴を比較します。完了履歴にはオフセット 14 にフラグがありますが、HW が 15 に到達して初めて、トランザクションが安定して終了したと見なされます。すべての可視性境界は排他的上限です。中止後のギャップはフィルタリングによるもので、コンシューマーが決して配信されないオフセット 12 を待つべきではありません。

消費位置とコミット済み位置

1回の poll() は複数のレコードを返し、クライアントの読み取り位置は随之進みますが、ビジネス処理は必ずしも完了していません。コミット済みオフセットは障害復旧時の進捗の約束であり、これらは分離されるべきです。

通常のコンシューマーグループでは、1つのパーティションは同時に1つのメンバーに割り当てられ、異なるグループは同じパーティションを独立して読み取ることができます。クライアントはオフセットに従って読み取りますが、非同期ワースレッドが同じ順序で完了することを自動的に保証するものではありません。引き継ぎプロセスと並列コミットルールについては、コンシューマーグループとコラボレーション を参照してください。

保持（Retention）とログ圧縮（Log Compaction）

2つのクリーンアップ方法は、2つのデータ用途に対応しています。

圧縮はバックグラウンドで行われ、読み取り時に各キーが1回だけ出現することが保証されるわけでも、残りのレコードの順序が入れ替えられたり再番号付けされたりするわけでもありません。削除状態は通常、tombstone（墓石レコード）によって表現され、対応する保持期間を過ぎると、tombstone 自体もクリーンアップされる可能性があります。遅れすぎた読者は、残りのログだけでは正しい削除履歴を復元できない場合があります。Log Compaction には、これらの保証の詳細が説明されています。

したがって、再生可能かどうかは保持の境界にかかり、状態を再構築できるかどうかは圧縮キーと削除セマンティクスにかかります。イベント監査と最新状態の同期は、「Kafka に存在する」というだけでは同じニーズとして見なすことはできません。

削除マークが消えた後、空のデータベースからの再構築と古いスナップショットからの継続は等価ではない

ログが順に 0:A=1、1:B=7、2:A=2、3:B=null であると仮定します。最後のレコードで B が削除されます。以下では、「すべての古いバージョンがすでにクリーンアップされた」結果をシミュレートしますが、実際のクリーナーの非同期スケジューリングは意図的にシミュレートしません。

log = [(0, "A", 1), (1, "B", 7), (2, "A", 2), (3, "B", None)]

def compact_snapshot(records, keep_tombstones):
    latest = {}
    for record in records:
        latest[record[1]] = record
    return sorted((record for record in latest.values()
                   if keep_tombstones or record[2] is not None),
                  key=lambda record: record[0])

def replay(initial, records):
    state = dict(initial)
    for _, key, value in records:
        if value is None:
            state.pop(key, None)
        else:
            state[key] = value
    return state

with_delete = compact_snapshot(log, True)
without_delete = compact_snapshot(log, False)
assert with_delete == [(2, "A", 2), (3, "B", None)]
assert without_delete == [(2, "A", 2)]  # オフセットは 2 のまま
assert replay({}, log) == replay({}, without_delete) == {"A": 2}
stale_snapshot = {"A": 1, "B": 7}
assert replay(stale_snapshot, with_delete) == {"A": 2}
assert replay(stale_snapshot, without_delete) == {"A": 2, "B": 7}

古いスナップショット内の B は、ログに「B がない」からといって自動的に削除されません。コンシューマーは「B は一度も変更されていない」と「B は削除されたが、削除の証拠が期限切れになった」を区別できません。したがって、復元プロトコルでは、スナップショットに対応するオフセット、増分ログの利用可能な範囲、および削除の保持範囲を一緒に合意する必要があります。単にブローカーに接続できるかどうかを確認するだけでは不十分です。

ここで空のデータベースからの再構築が正しい結果を得られるのは、入力が固定された完全な圧縮後の集合であるためです。実際の再構築中、クリーナーは引き続き実行されています。読者が B の古い値を先に読み取り、その後クリーンアップされる tombstone を読む時間がなかった場合、誤った状態を保持する可能性があります。したがって、空のデータベースから開始する場合は、読み取り進捗と削除の保持条件を満たす必要があり、モデルの等式を無限時間の復旧の保証として扱うことはできません。関連する条件については Log Compaction の保証 を参照してください。

アーキテクチャの推論：注文フローの役割分担

合理的な分割は、注文 ID でパーティション分けし、同じ注文のイベントにパーティション内順序を持たせることです。在庫グループと集計グループは各自が進行状況をコミットします。監査ニーズは履歴の保持期間を決定し、レプリケーションと確認ポリシーは障害時の書き込みの約束を決定します。

ある大口顧客がトラフィックの半分を生成する場合、そのキーはホットスポットを形成する可能性があります。コンシューマーを増やしても、排他的なパーティションを自動的に分割することはできません。より細かいキーを使用すると並列度が向上しますが、順序を保つ必要があるビジネスの範囲を再定義する必要があります。パフォーマンスのトレードオフは、設定を単に大きくするだけでなく、ビジネスの関係性から生じます。

参照

• Kafka 4.1：Design
• Kafka 4.1：Producer Configs
• Kafka 4.1：KRaft
• メッセージセマンティクス：リトライ、冪等性、トランザクションの境界

キーワード：Kafka、partition、replica、ISR、acks、offset、HW、LSO、retention、log compaction