Введение в Nginx

Nginx (произносится как "engine x") — это высокопроизводительное программное обеспечение с открытым исходным кодом, выполняющее функции веб-сервера, обратного прокси-сервера, балансировщика нагрузки, TCP/UDP-прокси и почтового прокси-сервера. Созданное Игорем Сысоевым в 2004 году, оно распространяется под лицензией BSD из 2 пунктов. Nginx завоевал популярность благодаря своей скорости, стабильности и низкому потреблению ресурсов, что делает его выбором для многих высоконагруженных сайтов, таких как Netflix, Dropbox, Яндекс и ВКонтакте.

Согласно данным W3Techs (по состоянию на апрель 2025 года), Nginx занимает первое место среди веб-серверов, обслуживая 33,8% всех веб-сайтов, опережая Apache (26,4%) и Cloudflare Server (23,4%). Это подчеркивает его широкое признание и надежность.


Как работает Nginx?

Nginx использует асинхронную событийно-ориентированную архитектуру, которая позволяет обрабатывать тысячи одновременных соединений с минимальным использованием ресурсов.

Основные аспекты его работы включают:
Один основной процесс и несколько рабочих процессов: Основной процесс управляет конфигурацией и координирует работу, а рабочие процессы обрабатывают запросы пользователей. Это снижает накладные расходы по сравнению с многопоточной моделью, используемой, например, в Apache.
Механизмы событий: Nginx поддерживает такие технологии, как kqueue (FreeBSD), epoll (Linux) и другие, для эффективной обработки сетевых соединений.
Оптимизация передачи данных: Использование технологий, таких как sendfile и асинхронный ввод/вывод (AIO), минимизирует копирование данных и ускоряет доставку контента.
Низкое потребление памяти: Например, для 10 000 неактивных HTTP keep-alive соединений требуется всего около 2,5 МБ памяти.

Запросы пользователей разбиваются на небольшие сетевые соединения, которые обрабатываются асинхронно. После обработки они собираются в единый ответ и отправляются клиенту. Одно соединение может обрабатывать до 1024 запросов, что значительно повышает производительность.


Для чего нужен Nginx?

Nginx универсален и применяется в различных сценариях:
Веб-сервер
Обслуживает статический контент (HTML, CSS, изображения, JavaScript) с высокой скоростью.

Обратный прокси
Перенаправляет запросы к другим серверам, скрывая их от клиента.

Балансировка нагрузки
Распределяет входящий трафик между несколькими серверами для повышения отказоустойчивости.

Кеширование
Сохраняет часто запрашиваемый контент для ускорения доставки.

Почтовый прокси
Поддерживает протоколы IMAP, POP3, SMTP с возможностью аутентификации через HTTP.

Безопасность
Поддерживает SSL/TLS, ограничение доступа по IP и защиту от DDoS-атак.

Nginx особенно эффективен для высоконагруженных веб-приложений, где требуется быстрая доставка контента и стабильность при большом количестве запросов.


Почему выбрать Nginx?

Nginx выделяется среди других веб-серверов, таких как Apache, по нескольким причинам:
Высокая производительность: Асинхронная архитектура позволяет обрабатывать больше запросов с меньшими ресурсами.
Эффективность для статического контента: Nginx быстрее Apache в доставке статических файлов, таких как изображения и CSS.
Модульная архитектура: Легко расширяется с помощью модулей для добавления новых функций.
Низкое потребление ресурсов: Минимизирует использование памяти и процессора, что идеально для серверов с ограниченными ресурсами.
Широкое применение: Используется крупными компаниями, такими как Netflix, Dropbox и WordPress.com, что подтверждает его надежность.

Nginx также может работать в связке с Apache: Nginx обрабатывает статический контент, а Apache — динамический, что оптимизирует производительность сайта.


#Java #middle #on_request #nginx

Простая установка Nginx на Ubuntu

Установка Nginx на Ubuntu проста и занимает всего несколько минут.

Обновите списки пакетов:

sudo apt update

Эта команда обновляет индекс пакетов для системы управления пакетами apt.

Установите Nginx:

sudo apt install nginx

Подтвердите установку, нажав Y и Enter, когда система запросит разрешение.

Запустите Nginx:

sudo systemctl start nginx

Эта команда запускает веб-сервер.

Включите автозапуск Nginx:

sudo systemctl enable nginx

Это гарантирует, что Nginx будет запускаться автоматически при перезагрузке системы.

Проверьте статус Nginx:

