spakky-kafka

Kafka 통합 — 이벤트 전송/수신

메인

initialize(app)

Initialize the spakky-kafka plugin.

Registers Kafka consumers, transports, and post-processor.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/main.py

def initialize(app: SpakkyApplication) -> None:
    """Initialize the spakky-kafka plugin.

    Registers Kafka consumers, transports, and post-processor.
    """
    app.add(KafkaConnectionConfig)

    app.add(KafkaEventConsumer)
    app.add(KafkaEventTransport)

    app.add(AsyncKafkaEventConsumer)
    app.add(AsyncKafkaEventTransport)

    app.add(KafkaPostProcessor)

이벤트 Transport

KafkaEventTransport(config)

Bases: IEventTransport, IService

Synchronous Kafka event transport using confluent_kafka Producer.

One producer is created with the transport and reused for every publish. send() only queues a record for the producer's own batching, so the transport flushes when the application stops, closing the window where a queued record would die with the process.

Initialize the Kafka producer with connection config.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

def __init__(self, config: KafkaConnectionConfig) -> None:
    """Initialize the Kafka producer with connection config."""
    self.config = config
    self.admin = AdminClient(self.config.connection_configuration_dict)
    self.producer = Producer(
        self.config.producer_configuration_dict,
        logger=logger,
    )
    self._delivery_errors = []

set_stop_event(stop_event)

Ignore the shutdown signal: publishing is on demand, with no loop to stop.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def set_stop_event(self, stop_event: Event) -> None:
    """Ignore the shutdown signal: publishing is on demand, with no loop to stop."""

start()

Open nothing: the producer is created together with the transport.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def start(self) -> None:
    """Open nothing: the producer is created together with the transport."""

stop()

Deliver records still queued in the producer before the process exits.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def stop(self) -> None:
    """Deliver records still queued in the producer before the process exits."""
    self.producer.flush()

send(event_name, payload, headers, partition_key=None)

Hand a pre-serialized event payload to the Kafka producer queue.

The payload is queued for the producer's own batching and is confirmed only by flush(), which the caller invokes at the end of its batch.

Parameters:

Name	Type	Description	Default
event_name	str	Topic name (typically the event class name).	required
payload	bytes	Pre-serialized JSON bytes.	required
headers	dict[str, str]	Metadata headers for trace propagation.	required
partition_key	str | None	Key routing the message to one partition. None lets Kafka assign partitions round-robin.	None

Raises:

Type	Description
EventDeliveryRejectedError	The producer rejected this record as too large.
BufferError	The local producer queue is full.
KafkaException	A different synchronous Kafka failure occurred.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def send(
    self,
    event_name: str,
    payload: bytes,
    headers: dict[str, str],
    partition_key: str | None = None,
) -> None:
    """Hand a pre-serialized event payload to the Kafka producer queue.

    The payload is queued for the producer's own batching and is confirmed
    only by flush(), which the caller invokes at the end of its batch.

    Args:
        event_name: Topic name (typically the event class name).
        payload: Pre-serialized JSON bytes.
        headers: Metadata headers for trace propagation.
        partition_key: Key routing the message to one partition. None lets
            Kafka assign partitions round-robin.

    Raises:
        EventDeliveryRejectedError: The producer rejected this record as too
            large.
        BufferError: The local producer queue is full.
        KafkaException: A different synchronous Kafka failure occurred.
    """
    self._create_topic(topic=event_name)
    try:
        self.producer.produce(
            topic=event_name,
            value=payload,
            key=partition_key.encode() if partition_key is not None else None,
            headers=dict(headers),
            callback=self._message_delivery_report,
        )
    except KafkaException as error:
        rejected_records = [
            reason
            for reason in error.args
            if isinstance(reason, KafkaError)
            and reason.code() == KafkaError.MSG_SIZE_TOO_LARGE
        ]
        if not rejected_records:
            raise
        raise EventDeliveryRejectedError(
            [str(reason) for reason in rejected_records]
        ) from error
    self.producer.poll(0)

flush()

Block until the producer has sent every queued record.

A message-size rejection collected by the delivery callback is raised as EventDeliveryRejectedError. Other callback errors remain transport-wide KafkaException failures, so callers do not charge them to a record.

