Claude CodeでKafkaのexactly-once設定をレビューする手順

Kafkaのexactly-onceは、設定が1つ欠けても「動いているように見える」まま重複を生みます。Claude Codeに任せやすいのは、プロデューサー・コンシューマー・ブローカーの設定ファイルを横断して、矛盾する組み合わせを拾う作業です。

この記事では、enable.idempotence・transactional.id・isolation.levelの確認項目を表にし、それを検査スクリプトとして固定したうえで、Claude CodeのStopフックに載せる構成を扱います。最後に、再起動時の重複を再現する試験の組み立て方も書きます。

exactly-onceは3つの設定が揃って初めて成立する

Kafkaのexactly-onceは、1つのスイッチではありません。役割の違う設定を組み合わせた結果です。

• 冪等プロデューサー: ブローカーがプロデューサーIDを割り当て、送信ごとのシーケンス番号で再送の重複を除く。これだけなら、リトライでログに同じレコードが二重に入ることを防ぐ
• トランザクション: 複数パーティションへの書き込みと、コンシューマーのオフセット更新を1つの原子的な単位にする
• read_committedのコンシューマー: コミット済みのトランザクションだけを読む

設計ドキュメントは、コンシューマーのオフセットを出力トピックと同じトランザクションで書くことで、読み取り・処理・書き込みを通した保証になると説明しています。Kafka Streamsはexactly-onceの対象として挙げられていますが、素のProducer/Consumerで組むと、この3点を自分で揃えることになります。

レビューが難しい理由は、どれか1つが欠けても例外が出ないことです。たとえばコンシューマー側のisolation.levelの既定はread_uncommittedで、中止されたトランザクションのレコードもそのまま読めます。

設定の組み合わせで見るべき項目

プロデューサー側の確認項目を、ドキュメントの記述に沿って表にします。

最も見落としやすいのは、acks=1などの矛盾する値を入れたときの挙動です。ドキュメントには、冪等性を明示していないときは冪等性が無効になり、明示しているときはConfigExceptionが投げられると書かれています。

つまりenable.idempotenceを書かずにacks=1だけ残した設定ファイルは、起動に成功したうえで冪等性が黙って外れます。例外が出ないため、本番で重複が見つかるまで気づけません。

トランザクションとコンシューマー側の項目

トランザクションを使う構成では、次の項目が加わります。

設計ドキュメントには、コンシューマー設定にisolation.level=read_committedとenable.auto.commit=falseを含めること、プロデューサー設定にtransactional.idを置くことが明記されています。また、リバランスとの兼ね合いから、コンシューマーのインスタンスごとにプロデューサーを1つ持たせる構成が勧められています。

transactional.idをインスタンスごとに固定するか、起動のたびに乱数で作るかは、設定ファイルだけでは判断できません。${HOSTNAME}のようなプレースホルダが入っていれば、環境変数側まで追う必要があります。

Claude Codeにレビューさせる前に、CLAUDE.mdで観点を固定する

「exactly-onceをレビューして」とだけ頼むと、返ってくる指摘は実行のたびにばらつきます。観点を先に固定しておくと、レビューの粒度が揃います。Claude Codeのベストプラクティスも、リポジトリ固有の確認方法やコードだけでは分からない決まりをCLAUDE.mdに書くよう勧めています。

次のような断片を、Kafkaを使うリポジトリのCLAUDE.mdに置く形が考えられます。

# Kafka exactly-once のレビュー観点
- 設定の確認コマンド: `python3 tools/eos_lint.py producer|consumer <file>`
- read-process-write のアプリは transactional.id を必ず固定値にする
- isolation.level=read_committed と enable.auto.commit=false をセットで確認する
- DB など Kafka 外への書き込みは exactly-once の対象外。冪等キーの有無を指摘する
- 指摘は「設定名・現在値・期待値・影響」の4点で書く。スタイルの指摘はしない

最後の行は、ベストプラクティスで示されているレビュー依頼の考え方に沿っています。指摘を求められた側は何かしら見つけてくるため、観点と範囲を絞ることで、過剰な修正提案を避けます。

検査スクリプトで矛盾を機械的に拾う

