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"

사용법

이벤트 발행

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 역직렬화 의미는 그대로 유지됩니다.

주요 기능

• 자동 topic 생성: 이벤트 타입 이름을 기준으로 topic 생성
• 동기/비동기 지원: 동기 및 비동기 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