Raises:

Type	Description
EventDeliveryRejectedError	The broker refused a record as too large.
KafkaException	Delivery failed for a transport-wide Kafka error.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def flush(self) -> None:
    """Block until the producer has sent every queued record.

    A message-size rejection collected by the delivery callback is raised as
    `EventDeliveryRejectedError`. Other callback errors remain transport-wide
    `KafkaException` failures, so callers do not charge them to a record.

    Raises:
        EventDeliveryRejectedError: The broker refused a record as too large.
        KafkaException: Delivery failed for a transport-wide Kafka error.
    """
    self.producer.flush()
    if not self._delivery_errors:
        return
    delivery_errors = self._delivery_errors
    self._delivery_errors = []
    transport_errors = [
        error
        for error in delivery_errors
        if error.code() != KafkaError.MSG_SIZE_TOO_LARGE
    ]
    if transport_errors:
        raise KafkaException(transport_errors[0])
    raise EventDeliveryRejectedError([str(error) for error in delivery_errors])

AsyncKafkaEventTransport(config)

Bases: IAsyncEventTransport, IAsyncService

Asynchronous Kafka event transport using aiokafka AIOKafkaProducer.

The producer lives as long as the application: it is opened when the application starts services and closed when the application stops them. A producer per publish would reopen the broker connection every time, defeat batching, and reset the idempotent producer sequence, which is bound to a producer instance lifetime.

send() hands a record over without waiting for its broker acknowledgement, so consecutive publishes fill one batch; flush() sends the batch out and reports whichever record the broker rejected. A publisher only ever learns the outcome of the records it handed over itself, because concurrent publishers share this Pod and one publisher's flush must not swallow another's rejected record.

aiokafka binds a producer to the event loop that created it, while the ApplicationContext runs its services on an internal event loop of its own. Publishers living on another loop (HTTP request handlers, tests) therefore have their producer calls routed back to the producer's own loop.

Initialize the async Kafka transport with connection config.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

def __init__(self, config: KafkaConnectionConfig) -> None:
    """Initialize the async Kafka transport with connection config."""
    self.config = config
    self.admin = AdminClient(self.config.connection_configuration_dict)
    self._running = None

set_stop_event(stop_event)

Ignore the shutdown signal: publishing is on demand, with no loop to stop.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
def set_stop_event(self, stop_event: locks.Event) -> None:
    """Ignore the shutdown signal: publishing is on demand, with no loop to stop."""

start_async() async

Open the producer that serves every publish until the application stops.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
async def start_async(self) -> None:
    """Open the producer that serves every publish until the application stops."""
    producer = AIOKafkaProducer(**self.config.async_producer_configuration_dict)
    await producer.start()
    self._running = _LoopBoundProducer(producer=producer, loop=get_running_loop())

stop_async() async

Deliver what is still buffered, then close the producer.

The transport stops accepting publishes before the producer closes, so a service still running during shutdown is told the transport stopped instead of being handed a delivery failure.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
async def stop_async(self) -> None:
    """Deliver what is still buffered, then close the producer.

    The transport stops accepting publishes before the producer closes, so a
    service still running during shutdown is told the transport stopped
    instead of being handed a delivery failure.
    """
    running = self._running_producer
    self._running = None
    await self._on_producer_loop(
        self._close_producer(running.producer, self._claim_deliveries()),
        running.loop,
    )

send(event_name, payload, headers, partition_key=None) async

Hand a pre-serialized event payload to the long-lived Kafka producer.

The record joins the producer's current batch and is confirmed only by flush(), which the caller invokes at the end of its batch.

Parameters:

Name	Type	Description	Default
event_name	str	Topic name (typically the event class name).	required
payload	bytes	Pre-serialized JSON bytes.	required
headers	dict[str, str]	Metadata headers for trace propagation.	required
partition_key	str | None	Key routing the message to one partition. None lets Kafka assign partitions round-robin.	None

Raises:

Type	Description
EventTransportNotRunningError	When the application has not started the transport's producer, or has already stopped it.
EventDeliveryRejectedError	The producer rejected this record as too large.
KafkaError	Any producer failure not attributable to this record's size, propagated without conversion.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
async def send(
    self,
    event_name: str,
    payload: bytes,
    headers: dict[str, str],
    partition_key: str | None = None,
) -> None:
    """Hand a pre-serialized event payload to the long-lived Kafka producer.

    The record joins the producer's current batch and is confirmed only by
    flush(), which the caller invokes at the end of its batch.

    Args:
        event_name: Topic name (typically the event class name).
        payload: Pre-serialized JSON bytes.
        headers: Metadata headers for trace propagation.
        partition_key: Key routing the message to one partition. None lets
            Kafka assign partitions round-robin.

    Raises:
        EventTransportNotRunningError: When the application has not started
            the transport's producer, or has already stopped it.
        EventDeliveryRejectedError: The producer rejected this record as too
            large.
        aiokafka.errors.KafkaError: Any producer failure not attributable to
            this record's size, propagated without conversion.
    """
    self._create_topic(topic=event_name)
    running = self._running_producer
    try:
        delivery = await self._on_producer_loop(
            running.producer.send(
                topic=event_name,
                value=payload,
                key=partition_key.encode() if partition_key is not None else None,
                headers=[(k, v.encode()) for k, v in headers.items()],
            ),
            running.loop,
        )
    except MessageSizeTooLargeError as error:
        raise EventDeliveryRejectedError([str(error)]) from error
    self._record_delivery(delivery)

flush() async

Block until the producer sent every record this publisher handed over.

Raises:

Type	Description
EventTransportNotRunningError	When the application has not started the transport's producer, or has already stopped it.
EventDeliveryRejectedError	When the broker rejected one or more of this publisher's records as too large. Any other delivery failure keeps its original exception type.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/transport.py

@override
async def flush(self) -> None:
    """Block until the producer sent every record this publisher handed over.

    Raises:
        EventTransportNotRunningError: When the application has not started
            the transport's producer, or has already stopped it.
        EventDeliveryRejectedError: When the broker rejected one or more of
            this publisher's records as too large. Any other delivery failure
            keeps its original exception type.
    """
    running = self._running_producer
    await self._on_producer_loop(
        self._confirm_deliveries(running.producer, self._claim_deliveries()),
        running.loop,
    )

이벤트 Consumer

MessageOutcome

Bases: StrEnum

What running the handlers settled about one consumed message.

The consumer turns this into the offset action that gives the delivery guarantee: an offset must never move past a message whose failure is not stored anywhere else.

PROCESSED = 'processed' class-attribute instance-attribute

The consumer is done with the message and commits its offset.

Covers handler success, every failure a retry cannot fix (a refused auth boundary, an empty message body), and a failure whose dead-letter record reached the broker.

RETRYABLE = 'retryable' class-attribute instance-attribute

The message is still the only copy of its own failure, so it comes back.

Reached when dead-lettering itself failed. The consumer rewinds its position to the message instead of committing, and the next poll delivers it again.

KafkaEventConsumer(config)

Bases: IEventConsumer, AbstractBackgroundService

Synchronous Kafka event consumer that polls messages and dispatches to handlers.

Initialize the Kafka consumer with connection config.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def __init__(self, config: KafkaConnectionConfig) -> None:
    """Initialize the Kafka consumer with connection config."""
    super().__init__()
    self.config = config
    self.type_lookup = {}
    self.type_adapters = {}
    self.handlers = {}
    self._propagator = None
    self._auth_boundary_handlers = set()
    self.admin = AdminClient(self.config.connection_configuration_dict)
    self.consumer = Consumer(
        self.config.consumer_configuration_dict,
        logger=logger,
    )

producer instance-attribute

Dead-letter producer, bound by initialize for the service lifetime.

set_propagator(propagator)

Set the trace propagator for extracting trace context from messages.

Parameters:

Name	Type	Description	Default
propagator	ITracePropagator	An ITracePropagator instance.	required

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def set_propagator(self, propagator: ITracePropagator) -> None:
    """Set the trace propagator for extracting trace context from messages.

    Args:
        propagator: An ITracePropagator instance.
    """
    self._propagator = propagator

register_auth_boundary(handler)

Mark a registered post-processor endpoint as Kafka auth-aware.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def register_auth_boundary(self, handler: EventHandlerCallback[Any]) -> None:
    """Mark a registered post-processor endpoint as Kafka auth-aware."""
    self._auth_boundary_handlers.add(handler)

