Kafka + FastAPI: введение в event-driven архитектуру

@app.post("/orders")
async def create_order(order: OrderCreate):
    db_order = await save_order(order)
 
    await send_email(order.user_email)        # 500ms latency
    await update_inventory(order.items)       # 300ms latency
    await calculate_bonuses(order.user_id)    # 200ms latency
    await notify_delivery(order.id)           # 400ms latency
 
    return db_order  # Пользователь ждёт 1.4 секунды

# Producer отправил, но не дождался подтверждения
producer.send('topic', value, acks=0)  # Fire and forget

# Producer ждёт подтверждения, но может повторить при таймауте
producer.send('topic', value, acks=1)  # Leader подтвердил

# Идемпотентный producer + транзакции
producer = KafkaProducer(
    enable_idempotence=True,
    transactional_id='my-transactional-id'
)

# ❌ Плохо: пользователь ждёт завершения всех операций
@app.post("/users/register")
async def register(user: UserCreate):
    db_user = await create_user(user)  # 50ms
 
    await send_welcome_email(user.email)      # 500ms — ждём SMTP
    await create_bonus_account(db_user.id)    # 200ms — запрос в другую БД
 
    return db_user  # Пользователь ждал 750ms вместо 50ms

# ✅ Хорошо: API отвечает быстро, обработка в фоне
@app.post("/users/register")
async def register(user: UserCreate):
    db_user = await create_user(user)  # 50ms
 
    # Отправили событие в Kafka — занимает ~5ms
    await producer.send('users.registered', {
        'user_id': db_user.id,
        'email': user.email,
        'timestamp': datetime.utcnow().isoformat()
    })
 
    return db_user  # Ответ за 55ms вместо 750ms
 
# Отдельный consumer (можно в том же приложении или отдельном)
async def handle_user_registered(event):
    # Обрабатываем в фоне, не блокируя API
    await send_welcome_email(event['email'])
    await create_bonus_account(event['user_id'])
    # Если упадёт — повторим автоматически (at-least-once)

# ❌ Плохо: Order Service жёстко связан с другими сервисами
@app.post("/orders")
async def create_order(order: OrderCreate):
    db_order = await save_order(order)
 
    # Прямой HTTP вызов → если Inventory упал, заказ не создаётся
    await http_client.post("http://inventory-service/reserve", {...})
    await http_client.post("http://notification-service/send", {...})
 
    return db_order

# ✅ Хорошо: Order Service публикует событие и забывает
@app.post("/orders")
async def create_order(order: OrderCreate):
    db_order = await save_order(order)
 
    # Отправили событие → ответили пользователю мгновенно
    await producer.send('orders.created', {
        'order_id': db_order.id,
        'user_id': order.user_id,
        'items': order.items,
        'total': order.total
    })
 
    return db_order  # Не ждём Inventory и Notification
 
# Inventory Service — отдельное приложение, которое слушает события
async def handle_order_created(event):
    await reserve_items(event['items'])
    # Если упадёт — не повлияет на создание заказа
 
# Notification Service — ещё одно отдельное приложение
async def handle_order_created(event):
    await send_order_confirmation(event['order_id'])
    # Можно добавить новый сервис, не меняя Order Service

# ❌ Традиционный подход: UPDATE затирает историю
@app.post("/transfer")
async def transfer_money(from_account: str, to_account: str, amount: float):
    # Было: 5000 руб
    await db.execute(
        "UPDATE accounts SET balance = balance - ? WHERE id = ?",
        (amount, from_account)
    )
    # Стало: 4000 руб
    # Потеряли информацию: кто, когда, зачем списал 1000 руб

# ✅ Event Sourcing: храним все события
@app.post("/transfer")
async def transfer_money(from_account: str, to_account: str, amount: float):
    transfer_id = generate_uuid()
 
    # Событие 1: Списание
    await producer.send('accounts.debited', {
        'transfer_id': transfer_id,
        'account_id': from_account,
        'amount': -1000,
        'timestamp': datetime.utcnow().isoformat(),
        'description': f'Перевод на {to_account}'
    })
 
    # Событие 2: Зачисление
    await producer.send('accounts.credited', {
        'transfer_id': transfer_id,
        'account_id': to_account,
        'amount': 1000,
        'timestamp': datetime.utcnow().isoformat(),
        'description': f'Перевод от {from_account}'
    })
 
