Spakky Kafka

Spakky Framework를 위한 Apache Kafka 플러그인입니다. Event transport, consumer lifecycle, tracing/auth metadata propagation을 Kafka boundary에 연결합니다.

설치

pip install spakky-kafka

또는 Spakky extra로 설치합니다:

pip install spakky[kafka]

IAsyncEventPublisher와 함께 쓰는 Kafka 이벤트 서비스라면 event core까지 포함하는 spakky[events-kafka]를 권장합니다.

설정

SPAKKY_KAFKA__ prefix를 가진 환경변수를 설정합니다:

export SPAKKY_KAFKA__GROUP_ID="my-consumer-group"
export SPAKKY_KAFKA__CLIENT_ID="my-app"
export SPAKKY_KAFKA__BOOTSTRAP_SERVERS="localhost:9092"
export SPAKKY_KAFKA__AUTO_OFFSET_RESET="earliest"  # earliest, latest, none

SASL 인증 (선택)

export SPAKKY_KAFKA__SECURITY_PROTOCOL="SASL_SSL"
export SPAKKY_KAFKA__SASL_MECHANISM="PLAIN"
export SPAKKY_KAFKA__SASL_USERNAME="username"
export SPAKKY_KAFKA__SASL_PASSWORD="password"

Topic 설정 (선택)

export SPAKKY_KAFKA__NUMBER_OF_PARTITIONS="3"
export SPAKKY_KAFKA__REPLICATION_FACTOR="1"

처리 실패 메시지 설정 (선택)

export SPAKKY_KAFKA__DEAD_LETTER_TOPIC_SUFFIX=".dlt"
export SPAKKY_KAFKA__MAX_HANDLER_RETRIES="0"
export SPAKKY_KAFKA__DEAD_LETTER_DELIVERY_TIMEOUT="10.0"

전달 의미 (at-least-once)

KafkaConnectionConfig는 producer 설정과 consumer 설정을 분리하여 각각의 안전한 기본값을 프레임워크가 고정합니다. 환경변수로 덮어쓸 수 있는 항목이 아닙니다.

AsyncKafkaEventTransport는 producer를 애플리케이션 수명 동안 하나로 유지하므로, 멱등 producer의 시퀀스 번호가 그 수명 내내 이어집니다. 따라서 위 enable_idempotence 설정이 서로 다른 send 호출 사이에서도 중복 발행과 파티션 내 재정렬을 막습니다. 발행마다 producer를 새로 만들면 시퀀스가 초기화되어 이 보장이 성립하지 않습니다 — 자세한 수명주기는 아래 "Producer 수명주기"를 참고하십시오.

Consumer는 등록된 핸들러가 끝난 뒤 그 메시지를 처리 완료로 볼지 다시 받을지 결정합니다. MessageOutcome이 그 두 결정을 나타냅니다.

offset은 실패가 다른 곳에 저장된 뒤에만 전진합니다. dead-letter 발행이 실패했는데 커밋해 버리면 그 이벤트의 마지막 사본이 사라집니다. 그래서 발행 실패는 커밋 대신 되감기로 이어집니다.

커밋을 건너뛰는 것만으로는 재처리가 성립하지 않습니다. enable.auto.commit=false는 브로커 커밋만 끄고 consumer의 소비 위치는 poll마다 전진합니다. 되감지 않으면 뒤따르는 메시지의 성공 커밋이 실패한 메시지의 offset을 지나쳐 버려 그 메시지는 rebalance나 재시작으로도 돌아오지 않습니다. seek 되감기가 재처리를 성립시키는 수단입니다.

전달 의미가 at-most-once에서 at-least-once로 바뀌었으므로 같은 이벤트가 두 번 이상 전달될 수 있습니다. 이벤트 핸들러는 멱등해야 합니다. 이미 처리한 이벤트 식별자를 저장하거나 upsert로 반영하는 방식이 필요합니다.

커밋과 되감기 실패는 로그로 남기고 삼킵니다. consumer 루프는 background service 스레드 (비동기는 asyncio 태스크)에서 돌기 때문에, 커밋 실패 예외를 전파하면 애플리케이션은 살아 있는 채로 모든 구독이 멈춥니다.

메시지 단위 동기 커밋은 브로커 왕복을 한 번씩 추가합니다. 처리량이 커밋 지연에 지배되는 워크로드라면 파티션 수를 늘려 consumer를 병렬화합니다.

사용법

이벤트 발행

