Глоссарий — Apache Kafka

# Kafka как commit log:
# Запись добавляется в конец, offset монотонно растёт
# offset: 0  1  2  3  4  5  6
# data:   A  B  C  D  E  F  G
#                          ^--- Log End Offset (LEO)
# Consumer group A читает с offset 3
# Consumer group B читает с offset 6
# Записи не удаляются после чтения

# server.properties — базовая конфигурация брокера
broker.id=1
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://kafka-1.example.com:9092
log.dirs=/var/kafka/data
num.partitions=3
default.replication.factor=3
min.insync.replicas=2

# Создание топика через CLI
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create \
  --topic orders \
  --partitions 12 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete

# Топик 'orders' с 3 партициями на кластере из 3 брокеров:
# Partition 0: Leader=broker-1, Replicas=[1,2,3], ISR=[1,2,3]
# Partition 1: Leader=broker-2, Replicas=[2,3,1], ISR=[2,3,1]
# Partition 2: Leader=broker-3, Replicas=[3,1,2], ISR=[3,1,2]

kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092

# Фиксация смещения вручную в Java
consumer.poll(Duration.ofMillis(100)).forEach(record -> {
    processRecord(record);
    // Фиксация после обработки каждой записи
    consumer.commitSync(Collections.singletonMap(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1)
    ));
});

# Структура файлов партиции на диске
/var/kafka/data/orders-0/
  00000000000000000000.log       # сегмент от offset 0
  00000000000000000000.index     # индекс смещений
  00000000000000000000.timeindex # индекс по времени
  00000000000001048576.log       # следующий сегмент
  00000000000001048576.index
  00000000000001048576.timeindex

# Подключение клиента к кластеру
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092

# Клиент запрашивает метаданные у первого доступного брокера
# и узнаёт адреса всех брокеров и лидеров партиций
# Последующие запросы идут напрямую к нужному брокеру

# Мониторинг ISR через CLI
kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092
# Topic: orders  Partition: 0  Leader: 1  Replicas: 1,2,3  ISR: 1,2  Offline: 3
# ^ broker-3 выпал из ISR — под-реплицированная партиция!

# JMX метрика: kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

# Соотношение LEO и HW:
# Leader LEO = 10, Follower-1 LEO = 10, Follower-2 LEO = 8
# ISR = {Leader, Follower-1, Follower-2}
# HW = min(LEO всех в ISR) = 8
# Потребитель прочитает только offset 0..7
# После репликации Follower-2: HW продвинется до 10

# JMX: kafka.log:type=Log,name=LogEndOffset,topic=orders,partition=0

# Состояние реплик партиции
# Запись смещений на каждой реплике:
# Leader:     LEO=15  (принял все записи)
# Follower-1: LEO=15  (в ISR, в синхронизации)
# Follower-2: LEO=12  (отстаёт, ещё в ISR)
# Follower-3: LEO=5   (выбыл из ISR)
# HW = 12 (min LEO среди ISR)
# Лаг Follower-2: 15 - 12 = 3 записи

# Формат ответа на Fetch-запрос фолловера
# LeaderEpoch передаётся с каждым ответом
# Фолловер проверяет: если epoch изменился, нужно усечь лог

# Файл leader-epoch-checkpoint хранит историю
# /var/kafka/data/orders-0/leader-epoch-checkpoint
# 0
# 3  (epoch=3 начался с offset=0)
# 5  (epoch=5 начался с offset=200)
# Epoch 5 — текущий лидер

# KRaft-конфигурация для комбинированного узла
# (брокер + контроллер)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@kafka-1:9093,2@kafka-2:9093,3@kafka-3:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
controller.listener.names=CONTROLLER

# Инициализация кластера
kafka-storage.sh format -t <cluster-id> -c server.properties

# Проверка статуса кворума контроллеров
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status
# ClusterId:              abc123xyz
# LeaderId:               1
# LeaderEpoch:            5
# HighWatermark:          1248
# MaxFollowerLag:         0
# MaxFollowerLagTimeMs:   120
# CurrentVoters:          [1,2,3]
# CurrentObservers:       []

# Параметр топика — включить только при критической необходимости
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name orders \
  --alter \
  --add-config unclean.leader.election.enable=true

# Риск: если было 5 незафиксированных записей,
# после выборов нечистого лидера они будут потеряны
# и потребитель увидит дыру или перечитает старые данные

