Поток событий

Event Flow описывает асинхронную коммуникацию между микросервисами через Apache Kafka. Используется для операций, которые не требуют немедленного ответа, обеспечения eventual consistency и интеграции независимых сервисов.

Архитектура

graph TB subgraph "Producers" Identity[Identity Service
Producer] Credential[Credential Service
Producer] Auth[Auth Service
Producer] end subgraph "Event Bus (Kafka)" TopicUser[identity.user.events
3 partitions] TopicIdentifier[identity.identifier.events
3 partitions] TopicPassword[credential.password.events
3 partitions] end subgraph "Consumers" Herald[Herald Service
Notifications] Account[Account Service
Accounts] Analytics[Analytics Pipeline
Data Warehouse] end subgraph "Transactional Outbox" OutboxIdentity[(outbox_events
Identity DB)] OutboxCredential[(outbox_events
Credential DB)] WorkerIdentity[Outbox Worker
Identity] WorkerCredential[Outbox Worker
Credential] end Identity -->|INSERT in tx| OutboxIdentity WorkerIdentity -->|Poll & Publish| OutboxIdentity WorkerIdentity -->|PUBLISH| TopicUser WorkerIdentity -->|PUBLISH| TopicIdentifier Credential -->|INSERT in tx| OutboxCredential WorkerCredential -->|Poll & Publish| OutboxCredential WorkerCredential -->|PUBLISH| TopicPassword TopicUser -.->|CONSUME| Herald TopicUser -.->|CONSUME| Account TopicUser -.->|CONSUME| Analytics TopicIdentifier -.->|CONSUME| Herald TopicPassword -.->|CONSUME| Herald style TopicUser fill:#FF6B6B style TopicIdentifier fill:#FF6B6B style TopicPassword fill:#FF6B6B

Transactional Outbox Pattern

Гарантирует at-least-once доставку событий без distributed transactions между БД и Kafka.

Архитектура Outbox

sequenceDiagram participant UseCase participant UoW participant PostgreSQL participant OutboxTable participant Worker participant Kafka participant Consumer UseCase->>UoW: async with uow.transaction() UoW->>PostgreSQL: BEGIN TRANSACTION UseCase->>UoW: await tx.users.create(user) UoW->>PostgreSQL: INSERT INTO users (...) UseCase->>UoW: await tx.outbox.create(UserCreatedEvent) UoW->>OutboxTable: INSERT INTO outbox_events
(topic, aggregate_id, event_type, payload) UoW->>PostgreSQL: COMMIT Note over Worker: Background process
polling every 100ms Worker->>OutboxTable: SELECT * FROM outbox_events
WHERE status = 'PENDING'
LIMIT 100 FOR UPDATE SKIP LOCKED OutboxTable-->>Worker: [event1, event2, ...] Worker->>Kafka: PUBLISH event1 Kafka-->>Worker: ACK Worker->>OutboxTable: UPDATE SET status='PUBLISHED' Worker->>Kafka: PUBLISH event2 Kafka-->>Worker: ACK Worker->>OutboxTable: UPDATE SET status='PUBLISHED' Kafka->>Consumer: DELIVER events Consumer->>Consumer: Process events Consumer->>Kafka: COMMIT offset

Outbox Event Schema

Database Table:

CREATE TABLE outbox_events (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    topic VARCHAR(255) NOT NULL,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id UUID NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
    attempts_made INT NOT NULL DEFAULT 0,
    max_attempts INT NOT NULL DEFAULT 5,
    last_error TEXT,
    next_retry_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    published_at TIMESTAMPTZ,

    INDEX idx_pending_events (status, next_retry_at, created_at)
) PARTITION BY RANGE (created_at);

-- Партиции создаются автоматически через alembic migration
CREATE TABLE outbox_events_2024_03 PARTITION OF outbox_events
    FOR VALUES FROM ('2024-03-01') TO ('2024-04-01');

Python Model:

from omni_box import OutboxEvent

