sottoОтправка записи
Producer формирует ProducerRecord с key, value, headers и (опционально) timestamp. Key и value проходят через выбранные Serializer'ы — на JSON, Avro или Protobuf со schema registry; без согласованной сериализации consumer не прочитает данные.
Внутри producer асинхронен: send() кладёт запись в внутренний accumulator, отдельный sender-thread пакует записи в батчи и шлёт брокерам. Метод возвращает Future<RecordMetadata> и поддерживает callback для асинхронной обработки результата.
Выбор партиции
С ключом — partitioner по умолчанию вычисляет hash(key) % numPartitions, поэтому одинаковые ключи идут в одну партицию (порядок per-ключ).
Без ключа producer должен распределить по партициям равномерно. Современный «sticky partitioner» (с Kafka 2.4) выбирает партицию для целого батча, а не для каждого сообщения — это даёт лучший батчинг (меньше мелких отправок), не нарушая равномерность во времени. Можно подключить и custom partitioner для специфичного routing.
Батчинг и сжатие
Пропускная способность producer держится на батчинге. batch.size — максимальный размер партии в байтах. linger.ms — сколько ждать заполнения батча перед отправкой (0 = шлём сразу, маленькие батчи; чуть больше = больше throughput ценой задержки). compression.type (gzip/snappy/lz4/zstd) уменьшает сеть и диск ценой CPU; обычно берут lz4/zstd как баланс.
buffer.memory ограничивает общую память producer на буфер. Если брокер не успевает принимать и буфер заполняется, send() начинает блокироваться, а затем падает по timeout — это сигнал упереться в брокер/сеть.
Acks и долговечность
Параметр acks определяет, чего ждать producer перед подтверждением записи. acks=0 — fire-and-forget, возможны потери даже при сетевом сбое. acks=1 — ждать ACK от leader партиции (запись может потеряться, если leader упадёт ДО репликации). acks=all — ждать подтверждения от всех реплик в ISR (in-sync replicas); самая надёжная настройка.
Связка с min.insync.replicas на стороне брокера: acks=all даёт настоящую долговечность только если min.insync.replicas ≥ 2 — иначе запись успешна, даже когда осталась лишь одна реплика, и её потеря убьёт сообщение. Стандартная боевая комбинация — replication.factor=3, min.insync.replicas=2, acks=all.
Retries и порядок
Producer повторяет отправку при ретраиваемых ошибках (retries, delivery.timeout.ms). Здесь возникает опасность дублей и нарушения порядка: если первый запрос ещё в пути и producer уже отправил следующий, при повторе порядок может перепутаться.
max.in.flight.requests.per.connection ограничивает число неподтверждённых запросов на соединение. С enable.idempotence сохранять порядок при retry безопасно даже при значениях до 5; без идемпотентности для строгого порядка приходится ставить 1.
Идемпотентный producer и транзакции
enable.idempotence=true (по умолчанию в современных Kafka) присваивает producer'у уникальный PID и порядковый sequence number per partition. Брокер отбрасывает дубли при ретраях → нет дублей В РАМКАХ ОДНОЙ producer-сессии и партиции при сбоях клиента/сети. Это не сквозной exactly-once, а защита от собственных ретраев.
Для atomic-записи в несколько партиций/тем сразу есть транзакции producer'а (transactional.id, initTransactions/beginTransaction/commitTransaction). На них построен паттерн consume-process-produce внутри Kafka — основа exactly-once в потоковой обработке.
// типичный prod-набор для надёжной отправки
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("compression.type", "lz4");
props.put("linger.ms", "10"); // допускаем 10мс задержки ради батчингаКак отвечать на собеседовании
Разберите путь записи: send → accumulator → sender → батчи (linger.ms/batch.size/compression). Объясните выбор партиции с ключом (порядок per-ключ) и sticky-partitioner без ключа. Главное — acks и долговечность: acks=all + min.insync.replicas=2 + RF=3 как боевой дефолт. Назовите идемпотентного producer'а (PID+sequence → нет дублей при ретраях в одной сессии/партиции) и транзакции для atomic-многопартиционной записи; не путайте это с end-to-end exactly-once.