register(event, handler)

Register a handler for the given event type.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
def register[EventT_contra: AbstractEvent](
    self,
    event: type[EventT_contra],
    handler: EventHandlerCallback[EventT_contra],
) -> None:
    """Register a handler for the given event type."""
    if event not in self.handlers:
        self.handlers[event] = []
        self.type_adapters[event] = cast(
            TypeAdapter[AbstractEvent], TypeAdapter(event)
        )
        self.type_lookup[_event_routing_name(event)] = event
    self.handlers[event].append(handler)

initialize()

Create Kafka topics, open the dead-letter producer, and subscribe.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
def initialize(self) -> None:
    """Create Kafka topics, open the dead-letter producer, and subscribe."""
    topics: list[str] = [
        _event_routing_name(event_type) for event_type in self.handlers.keys()
    ]
    self.producer = Producer(self.config.producer_configuration_dict, logger=logger)
    self._create_topics(
        topics=topics
        + [f"{topic}{self.config.dead_letter_topic_suffix}" for topic in topics]
    )
    self.consumer.subscribe(topics=topics)

run()

Poll Kafka for messages and route them to registered handlers.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
def run(self) -> None:
    """Poll Kafka for messages and route them to registered handlers."""
    while not self._stop_event.is_set():
        message: Message | None = self.consumer.poll(
            timeout=self.config.poll_timeout
        )
        if message is None:
            continue
        self._route_event_handler(message)

dispose()

Flush pending dead-letter records and close the Kafka consumer.

The flush is bounded so an unreachable broker cannot block shutdown forever; records still queued at that point are reported by the delivery callback.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
def dispose(self) -> None:
    """Flush pending dead-letter records and close the Kafka consumer.

    The flush is bounded so an unreachable broker cannot block shutdown
    forever; records still queued at that point are reported by the
    delivery callback.
    """
    self.producer.flush(self.config.dead_letter_delivery_timeout)
    self.consumer.close()

AsyncKafkaEventConsumer(config)

Bases: IAsyncEventConsumer, AbstractAsyncBackgroundService

Asynchronous Kafka event consumer that polls messages and dispatches to handlers.

Initialize the async Kafka consumer with connection config.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def __init__(self, config: KafkaConnectionConfig) -> None:
    """Initialize the async Kafka consumer with connection config."""
    super().__init__()
    self.config = config
    self.type_lookup = {}
    self.type_adapters = {}
    self.handlers = {}
    self._propagator = None
    self._auth_boundary_handlers = set()
    self.admin = AdminClient(self.config.connection_configuration_dict)

producer instance-attribute

Dead-letter producer, bound by initialize_async for the service lifetime.

set_propagator(propagator)

Set the trace propagator for extracting trace context from messages.

Parameters:

Name	Type	Description	Default
propagator	ITracePropagator	An ITracePropagator instance.	required

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def set_propagator(self, propagator: ITracePropagator) -> None:
    """Set the trace propagator for extracting trace context from messages.

    Args:
        propagator: An ITracePropagator instance.
    """
    self._propagator = propagator

register_auth_boundary(handler)

Mark a registered post-processor endpoint as Kafka auth-aware.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

def register_auth_boundary(self, handler: AsyncEventHandlerCallback[Any]) -> None:
    """Mark a registered post-processor endpoint as Kafka auth-aware."""
    self._auth_boundary_handlers.add(handler)

register(event, handler)

Register an async handler for the given event type.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
def register[EventT_contra: AbstractEvent](
    self,
    event: type[EventT_contra],
    handler: AsyncEventHandlerCallback[EventT_contra],
) -> None:
    """Register an async handler for the given event type."""
    if event not in self.handlers:
        self.handlers[event] = []
        self.type_adapters[event] = cast(
            TypeAdapter[AbstractEvent], TypeAdapter(event)
        )
        self.type_lookup[_event_routing_name(event)] = event
    self.handlers[event].append(handler)

initialize_async() async