class OutboxEvent(BaseModel):
    id: UUID
    topic: str                  # "identity.user.events"
    aggregate_type: str         # "user"
    aggregate_id: UUID
    event_type: str             # "user.created"
    payload: dict[str, Any]
    status: OutboxStatus        # PENDING, PUBLISHED, FAILED
    attempts_made: int
    max_attempts: int = 5
    last_error: str | None
    next_retry_at: datetime | None
    created_at: datetime
    published_at: datetime | None

Use Case Integration

Пример: CreateUserUseCase:

from identity_service.core.uow import AsyncUnitOfWork
from service_contracts.events.users import USER_AGGREGATE, UserEventType, UserTopicName

class CreateUserUseCase:
    def __init__(self, uow: AsyncUnitOfWork):
        self._uow = uow

    async def execute(self, username: str) -> UserDTO:
        async with self._uow.transaction() as tx:
            # 1. Business logic
            user = User(
                id=uuid4(),
                username=username,
                status=UserStatus.ACTIVE,
                created_at=utc_now(),
                updated_at=utc_now(),
            )

            # 2. Persist to DB
            await tx.users.create(user)

            # 3. Create outbox event (same transaction!)
            event = OutboxEvent(
                id=uuid4(),
                topic=UserTopicName.IDENTITY_USER_EVENTS.value,
                aggregate_type=USER_AGGREGATE,
                aggregate_id=user.id,
                event_type=UserEventType.CREATED.value,
                payload={
                    "user_id": str(user.id),
                    "username": user.username,
                    "status": user.status.value,
                    "created_at": user.created_at.isoformat(),
                },
                status=OutboxStatus.PENDING,
                attempts_made=0,
                created_at=utc_now(),
            )
            await tx.outbox.create(event)

            # 4. COMMIT atomically
            # Либо оба операции успешны, либо обе откатываются

        return UserDTO.from_entity(user)

Outbox Worker

Background job (запускается в отдельном процессе):

from omni_box import AsyncOutboxPublisher
from kafka_publisher_kit import AsyncKafkaPublisher

async def run_outbox_worker():
    """Continuously poll and publish pending events"""
    repo = PostgresOutboxRepository(session_maker)
    kafka_publisher = AsyncKafkaPublisher(bootstrap_servers=KAFKA_BROKERS)

    publisher = AsyncOutboxPublisher(
        repo=repo,
        broker=kafka_publisher,
    )

    while not shutdown_requested():
        result = await publisher.publish_batch(
            worker_id=WORKER_ID,
            batch_size=100,
            publish_timeout=5.0,
            use_locking=True,  # SELECT FOR UPDATE SKIP LOCKED
            use_bulk=True,     # Bulk updates
        )

        if result and result.published_event_ids:
            logger.info(
                "Published events",
                count=len(result.published_event_ids),
            )

        await asyncio.sleep(0.1)  # 100ms poll interval

Worker Features: - Distributed locking: SELECT FOR UPDATE SKIP LOCKED — каждый worker берет свою порцию событий - Bulk operations: Batch INSERT/UPDATE для производительности - Retry with backoff: Exponential backoff для failed events - Idempotency: Kafka producer с enable.idempotence=true - Monitoring: Prometheus metrics для latency, throughput, errors

Event Topics

identity.user.events

Event Types:

class EventType(StrEnum):
    USER_CREATED = "user.created"
    USER_DISABLED = "user.disabled"
    USER_ENABLED = "user.enabled"
    USER_DELETED = "user.deleted"

Event Schema (user.created):

{
  "event_id": "7c9e6679-7425-40de-944b-e07fc1f90ae7",
  "event_type": "user.created",
  "aggregate_type": "user",
  "aggregate_id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2024-03-09T10:30:00.000Z",
  "payload": {
    "user_id": "550e8400-e29b-41d4-a716-446655440000",
    "username": "john_doe",
    "status": "active",
    "created_at": "2024-03-09T10:30:00.000Z"
  },
  "metadata": {
    "service": "identity-service",
    "version": "1.0.0",
    "trace_id": "a1b2c3d4e5f6",
    "correlation_id": "request-12345"
  }
}

