원문 캡처 · rsbh.dev
원문 캡처 · rsbh.dev

이 글은 파이썬으로 Kafka에 이벤트를 보내는 서비스와 이를 소비하는 서비스를 만들고 Docker와 docker-compose로 실행하는 과정을 설명한다. Kafka 브로커를 로컬에서 띄우고, confluent-kafka와 uv를 사용해 두 서비스를 구성한다.

Kafka에서는 같은 종류의 이벤트를 토픽에 모으며, 프로듀서는 브로커의 토픽에 메시지를 보낸다. 여러 컨슈머가 토픽을 구독해 이벤트를 처리할 수 있다. 예시에서는 test-topic을 사용하고, 프로듀서가 0.5초마다 임의의 사용자 데이터를 발행한다. 메시지에는 사용자 ID를 키로 넣어 같은 사용자의 메시지가 같은 파티션에 배정되도록 한다.

먼저 프로젝트 루트에 docker-compose.yaml을 만들고 Kafka 서비스를 추가한다. docker compose up kafka를 실행하면 브로커가 포트 9092에서 동작한다. 프로듀서는 KAFKA_BROKERS 환경 변수에서 브로커 주소를 읽으며, 기본값은 localhost:9092다. 컨테이너에서는 kafka:19092를 사용한다.

프로듀서 서비스에는 confluent-kafka, faker, structlog를 설치한다. 연결을 확인하고 메시지 전달 결과를 기록하며, 종료할 때 flush로 대기 중인 메시지를 브로커에 보낸다. 예시의 메시지에는 UUID, 사용자 ID, 이름, 이메일, 생성 시각이 포함된다. 토픽은 예시에서 자동 생성하지만, 글은 실제 운영 환경에서는 스크립트로 토픽을 명시적으로 만들고 파티션과 복제 계수 등을 설정하는 편이 바람직하다고 설명한다.

프로듀서와 컨슈머는 각각 Docker 이미지로 만들 수 있다. docker compose build producer로 프로듀서를 빌드하고 docker compose up producer로 실행한다. 컨슈머는 confluent-kafka와 structlog를 사용하며, 토픽과 그룹 ID, 브로커 주소를 설정한다. 메시지를 가져온 뒤 콜백 함수로 처리하고 브로커에 소비 완료를 커밋한다. 컨슈머 그룹의 구성원은 작업을 나눠 처리할 수 있다.

글의 예시에서는 토픽이 파티션 하나로 만들어져 컨슈머를 세 개 실행해도 하나만 메시지를 처리하고 나머지는 대기한다. kafkactl alter topic test-topic --partitions 3으로 파티션을 세 개로 늘리면 세 컨슈머가 메시지를 처리할 수 있다. 컨슈머 수가 파티션 수보다 많으면 초과한 컨슈머는 대기 상태가 된다.

프로젝트 소스 코드는 GitHub 저장소에서 확인할 수 있다.

부록: 원문 예시

text
import structlog log = structlog.get_logger() structlog.configure(     processors=[         structlog.processors.add_log_level,         structlog.processors.TimeStamper(fmt="iso"),         structlog.processors.JSONRenderer(),     ] )
text
class EventProducer:     def check_connection(self, timeout: float = 3.0) -> bool:         try:             cluster_id = self.producer.cluster_id(timeout=timeout)             log.info("kafka_connection_verified", cluster_id=cluster_id)             return True         except KafkaException as e:             log.error("kafka_connection_failed", error=str(e))             return False
text
class EventProducer:     def __init__(self, config: EventProducerConfig) -> None:         self.topic = config.topic         self.producer = Producer(             {                 "bootstrap.servers": config.servers,                 "client.id": socket.gethostname(),                 "acks": "all",             }         )     def check_connection(self, timeout: float = 3.0) -> bool:         try:             cluster_id = self.producer.cluster_id(timeout=timeout)             log.info("kafka_connection_verified", cluster_id=cluster_id)             return True         except KafkaException as e:             log.error("kafka_connection_failed", error=str(e))             return False     def _acked(self, err, msg):          if err is not None:              log.error("delivery_failed", topic=msg.topic(), key=msg.key(), error=str(err))          else:              log.info("message_delivered", topic=msg.topic(), partition=msg.partition(), offset=msg.offset())     def publish(self, key: str, msg: dict) -> None:         value = json.dumps(msg).encode()         self.producer.produce(self.topic, key=key, value=value, callback=self._acked)         self.producer.poll(0)     def close(self) -> None:         remaining = self.producer.flush(10)         if remaining:             log.error("messages_not_delivered", count=remaining)         self.producer.close()
text
def publish_test_data(producer: EventProducer, shutdown: Event) -> None:     fake = Faker()     while not shutdown.is_set():         user_id = fake.random_int(min=1, max=10)         msg = {             "id": str(uuid.uuid4()),             "user_id": user_id,             "name": fake.name(),             "email": fake.email(),             "created_at": datetime.now(timezone.utc).isoformat(),         }         try:             producer.publish(key=str(user_id), msg=msg)         except (BufferError, KafkaException) as e:             log.error("publish_failed", error=str(e))         shutdown.wait(0.5)     log.info("shutting_down")     producer.close()
text
def main():     try:         config = EventProducerConfig(             topic="test-topic",             servers=os.environ.get("KAFKA_BROKERS", "localhost:9092"),         )         producer = EventProducer(config=config)         if not producer.check_connection(timeout=3.0):             print(                 f"\n[Error] Could not connect to Kafka brokers at {config.servers}. Exiting."             )             return         shutdown = Event()         def handle_signal(signum: int, frame: FrameType | None) -> None:             shutdown.set()         signal.signal(signal.SIGINT, handle_signal)         signal.signal(signal.SIGTERM, handle_signal)         publish_test_data(producer=producer, shutdown=shutdown)     except Exception as e:         log.error("unexpected_error", error=str(e))         print(f"\n[Error] An unexpected error occurred: {e}")
text
class EventConsumer:          def consume(self, stop: Event, on_message: Callable[[Message], None]) -> None:         self.consumer.subscribe([self.topic])         log.info(             "consumer_started",             topic=self.topic,         )         try:             while not stop.is_set():                 msg = self.consumer.poll(timeout=1.0)                 if msg is None:                     continue                 if msg.error():                     raise KafkaException(msg.error())                 else:                     on_message(msg)                     self.consumer.commit(message=msg, asynchronous=True)         finally:             log.info("closing_consumer...")             self.consumer.close()             log.info("consumer_stopped")
text
def process_msg(msg: Message) -> None:     value = msg.value()     if value is None:         return     log.info("processing_message", value=value.decode("utf-8")) def main():     try:         config = EventConsumerConfig(             topic="test-topic",             group_id="test-group",             servers=os.environ.get("KAFKA_BROKERS", "localhost:9092"),         )         consumer = EventConsumer(config=config)         shutdown = Event()         def handle_signal(signum: int, frame: FrameType | None) -> None:             shutdown.set()         signal.signal(signal.SIGINT, handle_signal)         signal.signal(signal.SIGTERM, handle_signal)         consumer.consume(stop=shutdown, on_message=process_msg)     except Exception as e:         log.error("unexpected_error", error=str(e))         print(f"\n[Error] An unexpected error occurred: {e}")