Create Kafka topics, open the dead-letter producer, and subscribe.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
async def initialize_async(self) -> None:
    """Create Kafka topics, open the dead-letter producer, and subscribe."""
    self.consumer = AIOConsumer(self.config.consumer_configuration_dict)
    self.producer = AIOKafkaProducer(
        **self.config.async_producer_configuration_dict
    )
    await self.producer.start()
    topics: list[str] = [
        _event_routing_name(event_type) for event_type in self.handlers.keys()
    ]
    self._create_topics(
        topics=topics
        + [f"{topic}{self.config.dead_letter_topic_suffix}" for topic in topics]
    )
    await self.consumer.subscribe(topics=topics)

run_async() async

Poll Kafka asynchronously for messages and route them to handlers.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
async def run_async(self) -> None:  # pragma: no cover - 별도 asyncio 태스크로 실행
    """Poll Kafka asynchronously for messages and route them to handlers."""
    while not self._stop_event.is_set():
        message: Message | None = await self.consumer.poll(
            timeout=self.config.poll_timeout
        )
        if message is None:
            continue
        await self._route_event_handler(message)

dispose_async() async

Flush pending dead-letter records and close the async Kafka consumer.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/event/consumer.py

@override
async def dispose_async(self) -> None:
    """Flush pending dead-letter records and close the async Kafka consumer."""
    await self.producer.stop()
    await self.consumer.close()

Auth Boundary

Authentication helpers for Kafka consumer boundaries.

KAFKA_AUTH_BOUNDARY = 'kafka' module-attribute

Provider-neutral boundary name used for Kafka AuthInvocation values.

KAFKA_AUTH_HEADERS_PARAMETER = '_spakky_kafka_headers' module-attribute

Internal endpoint keyword used to pass Kafka message headers.

KafkaHandlerAuthBinding(*, operation, protected) dataclass

Auth metadata captured for one registered Kafka event handler.

operation instance-attribute

Canonical handler operation reference.

protected instance-attribute

Whether the handler has effective protected auth metadata.

KafkaAuthBoundary(container, application_context)

Verify propagated snapshots and seed AuthContext for Kafka handlers.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/auth.py

def __init__(
    self,
    container: IContainer,
    application_context: IApplicationContext,
) -> None:
    self._container = container
    self._application_context = application_context

seed_auth_context(headers, binding)

Verify a Kafka snapshot header and store AuthContext before user code.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/auth.py

def seed_auth_context(
    self,
    headers: dict[str, str],
    binding: KafkaHandlerAuthBinding,
) -> AuthorizationDecision:
    """Verify a Kafka snapshot header and store AuthContext before user code."""
    snapshot = self._snapshot_header(headers)
    if snapshot is None:
        if binding.protected:
            return MISSING_SNAPSHOT_DECISION
        return AuthorizationDecision.allow()
    verifier = self._container.get_or_none(IAuthContextSnapshotVerifier)
    if verifier is None:
        raise AuthVerificationProviderUnavailableError()
    try:
        auth_context = verifier.verify_snapshot(
            snapshot,
            self._invocation(binding),
        )
    except MissingAuthContextSnapshotError:
        return MISSING_SNAPSHOT_DECISION
    except InvalidAuthContextSnapshotError:
        return INVALID_SNAPSHOT_DECISION
    except ExpiredAuthContextSnapshotError:
        return EXPIRED_SNAPSHOT_DECISION
    except AuthVerificationProviderUnavailableError:
        raise
    store_auth_context(self._application_context, auth_context)
    return AuthorizationDecision.allow()

설정

Configuration for Kafka connections.

Provides configuration dataclass for Kafka connection parameters including bootstrap servers, consumer group, and security settings.

AutoOffsetResetType

Bases: StrEnum

Kafka consumer auto offset reset policies.

KafkaConnectionConfig()

Bases: BaseSettings

Kafka connection configuration loaded from environment variables.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/common/config.py

def __init__(self) -> None:
    super().__init__()

group_id instance-attribute

Kafka consumer group identifier.

client_id instance-attribute

Kafka client identifier.

bootstrap_servers instance-attribute

Kafka bootstrap servers.

security_protocol = None class-attribute instance-attribute

Security protocol for Kafka connection.

sasl_mechanism = None class-attribute instance-attribute

SASL mechanism for Kafka authentication.

sasl_username = None class-attribute instance-attribute

SASL username for Kafka authentication.

sasl_password = None class-attribute instance-attribute