identity.identifier.events

Event Types:

class EventType(StrEnum):
    IDENTIFIER_ADDED = "identifier.added"
    IDENTIFIER_VERIFIED = "identifier.verified"
    IDENTIFIER_REMOVED = "identifier.removed"

Event Schema (identifier.verified):

{
  "event_id": "uuid",
  "event_type": "identifier.verified",
  "aggregate_type": "identifier",
  "aggregate_id": "identifier-uuid",
  "timestamp": "2024-03-09T10:35:00.000Z",
  "payload": {
    "identifier_id": "identifier-uuid",
    "user_id": "user-uuid",
    "type": "email",
    "value": "john@example.com",
    "verified_at": "2024-03-09T10:35:00.000Z"
  }
}

credential.password.events

Event Types:

class EventType(StrEnum):
    PASSWORD_SET = "password.set"
    PASSWORD_CHANGED = "password.changed"
    PASSWORD_VERIFIED = "password.verified"

Event Schema (password.changed):

{
  "event_id": "uuid",
  "event_type": "password.changed",
  "aggregate_type": "password",
  "aggregate_id": "user-uuid",
  "timestamp": "2024-03-09T10:40:00.000Z",
  "payload": {
    "user_id": "user-uuid",
    "changed_at": "2024-03-09T10:40:00.000Z",
    "must_change": false
  }
}

Security Note: Никогда не публикуем пароли или хеши паролей в события!

Event Consumers

Herald Service (Notifications)

Subscriptions: - identity.user.events → Отправка welcome email при регистрации - identity.identifier.events → Отправка verification codes - credential.password.events → Уведомления о смене пароля

Consumer Implementation:

from aiokafka import AIOKafkaConsumer

async def consume_user_events():
    consumer = AIOKafkaConsumer(
        TopicName.IDENTITY_USER_EVENTS,
        bootstrap_servers=KAFKA_BROKERS,
        group_id="herald-service-user-events",
        auto_offset_reset="earliest",
        enable_auto_commit=False,  # Manual commit for reliability
    )

    await consumer.start()

    try:
        async for msg in consumer:
            event = parse_event(msg.value)

            if event.event_type == EventType.USER_CREATED:
                await send_welcome_email(event.payload["user_id"])

            elif event.event_type == EventType.USER_DISABLED:
                await send_account_disabled_email(event.payload["user_id"])

            # Commit offset only after successful processing
            await consumer.commit()

    finally:
        await consumer.stop()

Account Service

Subscriptions: - identity.user.events → Создание Account при регистрации User

Consumer Pattern (transactional):

async def handle_user_created(event: UserCreatedEvent):
    """Create account for new user"""
    async with uow.transaction() as tx:
        # Check idempotency (event already processed?)
        processed = await tx.idempotency.exists(event.event_id)
        if processed:
            logger.info("Event already processed", event_id=event.event_id)
            return

        # Business logic
        account = Account(
            id=uuid4(),
            user_id=event.payload["user_id"],
            status=AccountStatus.TRIAL,
            created_at=utc_now(),
        )
        await tx.accounts.create(account)

        # Mark as processed
        await tx.idempotency.mark_processed(event.event_id)

        # Commit

    logger.info("Account created", account_id=account.id, user_id=event.payload["user_id"])

Idempotency Table:

CREATE TABLE processed_events (
    event_id UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    consumer_group VARCHAR(100) NOT NULL
);

CREATE INDEX idx_processed_events_timestamp
    ON processed_events(processed_at);

Event Schema Evolution

Versioning Strategy

Backward Compatible Changes (allowed): - Добавление новых опциональных полей - Добавление новых event types

// v1 -> v2: добавление поля email (optional)
{
  "event_type": "user.created",
  "payload": {
    "user_id": "uuid",
    "username": "john_doe",
    "email": "john@example.com"  // NEW (optional)
  }
}

