spakky-kafka¶

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

메인¶

`spakky.plugins.kafka.main` ¶

`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)

options: show_root_heading: false

이벤트 Transport¶

`spakky.plugins.kafka.event.transport` ¶

`KafkaEventTransport(config)` ¶

Bases: IEventTransport

Synchronous Kafka event transport using confluent_kafka Producer.

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.configuration_dict)
    self.producer = Producer(self.config.configuration_dict, logger=logger)

`send(event_name, payload, headers)` ¶

Send a pre-serialized event payload to Kafka.

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

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],
) -> None:
    """Send a pre-serialized event payload to Kafka.

    Args:
        event_name: Topic name (typically the event class name).
        payload: Pre-serialized JSON bytes.
        headers: Metadata headers for trace propagation.
    """
    self._create_topic(topic=event_name)
    self.producer.produce(
        topic=event_name,
        value=payload,
        headers=dict(headers),
        callback=self._message_delivery_report,
    )
    self.producer.poll(0)
    self.producer.flush()

`AsyncKafkaEventTransport(config)` ¶

Bases: IAsyncEventTransport

Asynchronous Kafka event transport using aiokafka AIOKafkaProducer.

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.configuration_dict)

`send(event_name, payload, headers)` `async` ¶

Asynchronously send a pre-serialized event payload to Kafka.

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

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],
) -> None:
    """Asynchronously send a pre-serialized event payload to Kafka.

    Args:
        event_name: Topic name (typically the event class name).
        payload: Pre-serialized JSON bytes.
        headers: Metadata headers for trace propagation.
    """
    self._create_topic(topic=event_name)
    producer = AIOKafkaProducer(
        bootstrap_servers=self.config.bootstrap_servers,
    )
    await producer.start()
    try:
        await producer.send_and_wait(
            topic=event_name,
            value=payload,
            headers=[(k, v.encode()) for k, v in headers.items()],
        )
    finally:
        await producer.stop()

options: show_root_heading: false

이벤트 Consumer¶

`spakky.plugins.kafka.event.consumer` ¶