はじめに
コンシューマーがイベントを表示しても、引き渡しに使える結果を保存したとは限りません。このアプリケーションでは各値をデコードし、キーと Kafka 上の位置をローカルの JSONL ファイルに保持して、対応する書き込みが成功し、フラッシュされた後にだけ進行状況をコミットします。
これまでのグループとオフセットの仕組みを Python に対応付けます。subscribe はトピックを選び、poll はレコードを要求し、明示的な同期コミットは処理したメッセージの次のオフセットを保存します。自動コミットと自動オフセット保存を無効にし、コードと無関係に進行状況が進まないようにします。ファイルの書き込みと Kafka のコミットは別々の操作であり、システムをまたぐ不可分なトランザクションではありません。
この独立した演習では、プロデューサーの演習で使った最初の6件の注文イベントを、新しいトピックに用意しています。以前の VM の状態は不要です。Python の関数、ループ、辞書、JSON、スクリプトの実行の知識が必要です。パーサーと期限を管理する構造は用意されています。Terminal とインストール済みの bigdata-python 環境を使用してください。
購読、ポーリング、実際のレコード1件のデコードを行う
このステップでは、プレビューの進行状況をコミットせずに、受信したレコードを1件調べます。まず、用意されたイベント値を読みます。
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Consumer には、グループの識別情報と、初回に earliest から始める方針を設定します。subscribe は読みたいトピックを登録します。poll はメッセージ、エラーを含むメッセージ、または利用可能なレコードがないときの None を返します。割り当て中にも None が返る場合があります。ループは、全体の期限内でポーリングを続けます。
用意された制御フローによって、読み取りを時間内に収めます。time.monotonic() は、実際の日付や時刻に依存せずに経過時間を測ります。continue はメッセージがまだない場合に次のポーリングへ進み、break はレコードを表示した後にループを抜けます。while ループの else は、その break を実行せずにループが終わった場合にだけ動きます。ここでは、期限内にレコードが届かなかったことを意味します。
finally は、読み取りが成功してもエラーになってもコンシューマーを閉じます。この枠組みは Kafka 呼び出しの前後で待機と終了処理を行うものと捉えてください。再試行の枠組みを自分で設計する必要はありません。
このプレビューは別のグループ file-preview を使い、両方の自動設定を無効にします。デコードした値は Python の辞書です。それを包むレコードには、トピック、パーティション、オフセット、デコードしたキーも残します。コミットは呼び出しません。
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
JSON のエンベロープが1件表示され、その value は py1 から py6 のいずれかを含み、key はそのイベントの order_id と一致するはずです。パーティションが3つあるため、最初に観察するイベントは固定ではありません。パーティションとオフセットが、実際のブローカーレコードを識別します。
自動コミットが無効で、スクリプトにも明示的なコミットがないため、このプレビューを終了しても進行状況はコミットされません。次のアプリケーションは、ファイルへの引き渡し用に独自のグループを使います。
各エンベロープを書き込み、フラッシュしてからメッセージをコミットする
このステップでは、件数と時間を制限したファイル保存用コンシューマーを作ります。処理順序をよく読んでください。JSON オブジェクトとキーをデコード・検証し、JSONL のエンベロープを1件書き込み、ファイルをフラッシュし、そのメッセージを同期コミットしてから、完了件数に加えます。書き込み、デコード、コミットが失敗した場合は、エラーを飛ばして後からバッチ成功と見なすのではなく、アプリケーションを終了します。
with 文は、ブロックを抜けるときにローカルファイルを閉じます。Path はローカルファイル名を示します。再起動時に以前の配信結果を黙って切り詰めないよう、出力は追記モードで開きます。初回はファイルがない状態から始めます。max-messages は今回の実行での目標件数で、全体の期限に達して未完了なら0以外の状態で終了します。JSONL は外側の配列を付けず、1行ごとに完全な JSON オブジェクトを保存します。flush() は、コミット要求の前に、Python がバッファーしているファイルデータを OS に渡します。電源障害やディスク障害に耐えることを保証せず、ファイルと Kafka のコミットを1つの不可分なトランザクションにすることもありません。この演習では、動作中の VM でのプロセス中断と保存済みの進行状況を観察します。
commit(message=..., asynchronous=False) は、処理したメッセージの次のオフセットを、そのパーティションについてコミットします。返されたパーティションごとのエラーも確認します。enable.auto.offset.store=false は無関係な自動の進行状況保存を防ぎ、明示的な message 引数は、処理したどの位置をコミットするかを正確に指定します。
この処理順序を含む、用意されたパーサーとループの構造全体を保存します。
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Saved and committed records: 6 と、6件の JSONL エンベロープがあるはずです。各エンベロープには、完全な値、一致するキー、実際のレコード位置が保持されている必要があります。コールバックやコンソールログは、保存したファイルの代わりにはなりません。次のステップで、ファイルの結果を実際のコミット済み進行状況と比較します。
ファイルとパーティションごとのコミット済みの次のオフセットを照合する
このステップでは、すでに使ったコマンドラインツールで、コーディネーターの進行状況を調べます。
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
配信済みレコードを含む各パーティションでは、CURRENT-OFFSET がそのパーティションの LOG-END-OFFSET に達し、ラグは0になるはずです。この演習では、正確なパーティションの割り当てや分布は固定しません。レコードがないパーティションには、処理したメッセージがないため、コミット済みの位置がまだない場合があります。
保存したエンベロープのオフセットを、そのパーティションのコミット済みの次のオフセットと比較してください。パーティションで最後に処理したレコードのオフセットが n なら、次のコミット済み位置は n+1 です。この次の位置は、すべてのパーティションを合わせた件数ではありません。
コンシューマーは終了済みなので、アクティブなメンバーがいないというメッセージは正常です。保存したファイルとグループの進行状況は残ります。1つのトランザクションで書かれたと主張することなく、処理結果とコーディネーターの状態について別々の確認結果が得られました。
同じグループを再起動し、新しい配信がないことを観察する
このステップでは、earliest が常に巻き戻すと決めつけず、保存済みの位置を試します。6件のレコードは保持されたままですが、file-writer には有効なコミット済みの末尾位置があります。同じグループで再び起動した場合を予想してください。auto.offset.reset が earliest でも、新しいレコードは見つからないはずです。
元の6件のファイルを保持するため、別の出力パスを使います。全体の期限を短く設定して1件を要求します。要求件数に達しないため、このアプリケーションは、バッチ未完了を示す0以外の状態を意図的に返します。次のコマンドは失敗することを想定しています。
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
終了状態を直ちに保存します。
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
0以外の状態と Only received 0 of 1 requested records before the deadline. があるはずです。新しいファイルは空で、元の received-events.jsonl には6件のエンベロープが残るはずです。同じグループの有効なコミット済みオフセットは、初回に使う earliest 方針で置き換えられませんでした。
保持された履歴を意図的に再読するには、すでに学んだ別グループの使用、または非アクティブなグループのリセットを使います。プロセスが書き込み後、コミット前に停止した場合は事情が異なり、同じグループが再びそのレコードを受け取る可能性があります。次の演習では、この失敗が起こりうる区間を観察します。
まとめ
Python で購読とポーリングを行い、JSON/UTF-8 のレコードをデコードして、キーと Kafka 上の位置を JSONL に保持しました。フラッシュの成功後に明示的な同期コミットを行い、実際のコーディネーターの状態が完成したファイルと一致することを確認しました。同じグループで再起動すると、トピックの履歴を削除せずとも、新しいレコードは届きませんでした。次は、デコード失敗と書き込み後・コミット前の中断によって、レコードが再配信される理由を確認します。