Breaking Changes (требуют новой версии topic): - Удаление полей - Изменение типов полей - Изменение формата значений

Решение: Создать новый topic с версией

# Old consumers
TopicName.IDENTITY_USER_EVENTS_V1 = "identity.user.events.v1"

# New consumers
TopicName.IDENTITY_USER_EVENTS_V2 = "identity.user.events.v2"

# Dual-write period (producers publish to both topics)
# After migration, deprecate v1

Monitoring & Observability

Metrics

Producer Metrics:

from prometheus_client import Counter, Histogram

outbox_events_created = Counter(
    'outbox_events_created_total',
    'Number of events created in outbox',
    ['service', 'event_type']
)

outbox_events_published = Counter(
    'outbox_events_published_total',
    'Number of events published to Kafka',
    ['service', 'topic', 'event_type']
)

outbox_publish_latency = Histogram(
    'outbox_publish_latency_seconds',
    'Time from event creation to Kafka publish',
    ['service', 'topic']
)

Consumer Metrics:

events_consumed = Counter(
    'kafka_events_consumed_total',
    'Number of events consumed',
    ['consumer_group', 'topic', 'event_type']
)

events_processing_errors = Counter(
    'kafka_events_processing_errors_total',
    'Number of event processing errors',
    ['consumer_group', 'topic', 'error_type']
)

consumer_lag = Gauge(
    'kafka_consumer_lag',
    'Consumer lag in messages',
    ['consumer_group', 'topic', 'partition']
)

Tracing

OpenTelemetry integration:

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

async def publish_event(event: OutboxEvent):
    """Publish event with distributed tracing"""
    with tracer.start_as_current_span("outbox.publish") as span:
        span.set_attribute("event.type", event.event_type)
        span.set_attribute("event.topic", event.topic)

        # Inject trace context into Kafka message headers
        headers = {}
        inject(headers)

        await kafka_producer.send(
            topic=event.topic,
            value=event.payload,
            headers=headers.items(),
        )

Dead Letter Queue (DLQ)

Pattern для failed events:

async def consume_with_dlq():
    consumer = AIOKafkaConsumer(TopicName.IDENTITY_USER_EVENTS, ...)
    dlq_producer = AIOKafkaProducer(...)

    async for msg in consumer:
        try:
            await process_event(msg.value)
            await consumer.commit()
        except RetryableError:
            # Retry later (don't commit offset)
            await asyncio.sleep(5)
        except PermanentError as e:
            # Send to DLQ for manual investigation
            await dlq_producer.send(
                topic=f"{TopicName.IDENTITY_USER_EVENTS}.dlq",
                value=msg.value,
                headers=[("error", str(e))],
            )
            await consumer.commit()  # Skip this message

Best Practices

✅ DO

Используй Transactional Outbox для критичных событий
Версионируй события (добавляй schema_version в payload)
Реализуй idempotency на стороне consumer
Логируй все события с trace_id для debugging
Монитор consumer lag — алертируй при lag > threshold
Используй Dead Letter Queue для permanently failed events

❌ DON'T

НЕ публикуй PII в события без шифрования
НЕ публикуй пароли/секреты даже в зашифрованном виде
НЕ делай breaking changes без миграционного плана
НЕ блокируй business logic ожиданием Kafka publish
НЕ используй Kafka для request/response паттернов (используй gRPC)

Связанные страницы

Online Data Flow — синхронные потоки данных
Event Design Principles — принципы дизайна событий
Backend / Event-Driven Patterns — паттерны реализации

Пакетный поток данных Владение данными

На странице

Архитектура Transactional Outbox Pattern Архитектура Outbox Outbox Event Schema Use Case Integration Outbox Worker Event Topics identity.user.events identity.identifier.events credential.password.events Event Consumers Herald Service (Notifications) Account Service Event Schema Evolution Versioning Strategy Monitoring & Observability Metrics Tracing Dead Letter Queue (DLQ) Best Practices ✅ DO ❌ DON'T Связанные страницы