# Consumer восстанавливает текущее состояние из событий
async def rebuild_balance(account_id: str) -> float:
    events = await get_account_events(account_id)  # Читаем из Kafka
    balance = 0
    for event in events:
        if event['type'] == 'debited':
            balance -= event['amount']
        elif event['type'] == 'credited':
            balance += event['amount']
    return balance

# ❌ Традиционный подход: пишем каждый клик в БД
@app.get("/products/{item_id}")
async def get_product(item_id: int, user_id: int):
    # Записываем клик в PostgreSQL
    await db.execute(
        "INSERT INTO page_views (user_id, page, timestamp) VALUES (?, ?, ?)",
        (user_id, f'/products/{item_id}', datetime.utcnow())
    )
    # При 1000 RPS это 1000 INSERT/s в БД — узкое горло
 
    # Получаем популярные товары — медленный запрос
    popular = await db.execute(
        "SELECT page, COUNT(*) FROM page_views WHERE timestamp > now() - interval '1 hour' GROUP BY page"
    )
    # На больших объёмах это секунды ожидания

# ✅ Хорошо: пишем события в Kafka, агрегируем в памяти
@app.get("/products/{item_id}")
async def get_product(item_id: int, user_id: int):
    # Просто отправили событие — занимает 1-5ms
    await producer.send('user.clicks', {
        'user_id': user_id,
        'page': f'/products/{item_id}',
        'timestamp': time.time(),
        'session_id': request.cookies.get('session_id')
    })
 
    # Читаем популярные товары из Redis (обновляется consumer)
    popular = await redis.get('popular_products_1h')  # < 1ms
    return popular
 
# Consumer агрегирует клики в реальном времени
page_counters = defaultdict(int)  # В памяти
 
async def process_clicks(event):
    page = event['page']
 
    # Инкрементируем счётчик в памяти
    page_counters[page] += 1
 
    # Раз в 10 секунд сбрасываем в Redis
    if should_flush():
        top_pages = sorted(page_counters.items(), key=lambda x: x[1], reverse=True)[:10]
        await redis.set('popular_products_1h', json.dumps(top_pages))
 
    # Трекаем путь пользователя для рекомендаций
    await track_user_journey(event['user_id'], page)

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
 
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9093:9093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
 
  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    depends_on:
      - kafka
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9093
      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181

docker compose up -d
 
# Проверка статуса
docker compose ps
 
# Логи
docker compose logs -f kafka
 
# UI для мониторинга: http://localhost:8080

# Через docker exec
docker compose exec kafka kafka-topics \
  --create \
  --topic orders \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# Просмотр всех топиков
docker compose exec kafka kafka-topics \
  --list \
  --bootstrap-server localhost:9092
 
# Детали топика: партиции, реплики, конфигурация
docker compose exec kafka kafka-topics \
  --describe \
  --topic orders \
  --bootstrap-server localhost:9092
 
# Вывод:
# Topic: orders  PartitionCount: 3  ReplicationFactor: 1
#   Partition: 0  Leader: 1  Replicas: 1  Isr: 1
#   Partition: 1  Leader: 1  Replicas: 1  Isr: 1
#   Partition: 2  Leader: 1  Replicas: 1  Isr: 1

pip install aiokafka
 
# Для сериализации JSON
pip install aiokafka orjson
 
# Для Avro схем (optional, для production)
pip install aiokafka fastavro

from pydantic_settings import BaseSettings, SettingsConfigDict
 
class Settings(BaseSettings):
    # Kafka
    kafka_bootstrap_servers: list[str] = ["localhost:9092"]
    kafka_consumer_group: str = "fastapi-app"
    kafka_auto_offset_reset: str = "earliest"  # "earliest" или "latest"
 
    # Topics
    orders_topic: str = "orders"
    notifications_topic: str = "notifications"
 
    # App
    app_name: str = "Kafka FastAPI"
    debug: bool = False
 
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )
 