観点の一部は、LLMに読ませるより短いスクリプトで判定したほうが確実です。次のスクリプトは、プロパティファイルを読み、ここまでの表にある矛盾を表示します。依存ライブラリはありません。

#!/usr/bin/env python3
"""使い方: eos_lint.py producer|consumer <file>  (問題があれば exit 2)"""
import sys
 
def load(path):
    props = {}
    for line in open(path, encoding="utf-8"):
        line = line.strip()
        if not line or line[0] in "#!" or "=" not in line:
            continue
        k, v = line.split("=", 1)
        props[k.strip()] = v.strip()
    return props
 
def producer(p):
    out = []
    idem = p.get("enable.idempotence")          # 未指定 = 既定 true
    if idem == "false":
        out.append(("WARN", "enable.idempotence=false: 再送で重複が書かれうる"))
    conflicts = []
    if p.get("acks") in ("0", "1"):
        conflicts.append("acks=" + p["acks"])
    if p.get("retries") == "0":
        conflicts.append("retries=0")
    if int(p.get("max.in.flight.requests.per.connection", "5")) > 5:
        conflicts.append("max.in.flight.requests.per.connection>5")
    for c in conflicts:
        if idem == "true" or "transactional.id" in p:
            out.append(("ERROR", f"{c}: 冪等性の前提と衝突 (ConfigException)"))
        elif idem is None:
            out.append(("WARN", f"{c}: 冪等性が黙って無効になる"))
    if "transactional.id" not in p:
        out.append(("INFO", "transactional.id なし: トランザクション未使用"))
    if int(p.get("transaction.timeout.ms", "60000")) > 900000:
        out.append(("ERROR", "transaction.timeout.ms がブローカー既定の上限を超える"))
    return out
 
def consumer(p):
    out = []
    if p.get("isolation.level", "read_uncommitted") != "read_committed":
        out.append(("WARN", "isolation.level が read_committed でない"))
    if p.get("enable.auto.commit", "true") != "false":
        out.append(("WARN", "enable.auto.commit が false でない"))
    return out
 
role, path = sys.argv[1], sys.argv[2]
res = (producer if role == "producer" else consumer)(load(path))
for lvl, msg in res:
    print(f"{lvl:5} {path}: {msg}")
sys.exit(2 if any(l in ("WARN", "ERROR") for l, _ in res) else 0)

acks=1だけを書いたファイルと、トランザクションのタイムアウトを長くしすぎたファイルに実行すると、次の出力になります（サンプルのプロパティファイルを用意して実行したものです）。

python3 tools/eos_lint.py producer producer-silent.properties
python3 tools/eos_lint.py producer producer-tx.properties
python3 tools/eos_lint.py consumer consumer-bad.properties

WARN  producer-silent.properties: acks=1: 冪等性が黙って無効になる
INFO  producer-silent.properties: transactional.id なし: トランザクション未使用
ERROR producer-tx.properties: transaction.timeout.ms がブローカー既定の上限を超える
WARN  consumer-bad.properties: isolation.level が read_committed でない
WARN  consumer-bad.properties: enable.auto.commit が false でない

このスクリプトは「ファイルに書かれた値」しか見ません。Spring Kafkaのように設定をコードやYAMLから組み立てるアプリでは、最終的にクライアントへ渡る値を別の経路で取り出す必要があります。また、ブローカー側の上限（transaction.max.timeout.ms）は既定値を決め打ちしています。運用で変えているクラスタでは、値を書き換えてください。

Stopフックで検査を終了条件にする

検査コマンドが決まったら、Claude Codeが作業を終える前に必ず走る形にします。ベストプラクティスでは、Stopフックがexit code 2で終了すると、標準エラーの内容が理由として渡され、Claudeが作業を続けると説明されています。公式の例は型チェックですが、同じ仕組みでKafka設定の検査も強制できます。

{
  "hooks": {
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "bash tools/eos_check.sh 1>&2 || exit 2"
          }
        ]
      }
    ]
  }
}

#!/usr/bin/env bash
rc=0
for f in config/producer*.properties; do
  python3 tools/eos_lint.py producer "$f" || rc=2
done
for f in config/consumer*.properties; do
  python3 tools/eos_lint.py consumer "$f" || rc=2
done
exit $rc