sudo systemctl status nginx

Если в выводе указано active (running), сервер работает корректно.

Для настройки брандмауэра (если используется ufw) разрешите HTTP-трафик:

sudo ufw allow 'Nginx HTTP'
sudo ufw status

Демонстрация работы Nginx
После выполнения вышеуказанных шагов откройте веб-браузер и введите IP-адрес вашего сервера (например, http://your_server_ip). Вы увидите стандартную страницу приветствия Nginx.
Эта страница подтверждает, что Nginx установлен и функционирует корректно. Если страница не отображается, проверьте статус сервера и настройки брандмауэра.

#Java #middle #on_request #nginx

Apache Kafka.

Введение и архитектура

Apache Kafka представляет собой распределенную платформу для обработки потоков данных в реальном времени, которая сочетает в себе функции очереди сообщений, хранилища данных и системы обработки событий. Разработанная изначально в LinkedIn для решения задач высоконагруженных систем, Kafka эволюционировала в мощный инструмент для построения масштабируемых конвейеров данных.


Основные концепции: topic, partition, offset, segment, log, leader/follower, ISR

Kafka строится вокруг понятия Topic — логическая категория для потоков сообщений.

Topic — это не монолитная структура, а распределенная очередь, разделенная на партиции (partitions). Каждая партиция представляет собой упорядоченную, неизменяемую последовательность записей (records), которая хранится как append-only лог. Это значит, что данные в партиции добавляются только в конец, без возможности модификации существующих записей. Партиции позволяют параллелизовать обработку: разные партиции могут обрабатываться независимо, что обеспечивает масштабируемость.

Внутри партиции каждая запись идентифицируется offset'ом — это монотонно возрастающее целое число, начиная с 0, которое указывает позицию записи в логе. Offset уникален только в пределах партиции; для разных партиций offset'ы независимы. Когда потребитель (consumer) читает данные, он отслеживает текущий offset, чтобы знать, с какой позиции продолжить чтение. В памяти потребителя offset хранится локально, но для надежности Kafka предоставляет механизм коммита offset'ов в специальную внутреннюю тему __consumer_offsets, где они реплицируются как обычные записи.

Партиция физически хранится как лог (log) — последовательность файлов на диске брокера. Лог разбивается на сегменты (segments) для управления размером: каждый сегмент — это файл с записями, начиная с определенного offset'а (base offset). Когда сегмент достигает заданного размера (по умолчанию 1 ГБ, создается новый. Старые сегменты могут удаляться по политикам retention. В памяти брокера сегменты не загружаются целиком; вместо этого Kafka использует memory-mapped files (mmap) для доступа к диску, что позволяет ОС кэшировать горячие данные в page cache, минимизируя реальные I/O-операции.

Для репликации партиции имеют лидера (leader) и фолловеров (followers). Лидер — это реплика партиции на одном брокере, которая принимает все записи от продюсеров (producers) и обслуживает чтение от потребителей. Фолловеры — реплики на других брокерах, которые синхронизируют данные с лидером. ISR (In-Sync Replicas) — это подмножество реплик (включая лидера), которые полностью синхронизированы с лидером. ISR определяется по отставанию фолловеров: если фолловер не запрашивает данные в течение replica.lag.time.max.ms (по умолчанию 30 секунд), он исключается из ISR. Это обеспечивает баланс между доступностью и consistency: записи считаются закоммиченными, когда они реплицированы на все реплики в ISR (min.insync.replicas).

В памяти брокера для каждой партиции лидер хранит в RAM метаданные, такие как текущий high-watermark (максимальный offset, закоммиченный на всех ISR), а также буферы для входящих запросов. Фолловеры используют отдельные потоки (replica fetchers) для pull-запросов к лидеру, копируя данные в свои логи.


Архитектура брокера: роль брокера, контроллера, брокерная конфигурация

Брокер (broker) — это основной узел (сервер) Kafka-кластера, отвечающий за хранение и обслуживание данных. Каждый брокер управляет подмножеством партиций: для каждой партиции один брокер является лидером, а другие — фолловерами. Брокеры образуют кластер, координируемый через ZooKeeper (до Kafka 2.8) или встроенный KRaft (Kafka Raft) в новых версиях, который устраняет зависимость от ZooKeeper.

#Java #middle #Kafka

Контроллер (controller) — это специальный брокер, избираемый кластером для управления метаданными: распределение партиций, лидер-элекшн, обработка изменений в топиках. Контроллер мониторит состояние брокеров через heartbeat'ы и перераспределяет партиции при сбоях. В памяти контроллера хранится глобальное состояние кластера: mapping партиций к брокерам, ISR для каждой партиции. При сбое контроллера избирается новый, что занимает миллисекунды благодаря репликации метаданных в __controller_epoch теме.

Брокерная конфигурация определяет поведение:
broker.id — уникальный ID,
log.dirs — директории для логов,
num.network.threads — количество потоков для сетевых запросов (по умолчанию 3),
num.io.threads — для дисковых операций (по умолчанию 8).

В памяти брокера значительная часть heap'а (до 50% по умолчанию) выделяется под off-heap буферы для сетевых операций, чтобы избежать GC-пауз. Конфигурация влияет на производительность: слишком малое num.io.threads может привести к bottleneck'у на диске, а большое message.max.bytes увеличивает потребление памяти для батчинга.


Хранение данных: лог-сегменты, индексные файлы, retention vs compaction

Данные в партиции хранятся в лог-сегментах: каждый сегмент состоит из двух файлов — .log (сами записи) и .index (спарс-индекс для быстрого поиска по offset'у).
Запись в .log — это последовательность байт: заголовок (magic byte, attributes, timestamp), ключ, значение, headers. Индексный файл содержит пары (offset, position), где position — байтовое смещение в .log. Индекс спарсный (по умолчанию каждые 4 КБ, configurable via index.interval.bytes), чтобы минимизировать размер: поиск offset'а начинается с ближайшей индексной записи, за которой следует линейный скан.

Retention — это политика удаления старых данных: log.retention.hours (по умолчанию 168) удаляет сегменты по времени, log.retention.bytes — по размеру. Compaction — альтернатива для key-based тем: сохраняет только последнюю запись для каждого ключа, удаляя дубликаты. Compaction работает в фоне: cleaner thread сканирует сегменты, строит в памяти map ключей к последним offset'ам, затем перезаписывает сегмент. В памяти это требует heap'а пропорционально количеству уникальных ключей в сегменте.

Trade-offs: retention подходит для временных данных (например, логи), но тратит диск на устаревшие записи; compaction экономит место для stateful данных (например, конфиги), но увеличивает CPU/IO на cleanup. Влияние segment.bytes: маленькие сегменты (например, 100 МБ) ускоряют deletion/compaction, снижая GC (меньше объектов в heap во время cleanup), но увеличивают overhead на открытые файлы и индексы. Большие сегменты минимизируют фрагментацию, но замедляют GC/IO при compaction, так как требуют больше памяти для временных структур.


Репликация: лидеры, ISR, replica fetcher, лидер-элекшн

Репликация обеспечивает надежность: каждая партиция имеет replication.factor (по умолчанию 1-3) реплик. Продюсеры пишут только в лидера, который реплицирует данные в ISR. Replica fetcher — это поток на фолловере, который периодически посылает FetchRequest к лидеру, запрашивая данные с последнего fetched offset'а. Лидер проверяет, в ISR ли фолловер, и отправляет данные. High-watermark продвигается только когда все ISR зареплицировали запись, обеспечивая durability.

Лидер-элекшн происходит при сбое лидера: контроллер выбирает нового лидера из ISR (предпочтительно unclean.leader.election.enable=false, чтобы избежать data loss). Элекшн использует epoch для предотвращения split-brain: новый лидер увеличивает epoch, уведомляя фолловеров. В памяти брокера реплика хранит log end offset (LEO) — максимальный offset в логе, и high-watermark.

#Java #middle #Kafka

Сетевая модель: request/response, metadata, fetch/produce

Kafka использует асинхронный, бинарный протокол на TCP: клиенты посылают requests (ProduceRequest, FetchRequest), брокеры отвечают responses. MetadataRequest запрашивает топик-метаданные (партиции, лидеры) от любого брокера, который перенаправляет к контроллеру если нужно. Produce — для записи: продюсер батчит записи по партициям, посылая в лидера. Fetch — для чтения: потребитель запрашивает с offset'а, получая chunk данных.

В памяти: брокер использует NIO selectors для multiplexing соединений, буферы (ByteBuffer) для zero-copy передачи. Zero-copy с sendfile() позволяет передавать данные из page cache напрямую в socket, без копирования в user space.


Обзор API-моделей

- Producer API: Для отправки записей. Создает ProducerRecord (topic, partition, key, value), использует KafkaProducer с конфигами (acks=all для durability, batch.size для батчинга).
- Consumer API: Для чтения. KafkaConsumer с poll(), который возвращает ConsumerRecords. Поддерживает группы для балансировки партиций.
- Admin API: Для управления топиками (create, delete, describe).
- Streams API: Для обработки потоков (KStream, KTable) с state stores.
- Connect API: Для интеграции с внешними системами (sources/sinks).


Ordering guarantees, масштабирование vs ordering

Kafka гарантирует ordering только внутри партиции: записи с одним ключом (если key-based partitioning) идут в порядке отправки. Нет глобального ordering по топику. Масштабирование добавлением партиций улучшает throughput, но жертвует ordering: для строгого ordering используйте одну партицию, что лимитирует parallelism.


Почему Kafka быстрая

- Sequential I/O: Запись/чтение в append-only лог — последовательные операции на диске, эффективные для HDD/SSD (миллионы IOPS vs random access).
- Zero-copy: Sendfile() копирует данные из kernel cache в socket без user space, снижая CPU и latency.
- Batching: Продюсеры/потребители батчат записи (linger.ms), амортизируя overhead сети/диска. В памяти батчи сжимаются (compression.type).


Минимальный producer/consumer

Producer (Java):

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");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");
producer.send(record);
producer.close();

Consumer (Java):

import org.apache.kafka.clients.consumer.*;
import java.util.Properties;
import java.util.Collections;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singleton("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.value());
}
consumer.close();

#Java #middle #Kafka

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

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

Метрики: 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:

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

Consumer — группы, оффсеты, ребалансинг

Apache Kafka Consumer — это клиент, ответственный за чтение и обработку сообщений из топиков.


Group coordinator, heartbeat, session.timeout.ms

Consumer groups позволяют нескольким потребителям параллельно обрабатывать топик, распределяя партиции между ними для load balancing. Группа идентифицируется по group.id; каждый потребитель в группе — member с уникальным member.id (генерируется при join).

Group coordinator — это брокер, выбранный для управления группой (hash(group.id) % num_brokers определяет координатора).

Координатор хранит состояние группы: members, assignments партиций, committed offsets (в внутренней теме __consumer_offsets).

В памяти брокера: состояние в GroupMetadataManager, с off-heap структурами для efficiency; offsets реплицированы как обычные записи.

Потребители поддерживают связь через heartbeat'ы: отдельный Heartbeat thread в KafkaConsumer посылает HeartbeatRequest каждые heartbeat.interval.ms (по умолчанию 3 сек) к координатору. Координатор отвечает, подтверждая membership. Если heartbeat не приходит в течение session.timeout.ms (по умолчанию 45 сек), member считается dead, триггеря rebalance.

В памяти потребителя: Coordinator хранит generation (epoch группы), member.id. Нюанс: высокое session.timeout.ms позволяет пережить GC-паузы или network glitches, но замедляет detection failures; низкое — приводит к frequent rebalances. При poll() heartbeat отправляется implicitly, если interval истек.


Offset management: авто-коммит vs ручной (commitSync, commitAsync)

Offsets тракают прогресс чтения: для каждой партиции — committed offset, с которого группа начнет после restart/rebalance.

Авто-коммит: Включается enable.auto.commit=true (default), с auto.commit.interval.ms (5 сек). При poll() потребитель аккумулирует processed offsets в памяти (OffsetAndMetadata), и каждые interval commit'ит их async. В памяти: Map<TopicPartition, OffsetAndMetadata> в KafkaConsumer, flushed periodically.

Trade-offs: удобно, но риск at-most-once (если crash после обработки, но до commit — потеря) или at-least-once (duplicates если commit после обработки, но перед full ack). Не используйте для critical processing.

Ручной коммит: enable.auto.commit=false. commitSync() — synchronous, блокирует до подтверждения от координатора (via OffsetCommitRequest), обеспечивает durability, но увеличивает latency. commitAsync() — asynchronous, с optional callback для error handling; faster, но требует manual retry на failures.

В памяти: offsets буферизуются до commit; при commitSync все in-flight commits ждут. Нюанс: commit только assigned партиций; при rebalance revoked offsets commit'ятся в onPartitionsRevoked. Для EOS используйте с transactions (read-process-write pattern).


#Java #middle #Kafka #Consumer

Rebalancing: Range, RoundRobin, Cooperative Sticky

Rebalancing — процесс перераспределения партиций при changes в группе (join/leave, failures). Триггерится coordinator'ом: все members посылают JoinGroupRequest, coordinator выбирает leader (первый joiner), который вычисляет assignments via assignor.

Assignors (partition.assignor.class):
- Range: Дефолт до 2.4. Партиции сортируются, делятся по range (например, для 10 partitions, 3 consumers: 0-3,4-6,7-9). Skew если num_partitions не делится evenly.
- RoundRobin: Распределяет round-robin для fairness, минимизируя skew.
- Cooperative Sticky (default с 2.4): Эволюция Sticky (минимизирует перемещения партиций). Cooperative — incremental: вместо full revoke-assign, использует несколько раундов (Eager vs Cooperative protocol). В первом раунде revoke только conflicting partitions, затем assign. Снижает downtime: consumers продолжают process revoked-later.

В памяти coordinator'а: хранит previous assignments для sticky. Нюанс: custom assignor implement ConsumerPartitionAssignor; для large groups (>100) rebalance может занять секунды из-за sync.


Static membership

Static membership избегает rebalance при restarts: group.instance.id (уникальный static ID per instance). При join с тем же ID, coordinator распознает как restart, не триггеря rebalance (если assignments unchanged).

В памяти: coordinator тракает instances в GroupMetadata. Нюанс: полезно для stateful consumers (с local state); но если instance меняет host, rebalance все равно. Комбинируйте с cooperative для minimal disruption.


Pause/Resume и backpressure

Для контроля flow: consumer.pause(Collection<TopicPartition>) останавливает fetch для партиций, но poll() продолжает heartbeat. resume() возобновляет. Используйте для backpressure: если downstream slow, pause до обработки backlog.

В памяти: paused partitions в Set<TopicPartition> в Fetcher; fetch requests skip them. Нюанс: paused не влияет на rebalance; при длительном pause lag растет, рискуя eviction из группы если max.poll.interval.ms истекает.

Backpressure: мониторьте lag, dynamically pause если queue full. В Streams это built-in via task pausing.


Метрики: lag, rebalance-latency

Consumer экспортирует метрики via KafkaConsumer.metrics():
- consumer-lag: Records-lag-max — максимальный lag (LEO - committed offset) по партициям. Вычисляется via FetchRequest с metadata. Высокий lag указывает на slow processing; мониторьте per-partition.
- rebalance-latency-avg: Среднее время rebalance (от trigger до completion). Включает join, sync, assign. Высокое — из-за large groups или slow assignors.

Другие: poll-time, commit-latency. Нюанс: используйте ConsumerMetrics для JMX; в production alert на lag > threshold или frequent rebalances.


#Java #middle #Kafka #Consumer

Пример кода

Пример на Java с subscribe и RebalanceListener для handling revoke/assign:

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false"); // Manual commit

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Commit offsets before losing partitions
        commitOffsets(); // Custom method to commit processed offsets
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Restore state or seek to offsets
        /* warm-up: e.g., load local state for partitions */
    }
});

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        // Process records
        consumer.commitAsync(); // Or commitSync for sync
    }
} finally {
    consumer.close();
}