# Настройка аккумулятора для высокой пропускной способности
Properties props = new Properties();
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);     // 64 KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);         // ждать до 10 мс
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32 MB общий буфер
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); // сжатие батча
// При linger.ms=10 продюсер агрегирует записи 10 мс,
// затем отправляет один большой батч вместо многих мелких

Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// Автоматически устанавливает:
// acks=all, retries=Integer.MAX_VALUE, max.in.flight=5

// Брокер хранит: {PID=42, partition=0, lastSeq=100}
// Запись с seq=100 при повторе будет отброшена
// Запись с seq=101 будет принята как новая

producer.initTransactions();
try {
    producer.beginTransaction();
    // Запись в несколько топиков атомарна
    producer.send(new ProducerRecord<>("orders", key, value));
    producer.send(new ProducerRecord<>("audit-log", key, audit));
    // Фиксация смещений потребителя в рамках транзакции
    producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

# acks=all + min.insync.replicas=2 — золотой стандарт
# server.properties (глобально)
default.replication.factor=3
min.insync.replicas=2

# producer.properties
acks=all
# Семантика: запись подтверждена, если 2 из 3 реплик
# записали её на диск. При падении одного брокера
# данные не потеряны — у двух других есть копия

// Запись с ключом — гарантия порядка для user_id
ProducerRecord<String, Order> record = new ProducerRecord<>(
    "orders",
    "user-42",  // ключ — всегда та же партиция
    new Order(orderId, items)
);
producer.send(record);

// Кастомный партиционер для равномерного распределения
public class RoundRobinPartitioner implements Partitioner {
    public int partition(String topic, Object key,
        byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        return (int)(counter.getAndIncrement() % cluster.partitionCountForTopic(topic));
    }
}

Properties props = new Properties();
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));
// Все экземпляры с group.id="order-processor" делят партиции

# Определение брокера-координатора для группы
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group order-processor

# Вывод: GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
# Coordinator находится на брокере, ответственном за
# партицию hash("order-processor") % 50 топика __consumer_offsets

# Настройка Cooperative Sticky Assignor (минимальный простой)
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
    CooperativeStickyAssignor.class.getName());

# Настройка таймаутов сессии
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000); // 45 сек
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000); // 15 сек
# Потребитель должен отправлять heartbeat каждые 15 сек,
# иначе через 45 сек координатор инициирует ребалансировку

# Конфигурация статического членства
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "consumer-pod-3");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 300000); // 5 минут
# При рестарте пода consumer-pod-3 в течение 5 минут
# он снова получит те же партиции без ребалансировки
# Идеально для Kafka Streams и stateful обработчиков

// Ручной коммит после гарантированной обработки
consumer.poll(Duration.ofMillis(100)).forEach(record -> {
    try {
        processRecord(record);
        // Асинхронный коммит для производительности
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) log.error("Commit failed", exception);
        });
    } catch (Exception e) {
        // Синхронный коммит при завершении для надёжности
        consumer.commitSync();
        throw e;
    }
});

# Создание топика с уплотнением лога
kafka-topics.sh --bootstrap-server localhost:9092 \
  --create \
  --topic user-profiles \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.1 \
  --config delete.retention.ms=86400000

# Запись с value=null (tombstone) удаляет ключ
producer.send(new ProducerRecord<>("user-profiles", "user-42", null));

# server.properties — включение tiered storage
remote.log.storage.system.enable=true
remote.log.manager.task.interval.ms=30000

# Плагин для S3 (конкретная реализация зависит от провайдера)
remote.log.storage.manager.class.name=org.apache.kafka.server.log.remote.storage.RemoteLogStorageManager

# Конфигурация топика
kafka-configs.sh --alter --topic events \
  --add-config remote.storage.enable=true,\
  local.retention.ms=86400000

# Путь данных без zero-copy (4 копирования):
# Диск -> OS Page Cache -> User Space Buffer -> Kernel Socket Buffer -> NIC

# Путь данных с zero-copy (2 копирования, sendfile):
# Диск -> OS Page Cache -> NIC

# В Java: FileChannel.transferTo() использует sendfile()
# Kafka автоматически применяет zero-copy для FileChannel
# ВАЖНО: работает только без шифрования SSL на уровне брокера

# Структура запроса Produce (упрощённо)
# [4 байта] Length
# [2 байта] API Key = 0 (Produce)
# [2 байта] API Version = 9
# [4 байта] Correlation ID
# [2 байта] Client ID length + данные
# [body]    Topic name, acks, timeout, records

