🌊 들어가기에 앞서

지난 글 에서는 Console Producer와 Consumer를 직접 실행하면서 Consumer Group과 Offset의 동작을 확인했습니다. 하지만 실제 서비스에서 터미널에 메시지를 한 줄씩 입력할 수는 없습니다. 애플리케이션 코드가 이벤트를 만들고, 다른 애플리케이션이 그 이벤트를 읽어서 처리해야 합니다.

이번에는 Python으로 주문 이벤트를 전송하는 Producer와 처리하는 Consumer를 구현해보겠습니다. 이 과정에서 단순히 메시지를 보내고 받는 것보다 더 중요한 질문도 함께 다룹니다.

IMPORTANT

같은 주문에서 결제 완료와 결제 취소 이벤트가 차례로 발생했다면, Consumer도 반드시 그 순서대로 읽을 수 있을까요?

Kafka는 Topic 전체가 아니라 하나의 Partition 안에서만 순서를 보장합니다. 따라서 서로 연관된 이벤트를 같은 Partition에 넣어야 하고, 이때 사용하는 것이 메시지 Key입니다.

🐍 Python 실습 환경 구성하기

Kafka는 지난 글 과 동일하게 apache/kafka:4.3.1 Docker 이미지를 사용합니다. 먼저 컨테이너가 실행 중인지 확인합니다.

docker compose up -d
docker compose ps

Python에서는 confluent-kafka 라이브러리를 사용하겠습니다.¹ 내부적으로 librdkafka를 사용하기 때문에 처리량이 높고, Producer와 Consumer 설정도 Kafka의 개념과 비교적 직접적으로 연결됩니다.

python -m venv .venv
source .venv/bin/activate
pip install confluent-kafka

docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --create \
  --if-not-exists \
  --topic order-events \
  --partitions 3 \
  --replication-factor 1 \
  --bootstrap-server localhost:9092

📤 Python Producer 구현하기

주문 상태가 변경될 때마다 아래와 같은 이벤트를 Kafka에 보낸다고 가정해보겠습니다.

1
{
2
  "order_id": "order-1001",
3
  "status": "payment-completed"
4
}

Producer 코드를 producer.py에 작성합니다.

1
import json
2

3
from confluent_kafka import Producer
4

5

6
producer = Producer({
7
    "bootstrap.servers": "localhost:9092",
8
    "client.id": "order-api",
9
})
10

11

12
def delivery_report(error, message):
13
    if error is not None:
14
        print(f"전송 실패: {error}")
15
        return
16

17
    print(
18
        f"전송 성공: partition={message.partition()}, "
19
        f"offset={message.offset()}, key={message.key().decode()}"
20
    )
21

22

23
def send_order_event(order_id, status):
24
    event = {
25
        "order_id": order_id,
26
        "status": status,
27
    }
28

29
    producer.produce(
30
        topic="order-events",
31
        key=order_id.encode("utf-8"),
32
        value=json.dumps(event).encode("utf-8"),
33
        callback=delivery_report,
34
    )
35

36
    # 완료된 전송의 callback을 실행할 기회를 줍니다.
37
    producer.poll(0)
38

39

40
send_order_event("order-1001", "payment-completed")
41
send_order_event("order-1002", "payment-completed")
42
send_order_event("order-1001", "payment-canceled")
43

44
# 아직 전송 대기 중인 모든 메시지가 처리될 때까지 기다립니다.
45
producer.flush()

실행하면 메시지가 저장된 Partition과 Offset을 확인할 수 있습니다.

python producer.py

# 실행 결과 예시
전송 성공: partition=0, offset=0, key=order-1001
전송 성공: partition=2, offset=0, key=order-1002
전송 성공: partition=0, offset=1, key=order-1001

실제 Partition 번호와 Callback 출력 순서는 실행 환경에 따라 달라질 수 있습니다. 여기서 봐야 할 부분은 ==order-1001을 Key로 가진 두 메시지가 같은 Partition에 저장되었다==는 점입니다.