В production: implement commitOffsets() с Map<TopicPartition, OffsetAndMetadata>; handle exceptions в listener.


Нюансы

- Избегание frequent rebalancing: Frequent rebalances (от scaling, GC, network) приводят к downtime (poll blocks во время). Static membership стабилизирует группу при restarts. CooperativeStickyAssignor минимизирует перемещения (до 80% меньше чем Range). Увеличьте session.timeout.ms до 300 сек для tolerance; heartbeat.interval.ms= session/3. Мониторьте rebalance-rate; если high — investigate app stability.

- max.poll.interval.ms и долгие операции: По умолчанию 5 мин — максимальное время между poll(). Если processing в poll() превышает (например, heavy computation), coordinator считает dead, триггеря rebalance. Решение: разбейте работу на chunks, poll() frequently; используйте pause() для long ops, но resume timely. Для очень долгих — offload в separate thread, но sync с poll(). Нюанс: в Streams это processing.guarantee=at_least_once handles.

- Обработка с сохранением ordering: Consumer гарантирует order только внутри партиции, но rebalance может нарушить если state не сохранен. В onPartitionsRevoked commit offsets и persist state (например, в external store). В onPartitionsAssigned seek(committed offset) и restore state. Для strict ordering: single-threaded per partition, или assign manually (без subscribe, используйте assign()). Нюанс: в groups с multiple consumers ordering cross-partition не гарантировано; для global order — single partition или external sorting.


#Java #middle #Kafka #Consumer