# Просмотр поддерживаемых API брокером
kafka-broker-api-versions.sh --bootstrap-server localhost:9092

# Конфигурация JDBC Source коннектора
curl -X POST http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "postgres-orders",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:postgresql://localhost/shop",
      "table.whitelist": "orders",
      "mode": "timestamp+incrementing",
      "timestamp.column.name": "updated_at",
      "incrementing.column.name": "id",
      "topic.prefix": "pg."
    }
  }'

# Конфигурация SMT в коннекторе
# Маскировка PII + добавление топика как поля
transforms=maskPII,addTopic
transforms.maskPII.type=org.apache.kafka.connect.transforms.MaskField$Value
transforms.maskPII.fields=email,phone
transforms.addTopic.type=org.apache.kafka.connect.transforms.InsertField$Value
transforms.addTopic.topic.field=_kafka_topic
# Для условного применения:
transforms.maskPII.predicate=isProduction
predicates.isProduction.type=org.apache.kafka.connect.transforms.predicates.HasHeaderKey
predicates.isProduction.name=env

# Включение DLQ в конфигурации sink-коннектора
errors.tolerance=all
errors.deadletterqueue.topic.name=orders-dlq
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true
errors.log.enable=true
errors.log.include.messages=true
# Чтение DLQ для диагностики
kafka-console-consumer.sh \
  --topic orders-dlq \
  --property print.headers=true \
  --bootstrap-server localhost:9092

# Регистрация схемы Avro через REST API
curl -X POST http://schema-registry:8081/subjects/orders-value/versions \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"}

# Wire Format: [0x00][schema_id: 4 bytes][avro payload]

# Схема Avro (orders.avsc)
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": "string", "default": "PENDING"}
  ]
}
# Эволюция: добавление поля с default не нарушает
# обратную совместимость (BACKWARD mode)

# Установка режима совместимости для subject
curl -X PUT http://schema-registry:8081/config/orders-value \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d '{"compatibility": "BACKWARD"}'

# Таблица допустимых изменений при BACKWARD:
# Добавить опциональное поле (с default) — РАЗРЕШЕНО
# Удалить поле                           — РАЗРЕШЕНО
# Добавить обязательное поле             — ЗАПРЕЩЕНО
# Изменить тип поля                      — ЗАПРЕЩЕНО

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders =
    builder.stream("raw-orders");

// Stateless трансформация
KStream<String, EnrichedOrder> enriched = orders
    .filter((key, order) -> order.getAmount() > 0)
    .mapValues(order -> enrich(order));

// Запись в новый топик
enriched.to("enriched-orders",
    Produced.with(Serdes.String(), enrichedSerde));

StreamsBuilder builder = new StreamsBuilder();
// KTable из топика user-profiles
KTable<String, UserProfile> profiles =
    builder.table("user-profiles",
        Materialized.as("profiles-store"));

// Join KStream с KTable
KStream<String, EnrichedOrder> enriched = orders.join(
    profiles,
    (order, profile) -> new EnrichedOrder(order, profile.getName())
);
// Join мгновенный: lookup по ключу в локальном state store

StreamsBuilder builder = new StreamsBuilder();
// GlobalKTable реплицируется на все экземпляры
GlobalKTable<String, Product> products =
    builder.globalTable("product-catalog");

// Можно join по любому полю значения, не только по ключу
KStream<String, EnrichedOrder> enriched = orders.join(
    products,
    (orderKey, order) -> order.getProductId(), // извлечение ключа для join
    (order, product) -> new EnrichedOrder(order, product)
);

// Materialized store для агрегации
KTable<String, Long> orderCounts = orders
    .groupByKey()
    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>
        as("order-count-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Long())
    );

// Запрос состояния через Interactive Queries
ReadOnlyKeyValueStore<String, Long> store =
    streams.store(StoreQueryParameters.fromNameAndType(
        "order-count-store", QueryableStoreTypes.keyValueStore()));

// Подсчёт заказов за каждые 5 минут (Tumbling Window)
KTable<Windowed<String>, Long> windowedCounts = orders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("windowed-count-store"));

// Ключ результата: Windowed{key="user-1", window=[12:00, 12:05)}
// Hopping: окно 5 мин, шаг 1 мин — записи считаются в нескольких окнах
TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30))
    .advanceBy(Duration.ofMinutes(1));