WARNING

produce()가 예외 없이 끝났다고 해서 Broker에 메시지가 저장된 것은 아닙니다. 전송은 비동기로 이루어지므로 Callback에서 성공 여부를 확인하고, 프로그램을 종료하기 전에는 flush()로 대기 중인 메시지를 처리해야 합니다.

🔑 메시지 Key는 Partition을 결정한다

Producer가 메시지를 보낼 때 Partition 번호를 직접 지정하지 않으면 Partitioner가 메시지를 어느 Partition에 넣을지 결정합니다.² 이때 Key가 있으면 같은 Key는 같은 해시 결과를 만들기 때문에 같은 Partition으로 전달됩니다.

결제 완료 이벤트가 먼저 발생하고 결제 취소 이벤트가 나중에 발생했다고 가정해보겠습니다. 두 이벤트가 서로 다른 Partition으로 들어가면 서로 다른 Consumer가 동시에 처리할 수 있고, 결제 취소가 먼저 끝날 가능성도 생깁니다.

반대로 order_id를 Key로 지정하면 같은 주문의 이벤트가 같은 Partition에 저장됩니다. 하나의 Partition은 Consumer Group 안에서 한 Consumer만 담당하므로, 해당 주문의 이벤트는 저장된 순서대로 읽을 수 있습니다.

IMPORTANT

Kafka에서 메시지 Key는 단순한 식별자가 아닙니다. 어떤 이벤트를 같은 Partition에 묶어서 순서를 지킬 것인지 결정하는 기준입니다.

1
producer.produce(
2
    topic="order-events",
3
    value=json.dumps(event).encode("utf-8"),
4
)

📥 Python Consumer 구현하기

이제 Producer가 보낸 메시지를 읽는 Consumer를 consumer.py에 작성합니다.

1
import json
2

3
from confluent_kafka import Consumer, KafkaError, KafkaException
4

5

6
consumer = Consumer({
7
    "bootstrap.servers": "localhost:9092",
8
    "group.id": "order-worker",
9
    "auto.offset.reset": "earliest",
10
    "enable.auto.commit": False,
11
})
12

13
consumer.subscribe(["order-events"])
14

15
try:
16
    while True:
17
        message = consumer.poll(timeout=1.0)
18

19
        if message is None:
20
            continue
21

22
        if message.error():
23
            if message.error().code() == KafkaError._PARTITION_EOF:
24
                continue
25
            raise KafkaException(message.error())
26

27
        key = message.key().decode("utf-8") if message.key() else None
28
        event = json.loads(message.value().decode("utf-8"))
29

30
        print(
31
            f"수신: partition={message.partition()}, "
32
            f"offset={message.offset()}, key={key}, event={event}"
33
        )
34

35
        # 실제 서비스에서는 비즈니스 처리가 성공한 뒤 Commit합니다.
36
        consumer.commit(message=message, asynchronous=False)
37
except KeyboardInterrupt:
38
    pass
39
finally:
40
    consumer.close()

Consumer를 실행한 뒤 Producer를 다시 실행합니다.

# 첫 번째 터미널
python consumer.py

# 두 번째 터미널
python producer.py

Consumer 출력에서 order-1001의 두 이벤트가 같은 Partition의 연속된 Offset에 위치하는 것을 확인할 수 있습니다.

1
수신: partition=0, offset=0, key=order-1001, event={'order_id': 'order-1001', 'status': 'payment-completed'}
2
수신: partition=0, offset=1, key=order-1001, event={'order_id': 'order-1001', 'status': 'payment-canceled'}
3
수신: partition=2, offset=0, key=order-1002, event={'order_id': 'order-1002', 'status': 'payment-completed'}

Partition별 메시지는 병렬로 읽기 때문에 서로 다른 Partition의 출력 순서는 달라질 수 있습니다. Kafka가 보장하는 것은 Topic 전체의 출력 순서가 아니라 Partition 내부의 저장 순서입니다.

🧭 Offset은 언제 Commit해야 할까?