from spakky.core.common.mutability import immutable
from spakky.domain.models.event import AbstractIntegrationEvent
from spakky.event.event_publisher import IEventPublisher
from spakky.core.pod.annotations.pod import Pod

@immutable
class UserCreatedEvent(AbstractIntegrationEvent):
    user_id: int
    email: str

@Pod()
class UserService:
    def __init__(self, publisher: IEventPublisher) -> None:
        self.publisher = publisher

    def create_user(self, email: str) -> User:
        user = User(email=email)
        self.publisher.publish(UserCreatedEvent(user_id=user.id, email=email))
        return user

이벤트 수신

from spakky.event.stereotype.event_handler import EventHandler, on_event

@EventHandler()
class UserEventHandler:
    def __init__(self, notification_service: NotificationService) -> None:
        self.notification_service = notification_service

    @on_event(UserCreatedEvent)
    async def on_user_created(self, event: UserCreatedEvent) -> None:
        await self.notification_service.send_welcome_email(event.email)

비동기 변형

비동기 애플리케이션에서는 IAsyncEventPublisher를 사용합니다:

from spakky.event.event_publisher import IAsyncEventPublisher

@Pod()
class AsyncUserService:
    def __init__(self, publisher: IAsyncEventPublisher) -> None:
        self.publisher = publisher

    async def create_user(self, email: str) -> User:
        user = User(email=email)
        await self.publisher.publish(UserCreatedEvent(user_id=user.id, email=email))
        return user

분산 트레이싱

spakky-tracing은 필수 의존성으로 자동 설치됩니다. ITracePropagator가 컨테이너에 등록되어 있으면 이벤트 발행/소비 시 TraceContext가 자동으로 전파됩니다.

• 발행 측: IEventTransport.send() 시 현재 TraceContext를 Kafka 메시지 헤더에 주입합니다
• 소비 측: 수신 메시지에서 TraceContext를 추출하여 자식 스팬을 생성합니다
• 헤더가 없으면 새로운 루트 트레이스를 시작합니다

AuthContext 스냅샷 소비

spakky-auth 보호 decorator가 붙은 Kafka event handler는 raw bearer token을 받지 않습니다. Consumer는 사용자 handler 호출 직전에 x-spakky-auth-context-snapshot 또는 spakky.auth.context_snapshot Kafka header의 signed AuthContextSnapshot을 IAuthContextSnapshotVerifier로 검증하고, ApplicationContext에 AuthContext를 seed합니다.

• snapshot 검증은 KafkaPostProcessor가 message별 clear_context()를 수행한 뒤, 사용자 handler를 호출하기 전에 실행됩니다.
• missing, invalid, expired snapshot은 보호된 handler를 호출하지 않는 CHALLENGE fail-closed로 처리하며 Kafka consumer loop는 메시지를 처리 완료로 둡니다.
• 보호 요구사항 DENY도 handler 호출을 완료 처리하여 offset poison loop를 만들지 않습니다.
• verifier provider unavailable은 ERROR로 전파되어 consumer route에서 삼키지 않고 broker/runtime retry 정책에 맡깁니다.
• 기존 traceparent header와 event payload 역직렬화 의미는 그대로 유지됩니다.

처리 실패 메시지 (dead-letter)

핸들러 실패와 역직렬화 실패 메시지는 원본 topic 이름에 dead_letter_topic_suffix(기본 .dlt)를 붙인 topic으로 전달됩니다. 원본 본문과 key는 바이트 그대로 전달하고, 원본 header는 consumer가 읽은 문자열 형태로 함께 싣습니다(값이 없거나 UTF-8 문자열이 아닌 header는 제외). 재처리 도구가 본문을 열지 않고 판단할 수 있도록 원본 좌표와 예외 정보를 header로 덧붙입니다.

• x-spakky-dead-letter-original-topic / -partition / -offset / -timestamp: 원본 좌표
• x-spakky-dead-letter-consumer-group: 처리에 실패한 consumer group
• x-spakky-dead-letter-exception-type / -exception-message: 실패 원인

max_handler_retries를 0보다 크게 두면 dead-letter 전에 핸들러를 그 횟수만큼 다시 호출합니다. 재호출은 해당 이벤트의 모든 handler를 다시 실행하므로 handler 멱등성이 전제입니다. 역직렬화 실패는 재시도가 해결하지 못하므로 이 설정과 무관하게 즉시 dead-letter로 보냅니다.