SASL password for Kafka authentication.

number_of_partitions = 1 class-attribute instance-attribute

Default number of partitions for created topics.

replication_factor = 1 class-attribute instance-attribute

Default replication factor for created topics.

auto_offset_reset = AutoOffsetResetType.EARLIEST class-attribute instance-attribute

Consumer auto offset reset policy (earliest, latest, none).

poll_timeout = 1.0 class-attribute instance-attribute

Consumer poll timeout in seconds.

dead_letter_topic_suffix = '.dlt' class-attribute instance-attribute

Suffix appended to the original topic name to build its dead-letter topic.

An empty suffix would make the dead-letter topic the original topic itself, republishing every failure back into the stream it came from, so it is rejected.

max_handler_retries = 0 class-attribute instance-attribute

How many times a failed handler dispatch is retried before dead-lettering.

Retrying re-invokes every handler registered for the event, so handlers must be idempotent when this is raised above zero. Deserialization failures never succeed on retry and are dead-lettered immediately regardless of this value.

dead_letter_delivery_timeout = 10.0 class-attribute instance-attribute

Seconds to wait for a dead-letter record to reach the broker.

Bounds how long the consumer blocks on one failed message; without it a broker outage would stall the poll loop until the client's own message timeout.

connection_configuration_dict property

librdkafka settings that only describe how to reach the cluster.

Shared by the admin client, the producer, and the consumer. Delivery semantics live in producer_configuration_dict and consumer_configuration_dict instead, because the safe producer default and the safe consumer default are opposites of each other.

producer_configuration_dict property

Producer settings that keep retries from duplicating or reordering.

enable.idempotence makes the broker deduplicate retried batches and pins max.in.flight.requests.per.connection low enough that a retry cannot overtake an earlier batch on the same partition. acks=all waits for every in-sync replica so an acknowledged event survives the loss of the partition leader.

consumer_configuration_dict property

Consumer settings that tie the offset to the handler outcome.

enable.auto.commit=false is what makes delivery at-least-once: the offset advances only when KafkaEventConsumer commits it after the handlers finish, so a handler failure leaves the message to be redelivered instead of silently dropping it.

async_producer_configuration_dict property

The same producer guarantees expressed as aiokafka keyword names.

aiokafka takes constructor keywords instead of a librdkafka property dictionary, so the key spelling differs while the delivery guarantee is identical to producer_configuration_dict.

후처리기

KafkaPostProcessor

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Post-processor that registers event handlers with Kafka consumers.

Scans @EventHandler decorated classes for @event decorated methods and automatically registers them with the appropriate Kafka consumer (sync or async) with proper dependency injection.

set_container(container)

Set the container for dependency injection.

Parameters:

Name	Type	Description	Default
container	IContainer	The IoC container.	required

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/post_processor.py

@override
def set_container(self, container: IContainer) -> None:
    """Set the container for dependency injection.

    Args:
        container: The IoC container.
    """
    self.__container = container

set_application_context(application_context)

Set the application context.

Parameters:

Name	Type	Description	Default
application_context	IApplicationContext	The application context instance.	required

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/post_processor.py

@override
def set_application_context(self, application_context: IApplicationContext) -> None:
    """Set the application context.

    Args:
        application_context: The application context instance.
    """
    self.__application_context = application_context

post_process(pod)

Register event handlers from event handler classes.

Scans the event handler for methods decorated with @on_event and registers them with the appropriate Kafka consumer (sync or async) based on whether the method is a coroutine function.

Parameters:

Name	Type	Description	Default
pod	object	The Pod to process.	required

Returns:

Type	Description
object	The Pod, with event handlers registered if it's an event handler.

Source code in plugins/spakky-kafka/src/spakky/plugins/kafka/post_processor.py