Consumer 설정에서 enable.auto.commit을 False로 지정하고, 메시지 처리가 끝난 뒤 직접 Commit했습니다.

1
process_order(event)
2
consumer.commit(message=message, asynchronous=False)

Commit 시점은 장애가 발생했을 때 메시지를 놓칠지, 다시 처리할지를 결정합니다.

• 처리 전에 Commit: Commit 직후 Consumer가 죽으면 아직 처리하지 않은 메시지를 다시 읽지 못할 수 있습니다.
• 처리 후 Commit: 비즈니스 처리는 성공했지만 Commit 전에 죽으면 같은 메시지를 다시 읽을 수 있습니다.

데이터 유실보다 중복 처리를 허용하는 편이 안전하기 때문에, 일반적으로는 처리가 성공한 뒤 Commit합니다. 다만 이 방식도 처리와 Commit 사이에서 장애가 발생하면 중복을 완전히 피할 수 없습니다.

WARNING

Kafka Consumer를 사용한다고 해서 비즈니스 로직이 자동으로 Exactly Once가 되는 것은 아닙니다. 결제 요청, 재고 차감, 알림 발송처럼 외부 시스템에 영향을 주는 작업은 같은 이벤트를 다시 처리해도 결과가 달라지지 않도록 멱등성을 갖춰야 합니다.

예를 들어 이벤트 ID를 DB에 함께 저장하고, 이미 처리한 이벤트 ID라면 작업을 건너뛰는 방법을 사용할 수 있습니다.

1
def process_order(event):
2
    if event_repository.exists(event["event_id"]):
3
        return
4

5
    order_repository.update_status(
6
        order_id=event["order_id"],
7
        status=event["status"],
8
    )
9
    event_repository.mark_as_processed(event["event_id"])

실제로 적용할 때는 주문 상태 변경과 이벤트 처리 기록을 하나의 DB Transaction으로 묶어야 합니다. 그렇지 않으면 상태만 바뀌고 처리 기록은 남지 않는 또 다른 틈이 생깁니다.

🎯 메시지 Key는 어떻게 골라야 할까?

Key를 선택할 때는 무엇의 순서를 지켜야 하는가?를 먼저 질문해야 합니다.

Key를 너무 넓게 잡아 shop_id 하나로 모든 주문을 묶으면 특정 상점의 트래픽이 하나의 Partition에 몰릴 수 있습니다. 반대로 이벤트마다 새로운 UUID를 Key로 사용하면 분산은 잘 되지만 같은 주문의 이벤트가 서로 다른 Partition에 들어가 순서를 잃게 됩니다.

따라서 Key는 데이터를 구분하기 위한 Primary Key와 반드시 같을 필요는 없습니다. 함께 처리되어야 하는 이벤트의 경계를 표현해야 합니다.

😮‍💨 마무리

이번 글에서는 Python으로 Producer와 Consumer를 구현하고, 주문 ID를 Key로 사용해서 같은 주문의 이벤트를 같은 Partition에 저장해봤습니다.

• Producer의 전송은 비동기로 이루어지므로 Callback과 flush()로 결과를 확인해야 합니다.
• 같은 Key를 가진 메시지는 Partition 수가 유지되는 동안 같은 Partition으로 전달됩니다.
• Kafka의 순서 보장은 Topic 전체가 아니라 Partition 내부에만 적용됩니다.
• Consumer는 비즈니스 처리가 성공한 뒤 Offset을 Commit하는 편이 안전합니다.
• 처리 후 Commit 방식에서는 중복이 발생할 수 있으므로 Consumer 로직에 멱등성이 필요합니다.

결국 메시지 Key를 정하는 일은 단순한 Producer 설정이 아니라, 어떤 이벤트의 순서를 지키고 어떤 단위로 병렬 처리할지 결정하는 설계입니다. 다음 글에서는 Producer의 acks, 재시도와 중복 전송을 살펴보면서 Kafka가 메시지를 얼마나 안전하게 전달하는지 알아보겠습니다.