소개
컨슈머가 이벤트를 표시했다고 해서 전달에 쓸 수 있는 결과를 저장한 것은 아닙니다. 이 애플리케이션은 각 값을 디코딩하고 키와 Kafka 위치를 로컬 JSONL 파일에 보존한 뒤, 해당 쓰기가 성공하고 플러시된 후에만 진행 상태를 커밋합니다.
기존 그룹·오프셋 모델을 Python에 연결합니다. subscribe는 토픽을 선택하고, poll은 레코드를 요청하며, 명시적인 동기 커밋은 처리한 메시지의 다음 오프셋을 저장합니다. 자동 커밋과 자동 오프셋 저장을 비활성화하여 코드와 별개로 진행 상태가 앞서 나가지 않도록 합니다. 파일 쓰기와 Kafka 커밋은 여전히 별도 작업이며, 시스템 간 원자적 트랜잭션은 아닙니다.
이 독립된 실습은 프로듀서 실습에서 보았던 초기 주문 이벤트 여섯 개를 새 토픽에 준비합니다. 이전 VM 상태는 필요하지 않습니다. Python 함수, 반복문, 딕셔너리, JSON, 스크립트 실행 지식이 필요합니다. 파서와 제한 시간 구조는 제공됩니다. 터미널과 설치된 bigdata-python 환경을 사용하세요.
구독과 폴링으로 실제 레코드 하나 디코딩하기
이 단계에서는 미리 보기의 진행 상태를 커밋하지 않고 수신한 레코드 하나를 확인합니다. 먼저 준비된 이벤트 값을 읽어보세요.
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Consumer에는 그룹 식별자와 초기 earliest 정책을 설정합니다. subscribe는 원하는 토픽을 등록합니다. poll은 메시지 또는 오류가 있는 메시지를 반환하며, 할당 중인 경우를 포함해 사용 가능한 레코드가 없으면 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
value에 py1부터 py6 중 하나가 있고 key가 해당 이벤트의 order_id와 일치하는 JSON 봉투 객체 하나가 출력됩니다. 파티션이 세 개이므로 처음 관찰되는 이벤트는 고정되어 있지 않습니다. 파티션과 오프셋은 실제 브로커 레코드를 식별합니다.
자동 커밋이 비활성화되어 있고 스크립트도 명시적으로 커밋하지 않았으므로, 미리 보기를 종료해도 진행 상태는 커밋되지 않습니다. 다음 애플리케이션은 파일 전달을 위해 자체 그룹을 사용합니다.
각 봉투 객체를 쓰고 플러시한 뒤 메시지 커밋하기
이 단계에서는 제한된 시간 동안 동작하는 파일 컨슈머를 만듭니다. 처리 순서를 주의 깊게 읽으세요. JSON 객체와 키를 디코딩하고 검증한 뒤, JSONL 봉투 객체 하나를 쓰고 파일을 플러시하며, 해당 메시지를 동기 커밋한 다음 완료 개수에 포함합니다. 쓰기, 디코딩, 커밋이 실패하면 오류를 건너뛰고 나중에 전체 작업이 성공한 것처럼 처리하지 않고 애플리케이션을 종료합니다.
with문은 블록을 벗어날 때 로컬 파일을 닫습니다. Path는 로컬 파일 이름을 지정합니다. 출력은 추가 모드로 열리므로 재시작이 이전 전달 증거를 조용히 지우지 않습니다. 깨끗한 첫 실행에서는 파일이 없는 상태로 시작합니다. max-messages는 이번 실행의 목표 개수이며, 전체 제한 시간 안에 완료하지 못하면 0이 아닌 상태로 종료됩니다. JSONL은 감싸는 배열 없이 한 줄에 완전한 JSON 객체 하나를 저장합니다. flush()는 커밋 요청 전에 Python의 파일 버퍼 데이터를 운영체제에 전달합니다. 정전이나 디스크 장애에서도 데이터가 살아남는다고 보장하지 않으며, 파일과 Kafka 커밋을 하나의 원자적 트랜잭션으로 만들지도 않습니다. 이 실습은 실행 중인 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과 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입니다. 이 다음 위치는 모든 파티션을 합한 전역 개수가 아닙니다.
컨슈머가 종료되었으므로 활성 멤버가 없다는 메시지는 정상입니다. 저장된 파일과 그룹 진행 상태는 남습니다. 처리 출력과 코디네이터 상태의 증거를 각각 확보했으며, 둘을 한 트랜잭션으로 썼다고 주장하지 않습니다.
같은 그룹을 재시작하고 새 전달이 없는지 관찰하기
이 단계에서는 earliest가 항상 되감는다고 가정하지 않고 저장 위치를 시험합니다. 레코드 여섯 개는 여전히 보관되어 있지만 file-writer에는 유효하게 커밋된 끝 위치가 있습니다. 같은 그룹으로 다시 시작하면 어떻게 될지 예상해 보세요. auto.offset.reset이 earliest여도 새 레코드를 찾지 못해야 합니다.
원래 레코드 여섯 개의 파일을 보존하도록 별도 출력 경로를 사용하세요. 짧은 전체 제한 시간으로 레코드 하나를 요청합니다. 요청한 개수에 도달할 수 없으므로 애플리케이션은 의도적으로 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에는 여전히 봉투 객체 여섯 개가 있어야 합니다. 같은 그룹의 유효한 커밋 오프셋은 earliest 시작 기본값으로 대체되지 않았습니다.
보관된 기록을 의도적으로 다시 읽으려면 이미 배운 별도 그룹 사용 또는 비활성 그룹 초기화 절차를 사용하세요. 쓰기 후 커밋 전에 프로세스가 멈추면 상황이 다릅니다. 같은 그룹이 레코드를 다시 받을 수 있습니다. 다음 실습에서는 이 장애 가능 구간을 직접 확인합니다.
요약
이 실습에서는 Python으로 구독하고 폴링하며 JSON/UTF-8 레코드를 디코딩하고 키와 Kafka 위치를 JSONL에 보존했습니다. 플러시가 성공한 뒤 명시적으로 동기 커밋했으며, 실제 코디네이터 상태가 완료된 파일과 일치했습니다. 같은 그룹의 재시작은 보관된 토픽 기록을 지우지 않은 상태에서 새 레코드를 찾지 못했습니다. 다음으로 디코딩 실패와 쓰기 후 커밋 전 중단을 통해 레코드가 다시 전달될 수 있는 이유를 확인합니다.