dead-letter 발행은 dead_letter_delivery_timeout(기본 10초)까지만 배달을 기다립니다. 크기 제한 초과·producer 큐 포화·브로커 거절은 오류 로그를 남기고 consumer는 계속 폴링합니다. 비동기 경로에서 배달 확인이 시간 안에 오지 않은 경우는 확정 실패와 구분해 "unconfirmed"로 기록합니다 — 그 레코드는 나중에 배달될 수 있어 수동 재발행 시 중복이 생깁니다.

dead-letter topic은 consumer initialize 시점에 구독 topic과 함께 생성됩니다.

Producer 수명주기

• 두 transport 모두 producer를 하나만 두고 재사용합니다. 비동기 transport는 IAsyncService로 등록되어 애플리케이션이 서비스를 시작할 때 AIOKafkaProducer를 열고 종료할 때 닫습니다. 시작 시점에 브로커에 연결하지 못하면 app.start()가 실패합니다.
• send()는 레코드를 producer에 넘기고 broker 응답을 기다리지 않으므로 연속 발행이 한 배치로 묶입니다. 배치 끝에서 호출하는 flush()가 배치를 내보내고 결과를 확정합니다. DirectEventBus는 발행 1건마다, outbox relay는 배치 1개마다 flush()를 호출합니다.
• 동기·비동기 send()는 producer가 즉시 보고한 레코드 크기 거부만 EventDeliveryRejectedError로 올립니다. flush()도 동기 delivery callback의 MSG_SIZE_TOO_LARGE와 비동기 delivery future의 MessageSizeTooLargeError만 같은 공통 예외로 변환합니다. 동기 경로의 로컬 producer 큐 포화(BufferError)·그 밖의 KafkaException, 비동기 connection·timeout 오류를 포함한 나머지 실패는 원래 client 예외 타입을 보존하며, 한 배치에 두 종류가 섞이면 transport 장애를 우선합니다. 이 구분으로 영구적이고 특정 레코드에 귀속되는 거부만 호출자의 retry 예산을 소모합니다.
• 비동기 flush()가 확정하는 대상은 그 발행자가 넘긴 레코드입니다. 여러 발행자가 같은 transport Pod를 동시에 쓰더라도 한 발행자의 flush가 다른 발행자의 거부를 대신 받거나 삼키지 않습니다. 따라서 한 배치의 send와 flush는 같은 실행 문맥(asyncio 태스크)에서 호출해야 하며, event bus와 outbox relay가 그렇게 호출합니다.
• 애플리케이션이 transport를 시작하기 전이나 종료한 뒤의 비동기 발행은 EventTransportNotRunningError로 거부됩니다. outbox relay는 이 에러를 전달 실패와 구분하여 retry count를 올리지 않고 배치를 그대로 남깁니다.
• aiokafka는 producer를 생성한 event loop에 묶으므로, 다른 loop(HTTP 요청 핸들러·테스트)에서 발행하면 transport가 producer의 loop로 호출을 넘깁니다.
• dead-letter 발행은 consumer가 자기 producer로 수행하므로 위 수명주기와 무관합니다.

주요 기능

• 자동 topic 생성: 이벤트 타입 이름을 기준으로 topic 생성 (dead-letter topic 포함)
• at-least-once 전달: 멱등 producer + 실패가 저장된 뒤에만 전진하는 명시적 offset 커밋
• Dead-letter 경로: 처리 실패 메시지를 원본 좌표·예외 header와 함께 .dlt topic으로 전달
• 파티션 키 라우팅: AbstractIntegrationEvent.partition_key를 오버라이드하면 같은 키의 이벤트가 같은 파티션으로 가서 순서가 유지됩니다. 기본값 None은 라운드로빈 분산
• 동기/비동기 지원: 동기 및 비동기 publisher/consumer 모두 지원
• Background service 패턴: consumer polling을 background service로 실행
• Pydantic 직렬화: 이벤트를 Pydantic으로 직렬화/역직렬화
• Confluent Kafka client: 안정적인 confluent-kafka library 기반
• 분산 트레이싱: 서비스 간 trace 전파를 위한 spakky-tracing 통합

구성 요소

개발 검증

패키지 단위 검증은 해당 패키지 디렉토리에서 실행합니다.

uv run ruff format .
uv run ruff check .
uv run pyrefly check
uv run pytest

pytest는 각 패키지 pyproject.toml의 coverage 설정을 사용합니다.

라이선스

MIT License