소개
콘솔 프로듀서로 이벤트와 키 모델을 확인했습니다. 애플리케이션에서는 한 가지를 더 구분해야 합니다. 전송 API 호출은 브로커 확인 전에 레코드를 클라이언트의 로컬 큐에 넣을 수 있습니다. 이 시점에 종료되는 프로그램은 레코드를 전달하지 않고도 성공했다고 보고할 수 있습니다.
준비된 로컬 Apache Kafka 브로커와 설치된 confluent-kafka Python 클라이언트를 사용합니다. 이 패키지는 클라이언트 라이브러리이므로 Confluent Cloud 계정은 필요하지 않습니다. 딕셔너리를 JSON/UTF-8 바이트로 직렬화하고 주문 키를 전송한 뒤, 전달 콜백과 최종 큐 상태를 확인합니다. 접근할 수 없는 로컬 주소로 제한 시간 내 실패도 시험합니다.
이 독립된 실습에는 파티션이 세 개인 토픽과 입력 데이터 두 개가 준비됩니다. 이후 활동은 Python 함수, 반복문, 딕셔너리, JSON, 스크립트 실행 지식을 전제로 합니다. 필요한 기초는 Python 학습 경로에서 익힐 수 있습니다. Kafka 설치와 서비스 설정은 준비되어 있습니다. 터미널을 사용하세요.
딕셔너리를 전송할 키와 값으로 변환하기
이 단계에서는 Kafka에 접속하기 전에 변환 과정을 확인합니다. 이벤트 여섯 개가 있는 입력부터 살펴보세요.
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
각 줄은 완전한 JSON 객체입니다. json.loads는 이 텍스트를 Python 딕셔너리로 변환합니다. json.dumps는 딕셔너리를 JSON 텍스트로 변환하고, encode("utf-8")은 텍스트를 클라이언트 API에 맞는 바이트로 바꿉니다. 키는 event_id가 아니라 order_id에서 별도로 인코딩합니다.
다음 간단한 미리 보기 프로그램을 저장하고 실행하세요. 구분자를 따옴표로 감싼 히어 도큐먼트는 Python 소스를 그대로 보존합니다.
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
딕셔너리, JSON 텍스트, 바이트 표현을 확인하세요. Python bytes 표시는 b로 시작하지만, 이 표기 자체는 전송되는 값의 일부가 아닙니다. 첫 번째 키는 order-4001을 인코딩한 값이어야 합니다. JSON은 이름이 있는 이벤트 필드를 보존하고 UTF-8은 문자 인코딩을 제공합니다.
아직 이벤트를 보내지 않았습니다. 다음 단계에서는 이 바이트를 프로듀서에 전달하고 브로커가 확인하는 시점을 검사합니다.
발행하고 모든 실제 전달 확인하기
이 단계에서는 제공된 인수 처리와 전달 콜백이 포함된 전체 프로듀서를 실행합니다. Producer는 클라이언트를 설정하고 bootstrap.servers는 초기 주소를 제공합니다. acks=all은 현재 동기화된 복제본들의 확인을 요청합니다. 브로커가 하나인 이 환경에는 복제본도 하나이므로, 이 설정이 여러 브로커의 내결함성을 보여 주지는 않습니다. 제공된 murmur2_random 라우팅 설정은 파티션 수가 고정되어 있는 동안 앞선 Java 콘솔 예제와 키 라우팅의 호환성을 유지합니다.
프로그램의 핵심 차이를 살펴보세요. produce는 레코드를 로컬 큐에 넣습니다. poll은 콜백을 처리하고, flush는 콜백을 처리하면서 큐에 있는 작업이 끝나기를 기다립니다.
콜백은 전달 결과를 사용할 수 있을 때 클라이언트 라이브러리가 호출하도록 제공하는 함수입니다. callback=delivery는 함수 자체를 전달하며 즉시 호출하지 않습니다. poll(0)은 기다리지 않고 준비된 콜백을 처리하며, flush는 아직 완료되지 않은 전송도 기다립니다. 콜백은 브로커가 확인한 토픽/파티션/오프셋 또는 실패를 기록합니다. flush 후 프로그램은 남은 레코드, 실패, 확인된 개수를 검사한 뒤 성공을 선언합니다. 실패 콜백은 수집되어 0이 아닌 프로세스 종료 결과를 유발하며, 단순히 출력하고 무시하지 않습니다.
제공된 argparse 부분은 명령줄 옵션을 args 속성에 연결합니다. 예를 들어 --input은 args.input이 되고, --delivery-timeout-ms는 args.delivery_timeout_ms가 됩니다. Path(args.input).open(...)은 선택한 로컬 파일을 열고, 반복문은 비어 있지 않은 줄마다 JSON 이벤트 하나를 읽습니다. 이 클라이언트를 사용하기 위해 인수 파서를 직접 설계할 필요는 없습니다.
전체 프로그램을 저장하세요. 파서와 콜백 구조는 제공되며, 직렬화와 produce 호출은 방금 확인한 바이트를 적용합니다.
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
프로듀서를 실행하고 두 출력 스트림을 모두 저장하세요. >는 표준 출력을 로그로 보내며, 2>&1은 표준 오류도 같은 대상으로 보냅니다. 따라서 콜백 실패와 연결 진단도 함께 보존됩니다.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
DELIVERED 콜백 줄 여섯 개와 Delivered records: 6이 출력되어야 합니다. 각 콜백의 실제 파티션과 오프셋은 달라질 수 있습니다. produce의 정상 반환만으로는 이러한 확인 응답을 입증할 수 없습니다.
이미 배운 콘솔 컨슈머로 브로커를 독립적으로 확인하세요.
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
py1부터 py6까지 찾고 키가 각 order_id 값과 일치하는지 확인하세요. 독립적인 읽기는 애플리케이션의 전달 보고서와 실제 브로커 레코드를 연결합니다. 특정 파티션 번호나 전역 표시 순서일 필요는 없습니다.
새 이벤트의 필드 하나를 바꾸고 프로듀서 재사용하기
이 단계에서는 같은 애플리케이션을 새 이벤트 하나에 재사용합니다. 준비된 next-event 데이터에는 새 event_id인 py7이 있으므로 아직 발행되지 않은 이벤트입니다. channel 필드를 확인하세요.
cat /home/labex/project/data/next-event.jsonl
별도의 작업 파일에서 이 미발행 이벤트의 channel만 online에서 store로 바꾸세요. 이벤트는 여전히 order-4001을 참조하며 sequence 3으로 앞선 이벤트들을 잇습니다.
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
새 성공 콜백 하나와 독립적인 읽기에서 보관된 이벤트 일곱 개가 예상됩니다. py7의 channel은 store여야 하고 py1부터 py6까지의 내용은 원래대로여야 합니다. 관련된 order-4001 이벤트들은 같은 파티션 안에서 sequence 순서를 따릅니다. 파티션 간 표시 순서는 달라질 수 있습니다.
프로듀서 구현은 바뀌지 않았습니다. 고정된 실습 정답을 내장하는 대신 입력 경로를 전달하여 새 데이터를 직렬화하고 전송했습니다.
접근할 수 없는 주소에서 제한 시간 내 실패 관찰하기
이 단계에서는 큐 적재를 전달로 오인하지 않는지 확인합니다. localhost:1에 Kafka 브로커가 없어도 같은 클라이언트가 로컬 큐를 만들 수 있습니다. 제공된 전달 제한 시간 옵션은 메시지 완료 시간을 3000밀리초로 제한하며, flush는 짧고 제한된 여유 시간을 추가합니다. 실제 브로커는 계속 localhost:9092에 있습니다.
이벤트 하나가 있는 파일을 의도적으로 잘못된 주소로 보내세요. 다음 명령은 실패하는 것이 정상입니다.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
다른 명령이 덮어쓰기 전에 즉시 상태를 저장하세요. $?는 직전 명령의 종료 상태로, 성공이면 0이고 실패이면 0이 아닙니다. printf로 이를 텍스트 한 줄로 저장합니다.
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
성공한 최종 개수 대신 0이 아닌 상태, 연결 진단, DELIVERY_FAILED 콜백, Delivery incomplete가 나와야 합니다. 메시지 실패 후 큐 처리가 끝났더라도 프로그램은 콜백 실패를 검사했습니다.
실패한 주소로의 전송이 실제 토픽에 두 번째 py7을 추가해서는 안 됩니다. 앞서 성공한 이벤트 일곱 개는 모두 그대로 사용할 수 있습니다. 엔드포인트를 바로잡고 업무 이벤트를 재시도할지 결정하는 일과 큐에 넣은 레코드가 확인되었다고 가정하는 일은 별개입니다. 이 실습의 의도적인 실패는 실제 브로커에 아무것도 보내지 않았습니다. 다음으로 Python 컨슈머에서 수신한 JSON을 디코딩하고 저장한 뒤 읽기 진행 상태를 커밋합니다.
요약
이 실습에서는 딕셔너리와 주문 키를 UTF-8 바이트로 직렬화하고 실제 Python 클라이언트로 발행한 뒤 성공 콜백을 실제 브로커 위치와 대조했습니다. 수정한 새 이벤트에도 같은 프로그램을 재사용했으며, 접근할 수 없는 주소에서는 잘못된 성공 대신 제한 시간 내에 0이 아닌 실패 결과를 얻었습니다. 다음 Python 컨슈머 실습에서는 디코딩, 파일 출력, 쓰기 후 커밋 동작을 명시적으로 다룹니다.