@override
def post_process(self, pod: object) -> object:
    """Register event handlers from event handler classes.

    Scans the event handler for methods decorated with @on_event and registers
    them with the appropriate Kafka consumer (sync or async) based on
    whether the method is a coroutine function.

    Args:
        pod: The Pod to process.

    Returns:
        The Pod, with event handlers registered if it's an event handler.
    """
    if not EventHandler.exists(pod):
        return pod
    handler: EventHandler = EventHandler.get(pod)
    consumer = self.__container.get(IEventConsumer)
    async_consumer = self.__container.get(IAsyncEventConsumer)
    auth_boundary = KafkaAuthBoundary(self.__container, self.__application_context)
    propagator = self.__application_context.get_or_none(ITracePropagator)
    if propagator is not None:
        if hasattr(  # optional tracing bridge injection
            consumer, "set_propagator"
        ):  # 프레임워크 내부: consumer가 선택적 propagator를 지원하는지 확인
            consumer.set_propagator(propagator)
        if hasattr(  # optional tracing bridge injection
            async_consumer, "set_propagator"
        ):  # 프레임워크 내부: consumer가 선택적 propagator를 지원하는지 확인
            async_consumer.set_propagator(propagator)
    for name, method in getmembers(pod, ismethod):
        route: EventRoute[AbstractEvent] | None = EventRoute[
            AbstractEvent
        ].get_or_none(method)
        if route is None:
            continue
        if not issubclass(route.event_type, AbstractIntegrationEvent):
            continue

        # pylint: disable=line-too-long
        logger.info(
            f"[{type(self).__name__}] {route.event_type.__name__} -> {method.__qualname__}"
        )
        auth_metadata = get_effective_auth_metadata(
            method,
            owner_type=handler.type_,
        )
        auth_binding = KafkaHandlerAuthBinding(
            operation=f"{handler.type_.__module__}.{handler.type_.__qualname__}.{name}",
            protected=auth_metadata.protected,
        )

        if iscoroutinefunction(method):

            @wraps(method)
            async def async_endpoint(
                *args: Any,
                _spakky_kafka_headers: dict[str, str] | None = None,
                method_name: str = name,
                controller_type: type[object] = handler.type_,
                context: IContainer = self.__container,
                boundary: KafkaAuthBoundary = auth_boundary,
                binding: KafkaHandlerAuthBinding = auth_binding,
                **kwargs: Any,
            ) -> Any:
                # Each message is handled in isolation, so clear the
                # application context to avoid reusing dependency state.
                self.__application_context.clear_context()
                decision = boundary.seed_auth_context(
                    _spakky_kafka_headers or {},
                    binding,
                )
                if decision.state is not AuthorizationDecisionState.ALLOW:
                    return None
                controller_instance = context.get(controller_type)
                method_to_call = getattr(  # event handler method lookup
                    controller_instance, method_name
                )  # 프레임워크 내부: 이벤트 핸들러 메서드 동적 디스패치
                return await method_to_call(*args, **kwargs)

            async_consumer.register(route.event_type, async_endpoint)
            if hasattr(  # optional auth-aware Kafka consumer bridge
                async_consumer,
                "register_auth_boundary",
            ):
                async_consumer.register_auth_boundary(async_endpoint)
            continue

        @wraps(method)
        def endpoint(
            *args: Any,
            _spakky_kafka_headers: dict[str, str] | None = None,
            method_name: str = name,
            controller_type: type[object] = handler.type_,
            context: IContainer = self.__container,
            boundary: KafkaAuthBoundary = auth_boundary,
            binding: KafkaHandlerAuthBinding = auth_binding,
            **kwargs: Any,
        ) -> Any:
            # Synchronous consumers share threads, so drop any lingering
            # scoped data before invoking the handler.
            self.__application_context.clear_context()
            decision = boundary.seed_auth_context(
                _spakky_kafka_headers or {},
                binding,
            )
            if decision.state is not AuthorizationDecisionState.ALLOW:
                return None
            controller_instance = context.get(controller_type)
            method_to_call = getattr(  # async event handler method lookup
                controller_instance, method_name
            )  # 프레임워크 내부: 이벤트 핸들러 메서드 동적 디스패치
            return method_to_call(*args, **kwargs)

        consumer.register(route.event_type, endpoint)
        if hasattr(  # optional auth-aware Kafka consumer bridge
            consumer,
            "register_auth_boundary",
        ):
            consumer.register_auth_boundary(endpoint)
    return pod

추가 모듈

DeadLetterHeaderKey

Bases: StrEnum

Kafka header keys describing where a dead-lettered message came from.

A reprocessing tool must be able to decide what to do from these headers alone, without parsing the original message body.