-- Создание стрима поверх топика
CREATE STREAM orders (
    order_id VARCHAR KEY,
    amount DOUBLE,
    user_id VARCHAR
) WITH (
    KAFKA_TOPIC='raw-orders',
    VALUE_FORMAT='AVRO'
);

-- Постоянное агрегирование (CSAS)
CREATE TABLE order_totals AS
    SELECT user_id, SUM(amount) AS total
    FROM orders
    GROUP BY user_id
    EMIT CHANGES;

// Экспозиция Interactive Queries через REST
var queryableStore = streams.store(
    StoreQueryParameters.fromNameAndType(
        "order-count-store",
        QueryableStoreTypes.keyValueStore()
    )
);

// Получить метаданные (какой экземпляр хранит ключ)
KeyQueryMetadata metadata = streams.queryMetadataForKey(
    "order-count-store", "user-42", Serdes.String().serializer()
);
// metadata.activeHost() -> {host, port} нужного инстанса

# Конфигурация брокера для SASL/SCRAM
listeners=SASL_SSL://0.0.0.0:9093
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512

# Создание пользователя
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter --add-config 'SCRAM-SHA-512=[iterations=8192,password=secret]' \
  --entity-type users --entity-name alice

# Конфигурация клиента (JAAS)
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="alice" password="secret";

# Конфигурация брокера для mTLS
ssl.keystore.location=/etc/kafka/ssl/kafka.server.keystore.jks
ssl.keystore.password=keystorepass
ssl.key.password=keypass
ssl.truststore.location=/etc/kafka/ssl/kafka.server.truststore.jks
ssl.truststore.password=truststorepass
ssl.client.auth=required

# Конфигурация клиента
ssl.keystore.location=/etc/kafka/ssl/client.keystore.jks
ssl.truststore.location=/etc/kafka/ssl/client.truststore.jks
security.protocol=SSL

# Предоставить alice право записи в топик orders
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:alice \
  --operation Write \
  --topic orders

# Предоставить group order-processor право чтения
kafka-acls.sh --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:service-account \
  --operation Read \
  --group order-processor

# Просмотр всех ACL
kafka-acls.sh --list --bootstrap-server localhost:9092

# Установка квоты для конкретного пользователя
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'producer_byte_rate=1048576,consumer_byte_rate=2097152' \
  --entity-type users \
  --entity-name alice

# Квота по умолчанию для всех клиентов
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'producer_byte_rate=10485760' \
  --entity-type users \
  --entity-default

# log4j.properties — включение аудит-лога
log4j.appender.authorizerAppender=org.apache.log4j.DailyRollingFileAppender
log4j.appender.authorizerAppender.DatePattern='.'yyyy-MM-dd-HH
log4j.appender.authorizerAppender.File=/var/log/kafka/kafka-authorizer.log
log4j.appender.authorizerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.authorizerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
log4j.logger.kafka.authorizer.logger=INFO, authorizerAppender
log4j.additivity.kafka.authorizer.logger=false

# Примеры критических JMX метрик
kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec
kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions
kafka.server:type=ReplicaManager,name=OfflinePartitionsCount
kafka.controller:type=KafkaController,name=ActiveControllerCount
kafka.network:type=RequestMetrics,name=TotalTimeMs,request=Produce

# Запуск JMX Exporter для Prometheus
# -javaagent:jmx_prometheus_javaagent.jar=7071:kafka-jmx-config.yml

# Просмотр лага всех групп потребителей
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group order-processor
# GROUP           TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processor orders   0          1250            1300            50
# order-processor orders   1          980             980             0
# order-processor orders   2          700             750             50
# Общий лаг: 100 записей

# Метрики ISR для мониторинга
kafka.server:type=ReplicaManager,name=IsrShrinksPerSec
kafka.server:type=ReplicaManager,name=IsrExpandsPerSec
kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

# Параметры, влияющие на ISR
replica.lag.time.max.ms=30000  # реплика выбывает через 30 сек
replica.fetch.min.bytes=1
replica.fetch.wait.max.ms=500

# Алерт: UnderReplicatedPartitions > 0 дольше 5 минут

# Шаг 1: Сгенерировать план перераспределения
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --generate \
  --topics-to-move-json-file topics.json \
  --broker-list "1,2,3,4" > reassignment.json

# Шаг 2: Применить с троттлингом
kafka-reassign-partitions.sh \
  --bootstrap-server localhost:9092 \
  --execute \
  --reassignment-json-file reassignment.json \
  --throttle 50000000  # 50 MB/s максимум

