Kafka Producer — idempotence, acks=all, partition key

Опирается на правила: R-KFK-PROD-1 … R-KFK-PROD-4 и R-KFK-PROD-X1 … R-KFK-PROD-X4 из Kafka Style Guide → раздел 1. Producer.

Важно знать

• enable.idempotence: true всегда. Автоматически даёт acks=all, retries=MAX, max.in.flight ≤ 5. Exactly-once на уровне partition.
• Partition key обязателен для всех бизнес-событий. Дефолтный ключ — aggregate id.
• JSON сериализация (JsonSerializer) по умолчанию. Avro/Protobuf — только для bandwidth-чувствительных топиков.
• KafkaTemplate.send напрямую из use case handler для domain events — запрещён. События идут через outbox.
• acks=0/1 — никогда. Только acks=all.
• Без partition key для бизнес-событий round-robin теряет ordering для aggregate.
• Kafka не XA — в одной транзакции с PG быть не может. Только outbox.

Kafka producer — точка, где сервис «публикует факт» во внешний мир. Ошибка здесь — потерянное событие или дубликат, который downstream не отличит от настоящего. UCP формулирует правила так, чтобы producer всегда был exactly-once на partition и atomic с DB через outbox.

enable.idempotence: true — всегда

R-KFK-PROD-1: один флаг включает три гарантии сразу.

spring:
  kafka:
    producer:
      properties:
        enable.idempotence: true
      acks: all

Что Kafka делает с enable.idempotence: true:

• acks=all — broker подтверждает только после репликации.
• retries=Integer.MAX_VALUE — producer retry-ит до победы или session-timeout.
• max.in.flight.requests.per.connection ≤ 5 — гарантия ordering между retries.

Producer получает producer-id от broker-а; broker дедуплицирует записи по (producer-id, sequence-number) на partition. Это даёт exactly-once на уровне partition. Если producer retry-ит — broker молча отбрасывает дубль.

Без enable.idempotence retry на стороне producer создаёт дубликаты в Kafka, downstream consumer видит одно событие N раз.

Partition key — обязателен

R-KFK-PROD-2: ключ определяет, на какой partition уйдёт сообщение.

@Component
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishConfirmed(Order order) {
        kafkaTemplate.send(
            "order-service.order.confirmed",
            order.getId().toString(),
            OrderConfirmedEvent.from(order)
        );
    }
}

Дефолтный ключ — order.getId().toString() (aggregate id). Это гарантирует: все события одного order.id уходят на один partition, а внутри partition Kafka сохраняет порядок.

Без ключа — kafkaTemplate.send(topic, event) — round-robin между partitions. Сценарий поломки:

1. OrderCreated(orderId=42) уходит на partition 1.
2. OrderConfirmed(orderId=42) уходит на partition 5.
3. Consumer для partition 5 обрабатывает OrderConfirmed до того, как consumer для partition 1 обработал OrderCreated.
4. Downstream получает confirm для несуществующего заказа.

Правило: для каждого aggregate — стабильный ключ. Обычно — aggregateId.toString().

JSON по умолчанию

R-KFK-PROD-3: дефолтный serializer.

spring:
  kafka:
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

JSON прост в дебаге (kafka-console-consumer показывает читаемый payload), не требует Schema Registry. Цена — больше bytes на bandwidth.

Avro/Protobuf — для high-throughput топиков (миллиарды событий/сутки). Требует Schema Registry, дополнительной инфры и compatibility-check на CI. В UCP-сервисах не дефолт.

Не send из use case handler

R-KFK-PROD-4: domain events публикуются через outbox, не прямым kafkaTemplate.send.

// КАТАСТРОФА — Kafka и DB не атомарны
@UseCase
@RequiredArgsConstructor
public class ConfirmOrderHandlerWrong implements UseCaseHandler<ConfirmOrderCommand, Order> {

    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    @Transactional
    public Order handle(ConfirmOrderCommand command) {
        var order = orderRepository.findById(command.orderId()).orElseThrow();
        order.confirm();
        orderRepository.save(order);

        kafkaTemplate.send("order.events", order.getId().toString(),
            OrderConfirmedEvent.from(order));
        return order;
    }
}

Сценарии поломки:

1. kafkaTemplate.send прошёл, save упал с deadlock → событие опубликовано, в БД заказ не подтверждён. Inconsistency.
2. save прошёл, send упал с network error → заказ подтверждён, downstream не знает.

Корректно — через outbox:

@Override
@Transactional
public Order handle(ConfirmOrderCommand command) {
    var order = orderRepository.findById(command.orderId(), SelectMode.FOR_UPDATE).orElseThrow();
    order.confirm();
    orderRepository.save(order);

    outboxRepository.append(OutboxEvent.builder()
        .aggregateType("Order")
        .aggregateId(order.getId())
        .eventType("order.confirmed.v1")
        .payload(jsonbHelper.serialize(OrderConfirmedEvent.from(order)))
        .topic("order-service.order.confirmed")
        .partitionKey(order.getId().toString())
        .build());

    return order;
}

Запись в outbox_event идёт в той же DB-транзакции, что save. Атомарность гарантирована PG. Отдельный outbox-relay читает unpublished и публикует в Kafka. Подробнее — Outbox publishing.

Допустимый прямой kafkaTemplate.send:

• Технические audit-events (в дополнение к audit_log таблице).
• Метрики / health-сигналы.
• Команды другим сервисам без транзакционного контекста (например запрос на отчёт от admin-инструмента).

Что запрещено

enable.idempotence: false

R-KFK-PROD-X1: без идемпотентности retry producer-а создаёт дубликаты. Если retries=3 и broker подтвердил публикацию, но ACK потерялся в сети — producer повторяет, broker получает то же событие дважды.

acks=0 или acks=1

R-KFK-PROD-X2:

• acks=0 — fire-and-forget. Producer отправил, не ждёт подтверждения. Broker мог не записать — данные потеряны. Используется только для high-volume низкоценных метрик.
• acks=1 — leader подтвердил без репликации. Если leader падает между write и репликацией → данные потеряны.
• acks=all — leader ждёт репликации на min.insync.replicas followers. Только это гарантирует durability.

Send без partition key

R-KFK-PROD-X3: см. секцию выше. Round-robin = потеря порядка для aggregate.

KafkaTemplate.send в @Transactional с DB

R-KFK-PROD-X4: Kafka не поддерживает XA, не может быть участником 2PC с PostgreSQL. Любая попытка «атомарно commit в DB + send в Kafka» ломается при partial failure.

Подробнее — Distributed transactions и Outbox publishing.

Что запрещено — таблица

Антипаттерн	Правило	Что взамен
enable.idempotence: false	R-KFK-PROD-X1	true всегда
acks: 0 / acks: 1	R-KFK-PROD-X2	acks: all
Send без partition key для бизнес-событий	R-KFK-PROD-X3	aggregate id как key
KafkaTemplate.send в @Transactional с DB	R-KFK-PROD-X4	outbox pattern
@TransactionalEventListener для Kafka	R-KFK-PROD-4	outbox-relay
max.in.flight > 5 с идемпотентностью	R-KFK-PROD-1	≤ 5 (auto при idempotence=true)
Aggregate-целиком в payload	R-KFK-PROD-3	record с явными полями (см. Event design)

Куда дальше

• Kafka → раздел 1. Producer — нормативные формулировки.
• Outbox publishing — как send через outbox.
• Event design — payload format, eventId, version.
• Configuration — application.yml settings.
• Distributed → outbox + inbox — атомарность commit+publish.
• Distributed transactions — почему Kafka не XA.
• Security — TLS, ACLs.