Apache Kafka
Producer — гарантии, производительность, транзакции
Apache Kafka Producer — это клиентская компонента, ответственная за отправку сообщений в Kafka-кластер.
Основные конфиги: acks, linger.ms, batch.size, compression.type, buffer.memory
Продюсер настраивается через свойства в Properties объекте (в Java/Scala) или аналогичные в других клиентах. Эти конфиги определяют баланс между latency, throughput и durability.
- acks: Определяет уровень подтверждения от брокера. Возможные значения: 0 (fire-and-forget, нет подтверждения, минимальная latency, но возможна потеря данных при сбое), 1 (подтверждение от лидера, данные записаны на диск лидера, но не реплицированы), all (или -1, подтверждение после репликации на все ISR, максимальная durability). В памяти продюсера: при acks=all запрос блокируется до получения подтверждения от min.insync.replicas брокеров. Это влияет на throughput: acks=0 дает миллионы сообщений/сек, но без гарантий; acks=all снижает до сотен тысяч из-за ожидания репликации. Нюанс: при acks=all, если ISR сокращается (например, из-за лага фолловеров), продюсер может бросить MetadataException, требуя ручного вмешательства.
- linger.ms: Время ожидания перед отправкой батча (по умолчанию 0). Продюсер аккумулирует сообщения в памяти (в RecordAccumulator), группируя по партициям. Если батч не заполнен в течение linger.ms, он отправляется принудительно. Это trade-off: высокое значение (например, 5-10 мс) увеличивает размер батча, амортизируя network overhead, но повышает latency. В памяти: таймер на основе ScheduledExecutorService проверяет батчи; overhead — минимальный, но при высоком linger.ms память может заполняться, если трафик низкий.
- batch.size: Максимальный размер батча в байтах (по умолчанию 16 КБ). Когда батч достигает этого размера, он отправляется немедленно, игнорируя linger.ms. В памяти: каждый батч — это Deque<ProducerBatch> в RecordAccumulator, где ProducerBatch содержит ByteBuffer для сериализованных записей. Нюанс: слишком большой batch.size (например, 1 МБ) увеличивает память (buffer.memory), но снижает I/O на брокере; малый — приводит к частым мелким запросам, увеличивая CPU на serialization и network.
- compression.type: Тип сжатия (none, gzip, snappy, lz4, zstd). Сжатие происходит в памяти продюсера перед добавлением в батч: сериализованные записи компрессируются в ByteBuffer. Это снижает объем данных на сети/диске, но добавляет CPU overhead. Подробнее в senior-нюансах ниже.
- buffer.memory: Общий размер буфера в памяти для несент батчей (по умолчанию 32 МБ). Это off-heap память (DirectByteBuffer), чтобы избежать GC. Если буфер заполнен, send() блокируется (configurable via max.block.ms). Нюанс: при высоком трафике и медленной сети буфер может переполниться, вызывая ProducerFencedException в транзакционных режимах; мониторьте метрики bufferpool-usage.
В памяти продюсера: основной компонент — RecordAccumulator, который держит Map<TopicPartition, Deque<ProducerBatch>>. Каждое send() сериализует ProducerRecord, вычисляет партицию (via Partitioner), добавляет в батч. Отдельный Sender thread (в KafkaProducer) периодически проверяет батчи и отправляет ProduceRequest через NetworkClient (NIO-based).
#Java #middle #Kafka #Produser
Producer — гарантии, производительность, транзакции
Apache Kafka Producer — это клиентская компонента, ответственная за отправку сообщений в Kafka-кластер.
Основные конфиги: acks, linger.ms, batch.size, compression.type, buffer.memory
Продюсер настраивается через свойства в Properties объекте (в Java/Scala) или аналогичные в других клиентах. Эти конфиги определяют баланс между latency, throughput и durability.
- acks: Определяет уровень подтверждения от брокера. Возможные значения: 0 (fire-and-forget, нет подтверждения, минимальная latency, но возможна потеря данных при сбое), 1 (подтверждение от лидера, данные записаны на диск лидера, но не реплицированы), all (или -1, подтверждение после репликации на все ISR, максимальная durability). В памяти продюсера: при acks=all запрос блокируется до получения подтверждения от min.insync.replicas брокеров. Это влияет на throughput: acks=0 дает миллионы сообщений/сек, но без гарантий; acks=all снижает до сотен тысяч из-за ожидания репликации. Нюанс: при acks=all, если ISR сокращается (например, из-за лага фолловеров), продюсер может бросить MetadataException, требуя ручного вмешательства.
- linger.ms: Время ожидания перед отправкой батча (по умолчанию 0). Продюсер аккумулирует сообщения в памяти (в RecordAccumulator), группируя по партициям. Если батч не заполнен в течение linger.ms, он отправляется принудительно. Это trade-off: высокое значение (например, 5-10 мс) увеличивает размер батча, амортизируя network overhead, но повышает latency. В памяти: таймер на основе ScheduledExecutorService проверяет батчи; overhead — минимальный, но при высоком linger.ms память может заполняться, если трафик низкий.
- batch.size: Максимальный размер батча в байтах (по умолчанию 16 КБ). Когда батч достигает этого размера, он отправляется немедленно, игнорируя linger.ms. В памяти: каждый батч — это Deque<ProducerBatch> в RecordAccumulator, где ProducerBatch содержит ByteBuffer для сериализованных записей. Нюанс: слишком большой batch.size (например, 1 МБ) увеличивает память (buffer.memory), но снижает I/O на брокере; малый — приводит к частым мелким запросам, увеличивая CPU на serialization и network.
- compression.type: Тип сжатия (none, gzip, snappy, lz4, zstd). Сжатие происходит в памяти продюсера перед добавлением в батч: сериализованные записи компрессируются в ByteBuffer. Это снижает объем данных на сети/диске, но добавляет CPU overhead. Подробнее в senior-нюансах ниже.
- buffer.memory: Общий размер буфера в памяти для несент батчей (по умолчанию 32 МБ). Это off-heap память (DirectByteBuffer), чтобы избежать GC. Если буфер заполнен, send() блокируется (configurable via max.block.ms). Нюанс: при высоком трафике и медленной сети буфер может переполниться, вызывая ProducerFencedException в транзакционных режимах; мониторьте метрики bufferpool-usage.
В памяти продюсера: основной компонент — RecordAccumulator, который держит Map<TopicPartition, Deque<ProducerBatch>>. Каждое send() сериализует ProducerRecord, вычисляет партицию (via Partitioner), добавляет в батч. Отдельный Sender thread (в KafkaProducer) периодически проверяет батчи и отправляет ProduceRequest через NetworkClient (NIO-based).
#Java #middle #Kafka #Produser
👍3
Idempotence: enable.idempotence=true, гарантии без дублей
Idempotent producer обеспечивает exactly-once семантику без дубликатов при ретреях. Включается via enable.idempotence=true, что implicitly устанавливает acks=all, retries>0, max.in.flight.requests.per.connection=5 (или меньше).
Как работает: каждому продюсеру присваивается уникальный producer.id (PID) от брокера при init. Каждому батчу добавляется sequence number (начиная с 0 per partition). Брокер хранит last sequence per PID/partition и отвергает дубликаты (если sequence уже обработан). При ретрее продюсер переотправляет с тем же sequence.
В памяти: продюсер держит Map<TopicPartition, Integer> для sequences. Overhead: минимальный, но требует стабильного соединения (если connection drops, может потребоваться reinitialization). Гарантии: at-least-once becomes exactly-once для одной сессии; не покрывает множественные продюсеры или app restarts (для этого — transactions).
Нюанс: idempotence не влияет на ordering, если max.in.flight=1; но по умолчанию позволяет до 5 in-flight, что может нарушить order (см. senior-нюансы).
Transactions (EOS): transactional.id, initTransactions(), ограничения и caveats
Transactions предоставляют exactly-once semantics (EOS) через топики, включая atomicity: все или ничего для группы send(). Включается via transactional.id (уникальный ID, persistent across restarts).
Процесс: producer.initTransactions() регистрирует PID с брокером (via InitProducerIdRequest), устанавливая epoch. Затем beginTransaction(), send()..., commitTransaction() или abortTransaction(). В commit брокер пишет маркеры (commit/abort) в логи, делая транзакцию видимой для потребителей с isolation.level=read_committed.
В памяти: продюсер держит TransactionManager, который тракает открытые транзакции, pending batches. Batches помечаются transactionally; overhead — дополнительные метаданные в ProduceRequest.
Ограничения: только для idempotent producers; max.in.flight=1 (принудительно, чтобы сохранить ordering); нельзя смешивать transactional и non-transactional send в одном продюсере. Caveats: timeout via transaction.timeout.ms (по умолчанию 60 сек), после чего транзакция фенсится (ProducerFencedException). При restart с тем же transactional.id старый PID фенсится, позволяя продолжить. Нюанс: в кластере с downtime контроллера initTransactions() может блокироваться; для EOS в Streams используйте processing.guarantee=exactly_once.
Partitioner: кастомный и дефолтный
Partitioner определяет, в какую партицию идет запись. Дефолтный (DefaultPartitioner): если key=null, round-robin; если key не null, hash(key) % num_partitions (murmur2 hash для consistency).
Кастомный: implement Partitioner interface (partition() method). Полезно для affinity (например, все orders пользователя в одну партицию для ordering). В памяти: вызывается синхронно в send(), overhead — от hash computation.
Нюанс: плохой partitioner может привести к skew (горячие партиции), снижая throughput; всегда учитывайте num_partitions changes.
Retry/backoff политика
Retries (по умолчанию 2147483647) и retry.backoff.ms (100 мс) управляют повторными попытками при transient errors (например, NotLeaderForPartitionException).
Как работает: Sender thread ловит retriable exceptions, backoff (exponential с jitter), затем retry. В idempotence/transactions retry прозрачен, без дубликатов.
В памяти: pending requests в InFlightRequests Map, с timeout per request. Нюанс: высокие retries могут маскировать проблемы (например, network flaps), увеличивая latency; мониторьте retry-rate метрики.
#Java #middle #Kafka #Produser
Idempotent producer обеспечивает exactly-once семантику без дубликатов при ретреях. Включается via enable.idempotence=true, что implicitly устанавливает acks=all, retries>0, max.in.flight.requests.per.connection=5 (или меньше).
Как работает: каждому продюсеру присваивается уникальный producer.id (PID) от брокера при init. Каждому батчу добавляется sequence number (начиная с 0 per partition). Брокер хранит last sequence per PID/partition и отвергает дубликаты (если sequence уже обработан). При ретрее продюсер переотправляет с тем же sequence.
В памяти: продюсер держит Map<TopicPartition, Integer> для sequences. Overhead: минимальный, но требует стабильного соединения (если connection drops, может потребоваться reinitialization). Гарантии: at-least-once becomes exactly-once для одной сессии; не покрывает множественные продюсеры или app restarts (для этого — transactions).
Нюанс: idempotence не влияет на ordering, если max.in.flight=1; но по умолчанию позволяет до 5 in-flight, что может нарушить order (см. senior-нюансы).
Transactions (EOS): transactional.id, initTransactions(), ограничения и caveats
Transactions предоставляют exactly-once semantics (EOS) через топики, включая atomicity: все или ничего для группы send(). Включается via transactional.id (уникальный ID, persistent across restarts).
Процесс: producer.initTransactions() регистрирует PID с брокером (via InitProducerIdRequest), устанавливая epoch. Затем beginTransaction(), send()..., commitTransaction() или abortTransaction(). В commit брокер пишет маркеры (commit/abort) в логи, делая транзакцию видимой для потребителей с isolation.level=read_committed.
В памяти: продюсер держит TransactionManager, который тракает открытые транзакции, pending batches. Batches помечаются transactionally; overhead — дополнительные метаданные в ProduceRequest.
Ограничения: только для idempotent producers; max.in.flight=1 (принудительно, чтобы сохранить ordering); нельзя смешивать transactional и non-transactional send в одном продюсере. Caveats: timeout via transaction.timeout.ms (по умолчанию 60 сек), после чего транзакция фенсится (ProducerFencedException). При restart с тем же transactional.id старый PID фенсится, позволяя продолжить. Нюанс: в кластере с downtime контроллера initTransactions() может блокироваться; для EOS в Streams используйте processing.guarantee=exactly_once.
Partitioner: кастомный и дефолтный
Partitioner определяет, в какую партицию идет запись. Дефолтный (DefaultPartitioner): если key=null, round-robin; если key не null, hash(key) % num_partitions (murmur2 hash для consistency).
Кастомный: implement Partitioner interface (partition() method). Полезно для affinity (например, все orders пользователя в одну партицию для ordering). В памяти: вызывается синхронно в send(), overhead — от hash computation.
Нюанс: плохой partitioner может привести к skew (горячие партиции), снижая throughput; всегда учитывайте num_partitions changes.
Retry/backoff политика
Retries (по умолчанию 2147483647) и retry.backoff.ms (100 мс) управляют повторными попытками при transient errors (например, NotLeaderForPartitionException).
Как работает: Sender thread ловит retriable exceptions, backoff (exponential с jitter), затем retry. В idempotence/transactions retry прозрачен, без дубликатов.
В памяти: pending requests в InFlightRequests Map, с timeout per request. Нюанс: высокие retries могут маскировать проблемы (например, network flaps), увеличивая latency; мониторьте retry-rate метрики.
#Java #middle #Kafka #Produser
Метрики: request-latency, batch-size-avg
Продюсер экспортирует JMX метрики via KafkaProducer.metrics().
Ключевые:
- request-latency-avg: Средняя latency ProduceRequest (от send до response). Включает network + broker processing. Высокая — указывает на bottleneck (медленная сеть, перегруженный брокер).
- batch-size-avg: Средний размер батча в байтах. Идеально близко к batch.size для efficiency; низкий — увеличьте linger.ms.
Другие: record-queue-time-avg (время в accumulator), buffer-bytes (используемая память). Нюанс: используйте MetricsReporter для интеграции с Prometheus; в production мониторьте per-partition метрики для detection skew.
Пример кода
Вот расширенный пример на Java, демонстрирующий idempotence и transactions:
В production добавьте Callback в send() для async handling, и flush() перед close().
Нюансы: max.in.flight.requests.per.connection влияние на ordering, компрессия: zstd vs snappy vs lz4, когда idempotence недостаточно и нужна транзакция
- max.in.flight.requests.per.connection: По умолчанию 5, позволяет параллельные запросы по одному соединению для throughput. Но если >1 и retry происходит, ordering может нарушиться: failed batch retry после successful последующих. В idempotence это разрешено, но для strict ordering установите=1 (снижает throughput на 20-50%). В transactions forcibly=1.
- Компрессия: zstd — лучший ratio (до 5x), но высокий CPU (для высоких данных); snappy — быстрый, низкий CPU, средний ratio (2-3x), идеален для real-time; lz4 — баланс, faster than gzip. В памяти: compression в отдельном thread pool (если compression.type set), overhead — allocate temp buffers. Тестируйте: для text-heavy — zstd; binary — snappy.
- Когда idempotence недостаточно: Idempotence покрывает retries в одной сессии, но не atomicity across topics/partitions или app failures (дубликаты при restart). Транзакции нужны для EOS в multi-topic writes (например, order + inventory update) или с Kafka Streams. Caveat: transactions добавляют 10-20% overhead из-за маркеров и fencing. Используйте если SLA требует no duplicates/loss even on crashes; иначе idempotence достаточно для 99% случаев.
#Java #middle #Kafka #Produser
Продюсер экспортирует JMX метрики via KafkaProducer.metrics().
Ключевые:
- request-latency-avg: Средняя latency ProduceRequest (от send до response). Включает network + broker processing. Высокая — указывает на bottleneck (медленная сеть, перегруженный брокер).
- batch-size-avg: Средний размер батча в байтах. Идеально близко к batch.size для efficiency; низкий — увеличьте linger.ms.
Другие: record-queue-time-avg (время в accumulator), buffer-bytes (используемая память). Нюанс: используйте MetricsReporter для интеграции с Prometheus; в production мониторьте per-partition метрики для detection skew.
Пример кода
Вот расширенный пример на Java, демонстрирующий idempotence и transactions:
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("enable.idempotence", "true");
props.put("transactional.id", "order-tx-1");
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
producer.initTransactions();
producer.beginTransaction();
ProducerRecord<String, String> record = new ProducerRecord<>("orders", "key", "value");
producer.send(record);
producer.commitTransaction();
} catch (ProducerFencedException e) {
// Handle fencing, e.g., abort and reinitialize
} catch (KafkaException e) {
// Abort on error
producer.abortTransaction();
}
В production добавьте Callback в send() для async handling, и flush() перед close().
Нюансы: max.in.flight.requests.per.connection влияние на ordering, компрессия: zstd vs snappy vs lz4, когда idempotence недостаточно и нужна транзакция
- max.in.flight.requests.per.connection: По умолчанию 5, позволяет параллельные запросы по одному соединению для throughput. Но если >1 и retry происходит, ordering может нарушиться: failed batch retry после successful последующих. В idempotence это разрешено, но для strict ordering установите=1 (снижает throughput на 20-50%). В transactions forcibly=1.
- Компрессия: zstd — лучший ratio (до 5x), но высокий CPU (для высоких данных); snappy — быстрый, низкий CPU, средний ratio (2-3x), идеален для real-time; lz4 — баланс, faster than gzip. В памяти: compression в отдельном thread pool (если compression.type set), overhead — allocate temp buffers. Тестируйте: для text-heavy — zstd; binary — snappy.
- Когда idempotence недостаточно: Idempotence покрывает retries в одной сессии, но не atomicity across topics/partitions или app failures (дубликаты при restart). Транзакции нужны для EOS в multi-topic writes (например, order + inventory update) или с Kafka Streams. Caveat: transactions добавляют 10-20% overhead из-за маркеров и fencing. Используйте если SLA требует no duplicates/loss even on crashes; иначе idempotence достаточно для 99% случаев.
#Java #middle #Kafka #Produser
👍2