settings = Settings()

import json
import logging
from typing import Any, Optional
 
from aiokafka import AIOKafkaProducer
from aiokafka.errors import KafkaError
 
from app.config import settings
 
logger = logging.getLogger(__name__)
 
class KafkaProducerClient:
    """
    Singleton Kafka Producer для отправки событий.
 
    Использование:
        producer = await get_kafka_producer()
        await producer.send('orders', {'order_id': 123})
    """
 
    def __init__(self):
        self.producer: Optional[AIOKafkaProducer] = None
 
    async def start(self):
        """Создаёт и запускает producer"""
        self.producer = AIOKafkaProducer(
            bootstrap_servers=settings.kafka_bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            # Гарантии доставки
            acks='all',  # Ждём подтверждения от всех реплик
            retries=3,   # Повторные попытки
            # Производительность
            compression_type='gzip',
            max_batch_size=16384,
            linger_ms=10,  # Ждём 10ms для батчинга
        )
        await self.producer.start()
        logger.info("Kafka Producer started")
 
    async def stop(self):
        """Останавливает producer"""
        if self.producer:
            await self.producer.stop()
            logger.info("Kafka Producer stopped")
 
    async def send(
        self,
        topic: str,
        value: dict[str, Any],
        key: Optional[str] = None,
    ) -> None:
        """
        Отправляет событие в Kafka.
 
        Args:
            topic: Имя топика
            value: Payload события (будет сериализован в JSON)
            key: Ключ для партиционирования (optional)
        """
        if not self.producer:
            raise RuntimeError("Producer not started")
 
        try:
            # Ключ определяет партицию (события с одним ключом → одна партиция)
            key_bytes = key.encode('utf-8') if key else None
 
            await self.producer.send_and_wait(
                topic=topic,
                value=value,
                key=key_bytes,
            )
 
            logger.debug(f"Sent to {topic}: {value}")
 
        except KafkaError as e:
            logger.error(f"Failed to send to {topic}: {e}")
            raise
 
# Singleton instance
_producer_client: Optional[KafkaProducerClient] = None
 
async def get_kafka_producer() -> KafkaProducerClient:
    """Dependency injection для FastAPI"""
    global _producer_client
 
    if _producer_client is None:
        _producer_client = KafkaProducerClient()
        await _producer_client.start()
 
    return _producer_client

import asyncio
import json
import logging
from typing import Callable, Awaitable
 
from aiokafka import AIOKafkaConsumer
from aiokafka.errors import KafkaError
 
from app.config import settings
 
logger = logging.getLogger(__name__)
 
class KafkaConsumerClient:
    """
    Kafka Consumer для чтения событий.
 
    Использование:
        consumer = KafkaConsumerClient('orders', handler_func)
        await consumer.start()
    """
 
    def __init__(
        self,
        topic: str,
        handler: Callable[[dict], Awaitable[None]],
        group_id: Optional[str] = None,
    ):
        self.topic = topic
        self.handler = handler
        self.group_id = group_id or settings.kafka_consumer_group
        self.consumer: Optional[AIOKafkaConsumer] = None
        self._running = False
 
    async def start(self):
        """Создаёт consumer и начинает обработку"""
        self.consumer = AIOKafkaConsumer(
            self.topic,
            bootstrap_servers=settings.kafka_bootstrap_servers,
            group_id=self.group_id,
            auto_offset_reset=settings.kafka_auto_offset_reset,
            enable_auto_commit=False,  # Ручной commit для надёжности
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        )
 
        await self.consumer.start()
        logger.info(f"Consumer started for topic: {self.topic}")
 
        self._running = True
        asyncio.create_task(self._consume_loop())
 
    async def stop(self):
        """Останавливает consumer"""
        self._running = False
        if self.consumer:
            await self.consumer.stop()
            logger.info(f"Consumer stopped for topic: {self.topic}")
 
    async def _consume_loop(self):
        """Основной цикл чтения событий"""
        try:
            async for message in self.consumer:
                try:
                    # Обрабатываем событие
                    await self.handler(message.value)
 
                    # Commit offset после успешной обработки
                    await self.consumer.commit()
 
                    logger.debug(
                        f"Processed message from {self.topic}: "
                        f"partition={message.partition}, offset={message.offset}"
                    )
 
                except Exception as e:
                    logger.error(f"Error processing message: {e}", exc_info=True)
                    # Не делаем commit → сообщение будет перечитано
 
        except KafkaError as e:
            logger.error(f"Kafka error: {e}")
        except Exception as e:
            logger.error(f"Unexpected error: {e}", exc_info=True)

