Kafka コンシューマの構築
使用するタイミング
- Kafka コンシューマ/プロデューサの作成またはデバッグ時。
- オフセットコミット戦略と配信保証の選択時。
- コンシューマグループの並列性、リバランス、またはデッドレター処理のチューニング時。
- ストリーム処理/ウィンドウ処理には使用しない（
processing-streaming-dataを使用すること）。
ワークフロー
- [ ] ターゲットの並列性に合わせてパーティションをサイジングする (コンシューマ数 <= パーティション数)
- [ ] 処理が成功した後にオフセットをコミットする
- [ ] シンクを冪等にする (イベントキーによる upsert)
- [ ] リバランスを処理する (リボーク時にコミット、長時間のポーリングギャップを避ける)
- [ ] ポイズンメッセージをデッドレターキューにルーティングする
- パーティションが並列性を制限する。 コンシューマグループのスケールアウトはパーティション数が上限であり、余分なコンシューマはアイドル状態になります。ピークスループットに合わせてパーティションを選択してください。
- 処理後にコミットする。 処理が確実に完了した後にオフセットをコミットします。前にコミットすると、クラッシュ時にメッセージが失われます。
- 冪等なシンク。 At-least-once ではリトライ時に重複が発生します。安定したイベントキーで upsert することで、再処理が無害になります。
- リバランスは発生する。 パーティションリボーク時にコミットし、
poll()の間隔をmax.poll.interval.ms未満に保つことで、ブローカーがコンシューマを追い出さないようにします。 - ポイズンメッセージ はエラー情報とともにデッドレターキューに送信し、1つの不良レコードがパーティション全体をブロックしないようにします。
パターン
処理後の手動コミット:
consumer = KafkaConsumer("orders", group_id="etl",
enable_auto_commit=False,
max_poll_records=500)
for msg in consumer:
try:
upsert(process(msg)) # キーで冪等
consumer.commit() # 成功後にのみコミット
except PoisonError:
send_to_dlq(msg)
consumer.commit() # 不良レコードをスキップ
冪等 / トランザクショナルプロデューサ — enable.idempotence=true を設定してリトライの重複を排除し、トランザクションを使用してトピック間で読み取り-処理-書き込みの exactly-once を実現します。
スループットチューニング — max.poll.records、fetch.min.bytes を増やし、バッチで処理します。処理を高速に保ち、リバランスによる排除を回避します。
よくある落とし穴
- 自動コミット + 低速処理 — 処理が完了する前にオフセットが進み、クラッシュ時にメッセージが失われます。処理後の手動コミットを推奨します。
- パーティション数より多いコンシューマ — 余分なコンシューマはアイドル状態になります。スケールするには再パーティショニングしてください。
- ポーリング間の長時間処理 —
max.poll.interval.msを超え、無限リバランスを引き起こします。制限付きバッチで処理するか、バックグラウンドワーカーを使用してください。 - デッドレターのパスがない — 1つのポイズンメッセージがパーティション全体をブロックします。
- 冪等なシンクなしで exactly-once に依存する — どの at-least-once のホップでも重複が再導入されます。