# docker-compose.yml фрагмент
services:
  jmx-exporter:
    image: bitnami/jmx-exporter
    command: ["5556", "/etc/config/kafka.yml"]
    volumes:
      - ./kafka-jmx-config.yml:/etc/config/kafka.yml
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
  grafana:
    image: grafana/grafana
    # Импорт дашборда: grafana.com/dashboards/7589

# Формулы расчёта
# Требуемый дисковый I/O на запись:
#   write_throughput * replication_factor
#   Пример: 100 MB/s * 3 = 300 MB/s

# Требуемое хранилище:
#   write_throughput * retention_hours * 3600 * replication_factor
#   Пример: 100 MB/s * 168h * 3600 * 3 = ~181 TB

# Сетевой трафик на брокер (лидер партиции):
#   write + read * num_consumer_groups
#   = 100 + 100 * 3 = 400 MB/s исходящих

# mm2.properties — конфигурация репликации active-passive
clusters=primary, secondary
primary.bootstrap.servers=kafka-primary:9092
secondary.bootstrap.servers=kafka-secondary:9092

# Репликация из primary в secondary
primary->secondary.enabled=true
primary->secondary.topics=orders.*,payments.*

# Синхронизация смещений потребителей
primary->secondary.sync.group.offsets.enabled=true
primary->secondary.emit.checkpoints.interval.seconds=10

# Запуск
connect-mirror-maker.sh mm2.properties

# Пример анализа RPO/RTO для Kafka DR

# Сценарий: active-passive с MirrorMaker 2
# Задержка репликации: ~5 секунд
# RPO = 5 секунд (данные за 5 сек могут быть потеряны)

# Процедура переключения:
# 1. Обнаружение сбоя primary:      ~30 сек
# 2. Переключение DNS/LB:             ~2 мин
# 3. Перезапуск клиентов:             ~5 мин
# RTO = ~8 минут

# Для RPO=0: синхронная репликация
# (недоступна в стандартной Kafka, требует Kafka Bridge)

// Паттерн Event Sourcing с Kafka
// Команды -> события -> проекции

// Событие (неизменяемое)
public record OrderPlaced(
    String orderId,
    String userId,
    List<OrderItem> items,
    double totalAmount,
    Instant timestamp
) {}

// Публикация события
producer.send(new ProducerRecord<>(
    "order-events",
    orderId,           // ключ для партиционирования
    new OrderPlaced(orderId, userId, items, total, Instant.now())
));

// Проекция: восстановление состояния через Kafka Streams KTable

// CQRS с Kafka
// Write side: команды -> события
kafka-topic: order-events (append-only event log)

// Read side: несколько проекций для разных сценариев
// 1. PostgreSQL: текущий статус заказа (по ID)
// 2. Elasticsearch: поиск заказов по атрибутам
// 3. Redis: корзина пользователя (TTL)
// 4. KTable в Kafka Streams: агрегаты для дашборда

// Connect Sink обновляет все проекции
// асинхронно из топика order-events

-- Схема outbox таблицы
CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    topic VARCHAR NOT NULL,
    key VARCHAR,
    payload JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    sent_at TIMESTAMP  -- NULL = ещё не отправлено
);

-- В одной транзакции с бизнес-логикой
BEGIN;
  UPDATE orders SET status = 'CONFIRMED' WHERE id = $1;
  INSERT INTO outbox (topic, key, payload)
    VALUES ('order-events', $1, '{"event": "OrderConfirmed", ...}');
COMMIT;

// Saga хореография: оформление заказа
// 1. OrderService: OrderPlaced -> orders-events
// 2. PaymentService: слушает orders-events,
//    резервирует оплату -> PaymentReserved
// 3. InventoryService: слушает PaymentReserved,
//    резервирует товар -> InventoryReserved
// 4. ShippingService: создаёт доставку -> OrderCompleted

// Компенсация при ошибке на шаге 3:
// InventoryService: InventoryFailed ->
// PaymentService: отменяет резервирование

# Соглашение об именовании топиков
# формат: {domain}.{entity}.{event}.{version}

# Примеры:
ordersvc.order.placed.v1
ordersvc.order.shipped.v1
payment.transaction.completed.v2
analytics.user.session.ended.v1

# Метаданные топика (через Custom Config)
kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter --topic ordersvc.order.placed.v1 \
  --add-config 'owner=orders-team,\
  [email protected],\
  sla-tier=gold'