INFOだけのファイルはexit 0になるため、トランザクションを使わない単純なプロデューサーは止まりません。連続でブロックされた場合の上限はStopフックの連続ブロック上限で扱っています。

フックは全員に同じ検査を強制する手段です。個別のレビュー依頼なら、検査スクリプトの結果を踏まえて「差分のこの設定は、設計上exactly-onceを意図したものか」を聞くほうが向いています。設定の差分をレビューさせる別の型として、helm diffのアップグレード前差分をレビューする手順も同じ考え方です。

再起動時の重複を再現する試験

静的な検査で拾えるのは設定の矛盾までです。「再起動すると重複する」ことは、再起動を実際に起こして数えるしかありません。試験の骨格は次のとおりです。

設計ドキュメントによれば、トランザクションが中止されたとき、アプリケーションの状態とコンシューマーの位置は明示的に戻す必要があります。簡単な方針は、プロデューサーとコンシューマーを作り直してやり直すことです。コンシューマーグループのリバランスで最後にコミットされたオフセットから再開するため、トランザクション前の状態へ巻き戻る効果があります。試験では、この作り直しの経路も通してください。

Claude Codeには、この試験コードの下書きと実行を任せられます。プロンプトは次のように、期待値と読み方を指定する形が扱いやすくなります。

orders-enricher の再起動試験を書いて。入力100件、コミット前に強制終了、同じ設定で再起動。
出力トピックを read_committed と read_uncommitted の両方で読み、件数と重複キーを表示して。
実行結果をそのまま見せて、100件ちょうどでなければ原因の設定を指摘して。

最後の1行が、結果の確認までを任せる指示です。ベストプラクティスも、合格・不合格で返るチェックを持たせたうえで、実行したコマンドとその出力という証拠を求めるよう勧めています。

exactly-onceが成立しない範囲

設定が全部正しくても、保証の範囲には限りがあります。設計ドキュメントは、Kafka Streamsではexactly-onceが提供され、Kafkaのトピック同士の読み取り・処理・書き込みなら、トランザクション付きプロデューサーとread_committedのコンシューマーで同様に実現できるとしています。

一方、Kafka以外の宛先への書き込みは、その宛先との協調が必要です。DBへの書き込みを含む処理では、レビューの観点を「Kafkaの設定が正しいか」から「宛先側で冪等にできるか」へ広げます。Claude Codeに依頼するときも、DB書き込みのコードに冪等キー（主キーの一意制約など）があるかを聞くと、設定ファイルでは見えない漏れに気づけます。

読み取り側にも制約があります。read_committedのコンシューマーは、開いたままのトランザクションより後ろのメッセージを、そのトランザクションが完了するまで受け取れません。transaction.timeout.msを長くしすぎると、遅延として表面化します。ブローカーに上限があるのも、この停滞を避けるためです。

ブローカー側とクラスタ構成のレビュー

プロデューサー側が正しくても、ブローカーの構成で成立しないことがあります。ドキュメントの既定値では、トランザクション用の内部トピックのレプリケーション係数は3、最小ISRは2です。1台構成の開発クラスタでは、この既定のままだとトランザクションを使えません。

開発用の設定をそのまま本番へ持ち込む事故もあります。ブローカー設定のレビューでは、transaction.state.log.replication.factorとtransaction.state.log.min.isrが、実際のブローカー台数と食い違っていないかを見ます。min.insync.replicasも、acks=allと組み合わせて耐久性を決める値なので、同時に確認します。

クラスタの状態は、Claude CodeにKafka MCPサーバーを接続して調べさせる方法もあります。トピックやコンシューマーグループの情報はKafka MCPサーバーでトピックとコンシューマーグループを調べるの手順で取得できます。書き込み権限を持つ資格情報は渡さず、読み取り専用に絞る前提で使ってください。

まとめ

Kafkaのexactly-onceは、冪等プロデューサー・トランザクション・read_committedの3点が揃って成立します。設定ミスは例外にならず、重複として後から現れます。

設定の矛盾はスクリプトで機械的に拾い、Stopフックで終了条件にします。意図の確認と再起動試験の設計は、観点を固定したうえでClaude Codeに任せる分担が扱いやすくなります。DBなどKafka外への書き込みは別の問題として、宛先側の冪等性を確かめる必要があります。