from contextlib import asynccontextmanager
 
from fastapi import FastAPI
 
from app.config import settings
from app.kafka.producer import get_kafka_producer, _producer_client
from app.consumers.order_handler import start_order_consumer, stop_order_consumer
from app.routers import orders
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifecycle: startup and shutdown"""
 
    # Startup: запускаем Kafka Producer
    await get_kafka_producer()
 
    # Startup: запускаем Kafka Consumers
    await start_order_consumer()
 
    yield
 
    # Shutdown: останавливаем Producer и Consumers
    if _producer_client:
        await _producer_client.stop()
 
    await stop_order_consumer()
 
app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
)
 
# Подключаем роутеры
app.include_router(orders.router)
 
@app.get("/health")
async def health():
    """Health check"""
    return {"status": "healthy"}

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
 
from app.kafka.producer import KafkaProducerClient, get_kafka_producer
from app.config import settings
 
router = APIRouter(prefix="/orders", tags=["orders"])
 
class OrderCreate(BaseModel):
    """Схема создания заказа"""
    user_id: int
    items: list[str] = Field(..., min_items=1)
    total_amount: float = Field(..., gt=0)
 
class OrderResponse(BaseModel):
    """Ответ API"""
    order_id: int
    status: str = "pending"
    message: str = "Order is being processed"
 
@router.post("/", response_model=OrderResponse, status_code=status.HTTP_202_ACCEPTED)
async def create_order(
    order: OrderCreate,
    producer: KafkaProducerClient = Depends(get_kafka_producer),
):
    """
    Создаёт заказ и отправляет событие в Kafka.
 
    Возвращает 202 Accepted — обработка асинхронная.
    """
 
    # Генерируем ID (в реальности — из БД)
    order_id = 12345  # Заглушка
 
    # Отправляем событие в Kafka
    await producer.send(
        topic=settings.orders_topic,
        value={
            "event_type": "order.created",
            "order_id": order_id,
            "user_id": order.user_id,
            "items": order.items,
            "total_amount": order.total_amount,
        },
        key=str(order.user_id),  # Партиция по user_id
    )
 
    return OrderResponse(order_id=order_id)

import logging
 
from app.kafka.consumer import KafkaConsumerClient
from app.config import settings
 
logger = logging.getLogger(__name__)
 
# Глобальная переменная для consumer
_order_consumer: Optional[KafkaConsumerClient] = None
 
async def handle_order_created(event: dict):
    """
    Обработчик события order.created.
 
    Здесь выполняются все side-effects:
    - Отправка email
    - Обновление склада
    - Начисление бонусов
    """
    order_id = event["order_id"]
    user_id = event["user_id"]
 
    logger.info(f"Processing order {order_id} for user {user_id}")
 
    try:
        # Эмуляция работы
        await send_order_email(user_id, order_id)
        await update_inventory(event["items"])
        await calculate_bonuses(user_id, event["total_amount"])
 
        logger.info(f"Order {order_id} processed successfully")
 
    except Exception as e:
        logger.error(f"Failed to process order {order_id}: {e}")
        raise  # Retry через Kafka
 
async def send_order_email(user_id: int, order_id: int):
    """Заглушка отправки email"""
    logger.info(f"Email sent to user {user_id} for order {order_id}")
 
async def update_inventory(items: list[str]):
    """Заглушка обновления склада"""
    logger.info(f"Inventory updated for items: {items}")
 
async def calculate_bonuses(user_id: int, amount: float):
    """Заглушка начисления бонусов"""
    bonuses = amount * 0.05
    logger.info(f"Bonuses {bonuses} calculated for user {user_id}")
 
async def start_order_consumer():
    """Запускает consumer для обработки заказов"""
    global _order_consumer
 
    _order_consumer = KafkaConsumerClient(
        topic=settings.orders_topic,
        handler=handle_order_created,
    )
 
    await _order_consumer.start()
 
async def stop_order_consumer():
    """Останавливает consumer"""
    if _order_consumer:
        await _order_consumer.stop()

# Запускаем Kafka
docker compose up -d
 
# Запускаем FastAPI
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
 
# Тестируем API
curl -X POST http://localhost:8000/orders/ \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": 123,
    "items": ["item1", "item2"],
    "total_amount": 5000
  }'
 
# Смотрим логи consumer
# Должны увидеть: "Processing order 12345 for user 123"

# User 123: два события в разных партициях
await producer.send('orders', {'user_id': 123, 'action': 'create'})  # → partition 0
await producer.send('orders', {'user_id': 123, 'action': 'cancel'})  # → partition 2
 
# Consumer может обработать 'cancel' раньше 'create'!

# Все события user_id=123 попадут в одну партицию
await producer.send(
    topic='orders',
    value={'user_id': 123, 'action': 'create'},
    key=str(123)  # Ключ определяет партицию
)

async def handle_payment(event):
    # ❌ Плохо: повторная обработка = двойное списание
    await charge_user(event['user_id'], event['amount'])

async def handle_payment(event):
    payment_id = event['payment_id']
 
    # Проверяем, обработано ли уже
    if await is_payment_processed(payment_id):
        logger.info(f"Payment {payment_id} already processed, skipping")
        return
 
    # Обрабатываем
    await charge_user(event['user_id'], event['amount'])
 
    # Сохраняем факт обработки
    await mark_payment_processed(payment_id)

async def _consume_loop(self):
    """Цикл с обработкой ошибок"""
 
    async for message in self.consumer:
        try:
            await self.handler(message.value)
            await self.consumer.commit()
 
        except ValidationError as e:
            # Невалидное событие → в DLQ
            logger.error(f"Validation error: {e}")
            await self._send_to_dlq(message, error=str(e))
            await self.consumer.commit()  # Пропускаем
 
        except Exception as e:
            # Неожиданная ошибка → retry
            logger.error(f"Processing error: {e}")
            # Не делаем commit → Kafka повторит
 
async def _send_to_dlq(self, message, error: str):
    """Отправляет в Dead Letter Queue"""
 
    await self.dlq_producer.send(
        topic=f"{self.topic}.dlq",
        value={
            "original_message": message.value,
            "error": error,
            "timestamp": datetime.utcnow().isoformat(),
        },
    )

docker compose exec kafka kafka-consumer-groups \
  --bootstrap-server localhost:9092 \
  --describe \
  --group fastapi-app
 
# Вывод:
# TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders   0          1000            5000            4000  ← Проблема!

# Запускаем несколько инстансов приложения
# Kafka автоматически распределит партиции между ними

docker compose exec kafka kafka-topics \
  --alter \
  --topic orders \
  --partitions 6 \
  --bootstrap-server localhost:9092

async def _consume_loop(self):
    """Обрабатываем батчами"""
 
    batch = []
    batch_size = 100
 
    async for message in self.consumer:
        batch.append(message)
 
        if len(batch) >= batch_size:
            await self._process_batch(batch)
            await self.consumer.commit()
            batch = []
 
async def _process_batch(self, messages):
    """Параллельная обработка батча"""
 
    tasks = [self.handler(msg.value) for msg in messages]
    await asyncio.gather(*tasks, return_exceptions=True)

import signal
 
async def shutdown(signal, loop):
    """Graceful shutdown при SIGTERM"""
 
    logger.info(f"Received exit signal {signal.name}...")
 
    # Останавливаем consumer (завершает текущие сообщения)
    await stop_order_consumer()
 
    # Останавливаем producer
    if _producer_client:
        await _producer_client.stop()
 
    # Останавливаем event loop
    loop.stop()
 
# Регистрируем обработчики
loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
    loop.add_signal_handler(
        sig,
        lambda s=sig: asyncio.create_task(shutdown(s, loop))
    )

import pytest
from unittest.mock import AsyncMock, patch
 
@pytest.mark.asyncio
async def test_create_order_sends_event():
    """Проверяем, что API отправляет событие в Kafka"""
 
    # Мокируем producer
    mock_producer = AsyncMock()
 
    with patch('app.routers.orders.get_kafka_producer', return_value=mock_producer):
        from app.routers.orders import create_order
 
        order = OrderCreate(user_id=123, items=["item1"], total_amount=100)
        response = await create_order(order, producer=mock_producer)
 
        # Проверяем вызов
        mock_producer.send.assert_called_once()
        call_args = mock_producer.send.call_args
 
        assert call_args.kwargs['topic'] == 'orders'
        assert call_args.kwargs['value']['user_id'] == 123
        assert response.order_id > 0

import pytest
from testcontainers.kafka import KafkaContainer
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
 
@pytest.fixture(scope="session")
def kafka_container():
    """Запускает Kafka в Docker для тестов"""
 
    with KafkaContainer() as kafka:
        yield kafka
 
@pytest.mark.asyncio
async def test_kafka_producer_consumer(kafka_container):
    """E2E тест: producer → Kafka → consumer"""
 
    bootstrap_servers = kafka_container.get_bootstrap_server()
 
    # Producer
    producer = AIOKafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v).encode(),
    )
    await producer.start()
 
    # Consumer
    consumer = AIOKafkaConsumer(
        'test-topic',
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest',
        value_deserializer=lambda m: json.loads(m.decode()),
    )
    await consumer.start()
 
    # Отправляем событие
    test_message = {'test': 'data'}
    await producer.send_and_wait('test-topic', test_message)
 
    # Читаем событие
    async for message in consumer:
        assert message.value == test_message
        break  # Прочитали одно сообщение → выходим
 
    await producer.stop()
    await consumer.stop()

# Добавьте в docker-compose.yml
kafka:
  environment:
    KAFKA_JMX_PORT: 9101
    KAFKA_JMX_HOSTNAME: localhost
  ports:
    - "9101:9101"
 
jmx-exporter:
  image: sscaling/jmx-prometheus-exporter
  ports:
    - "5556:5556"
  environment:
    SERVICE_PORT: 5556
  volumes:
    - ./jmx-exporter-config.yml:/etc/jmx-exporter/config.yml
  command:
    - "5556"
    - "/etc/jmx-exporter/config.yml"

import structlog
 
logger = structlog.get_logger()
 
async def handle_order_created(event: dict):
    """Обработчик с structured logging"""
 
    logger.info(
        "order.processing.started",
        order_id=event["order_id"],
        user_id=event["user_id"],
        total_amount=event["total_amount"],
    )
 
    try:
        await process_order(event)
 
        logger.info(
            "order.processing.completed",
            order_id=event["order_id"],
            duration_ms=123,
        )
 
    except Exception as e:
        logger.error(
            "order.processing.failed",
            order_id=event["order_id"],
            error=str(e),
            exc_info=True,
        )
        raise

# SSL/TLS encryption
kafka:
  environment:
    KAFKA_SECURITY_PROTOCOL: SSL
    KAFKA_SSL_KEYSTORE_LOCATION: /etc/kafka/secrets/kafka.keystore.jks
    KAFKA_SSL_KEYSTORE_PASSWORD: changeit
 
# SASL Authentication
KAFKA_SASL_MECHANISM: PLAIN
KAFKA_SASL_ENABLED_MECHANISMS: PLAIN