Apache Kafka: базовые настройки и отказоустойчивый режим

Компоненты
Базовые настройки
Отказоустойчивый режим
Настройка сервера
Настройка клиента
Графический интерфейс Kafdrop

Окружение

• Версия ОС: 7.3
• Конфигурация ОС: Рабочая станция, Сервер графический
• Редакция ОС: Стандартная
• Версия ПО: apache-kafka-4.3.0-1

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

Серверы образуют кластер, в котором брокеры обеспечивают хранение данных, а средства подключения — интеграцию с внешними системами. Кластер обладает масштабируемостью и отказоустойчивостью: выход из строя отдельного сервера не приводит к потере данных при корректной настройке репликации. Kafka обеспечивает высокую пропускную способность, сохраняет данные на диск и предоставляет потребителям возможность многократного перечитывания сообщений. 

 

Компоненты

Брокеры (brokers)

Брокер — это сервер Kafka. Он принимает сообщения от производителей, хранит их на диске и отдаёт по запросу потребителям. Один из брокеров выполняет роль контроллера: отвечает за административные операции и распределяет разделы между узлами.

 

Кластер (cluster)

Кластер состоит из одного или нескольких брокеров. Брокеры кластера совместно обрабатывают потоки сообщений и обеспечивают отказоустойчивость и масштабируемость системы. Внутри одного кластера существует кворум.

Кворум — это минимальное большинство голосующих узлов (для N узлов — ⌊N/2⌋ + 1), доступность которого необходима для принятия согласованных решений и сохранения работоспособности кластера.

 

Темы и разделы (topics и partitions)

Сообщения распределяются по темам. Тема делится на разделы — упорядоченные последовательности записей, которые нельзя изменить, а новые элементы добавляются только в конец. Разделы позволяют хранить объём данных больше, чем вмещает один сервер, и читать тему параллельно.

У каждого раздела есть ведущая реплика и, как правило, несколько ведомых. Ведущая обслуживает чтение и запись, ведомые копируют данные. При выходе ведущей из строя её место автоматически занимает одна из ведомых.

 

Производители (producers)

Производители — клиентские приложения, которые публикуют события в выбранные темы. Производитель определяет раздел либо циклически для распределения нагрузки, либо по ключу записи.

 

Потребители (consumers)

Потребители подписываются на темы и получают сообщения. Объединяются в группы потребителей. Каждая запись доставляется одному экземпляру потребителя в каждой подписанной группе.

 

Смещение (offset)

Потребитель сам управляет своим смещением (offset), поэтому может перечитывать старые сообщения или переходить к более новым. Благодаря этому брокеры не хранят информацию о прогрессе потребителя.

 

Базовые настройки

Параметры брокера

Основные параметры брокера в файле /opt/kafka/config/server.properties:

Параметр	Описание	Рекомендация
process.roles	Роль узла в KRaft: broker, controller или broker,controller.	Для одиночного узла — broker,controller.
node.id	Идентификатор узла.	Целое число, уникальное в кластере.
listeners	Адреса и протоколы прослушивания.	PLAINTEXT://:9092,CONTROLLER://:9093.
advertised.listeners	Адреса, которые брокер сообщает клиентам.	Должен быть доступен клиентам; для удалённых подключений не используйте localhost.
log.dirs	Каталоги хранения журналов.	Можно указать несколько через запятую.
num.partitions	Количество разделов по умолчанию.	Зависит от нагрузки; для тестов — 1.
default.replication.factor	Коэффициент репликации по умолчанию.	Для продуктивной среды — 3.
log.retention.hours	Время хранения сообщений.	По умолчанию 168 часов (7 дней).

Пример настройки для внешнего IP-адреса (один интерфейс):

Если сервер Kafka имеет фиксированный внешний или внутренний IP (например, 192.168.1.100), а клиенты подключаются с других машин, измените конфигурацию следующим образом:

# Брокер
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://192.168.1.100:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT

где:

• listeners привязывается к 0.0.0.0 (все интерфейсы), 
• advertised.listeners заполняется фактическим IP или доменным именем, доступным клиенту.

 

Параметры производителя

Основные параметры производителя:

Параметр	Описание	Значения
acks	Уровень подтверждения записи	0 — без подтверждения (быстро, возможна потеря); 1 — от лидера; all (или -1) — от всех синхронизированных реплик (максимальная надёжность).
retries	Количество повторных попыток	Рекомендуется > 0 для обработки временных сбоев.

 

Отказоустойчивый режим

Репликация разделов

Основой отказоустойчивости Kafka является репликация. Каждый раздел имеет одну ведущую реплику (leader) и ноль или более ведомых (followers). Ведущая обрабатывает все запросы, ведомые пассивно реплицируют данные. При отказе лидера одна из ведомых автоматически становится новой ведущей.

Kafka в режиме KRaft использует контроллеры, которые выбирают лидера и хранят метаданные кластера. Для принятия решений нужен кворум — большинство голосов.

 

Синхронизированные реплики (ISR)

Лидер поддерживает набор In-Sync Replicas (ISR) — реплик, которые полностью догнали лидера. Сообщение считается зафиксированным, когда все реплики из ISR получили его. Если количество ISR опускается ниже значения min.insync.replicas, раздел перестаёт принимать записи (при acks=all), что предотвращает потерю данных.

    

Выбор несинхронизированного лидера

При отказе всех реплик раздела возможны два варианта поведения:

• ожидание восстановления реплики из ISR — раздел остаётся недоступным, но данные не теряются (поведение по умолчанию);
• выбор первой восстановившейся реплики (unclean.leader.election.enable=true) — раздел становится доступным быстрее, но возможна потеря данных. Рекомендуется оставить значение false, чтобы приоритет отдавался согласованности данных.

 

Рекомендации для отказоустойчивого кластера

Для отказоустойчивого кластера рекомендуются следующие параметры:

• фактор репликации темы (default.replication.factor): 3;
  
• min.insync.replicas: 2;
• acks: all на производителе;
• размещение реплик на разных физических серверах или в разных зонах доступности.

Регулярный мониторинг метрики (--under-replicated-partitions) — в норме значение должно быть 0.

 

Настройка сервера 

Входные данные

В данной статье рассматривается пример построения отказоустойчивого сервера, состоящего из трёх узлов.

В рамках текущей инструкции используются следующие параметры:

для основного сервера:

• IP-адрес – 192.168.114.151;
• имя хоста – kafka1.

для сервера репликации:

• IP-адрес – 192.168.114.153;
• имя хоста – kafka2.

для дополнительного сервера:

• IP-адрес – 192.168.114.78;
• имя хоста – kafka3.

 

Предварительная настройка на каждом узле

В репозитории РЕД ОС пакет apacha-kafka получает регулярные обновления. Каждая версия имеет зависимость от версии Java. Для apacha-kafka версии 4.3 требуется Java 17. Подробно о совместимости версий см. на официальной странице Apacha Kafka. Проверьте версию:

java -version

Пример вывода:

openjdk version "17.0.20.1" 2026-08-18

Если Java отсутствует, установите её (потребуются права администратора):

sudo dnf install java-17-openjdk

Добавьте записи в /etc/hosts:

sudo nano /etc/hosts

Добавьте строки:

192.168.114.151 kafka1
192.168.114.153 kafka2
192.168.114.78  kafka3

 

Установка на каждом узле

Для установки программы через терминал выполните команду (потребуются права администратора):

sudo dnf install apache-kafka

После установки все исполняемые файлы находятся в /opt/kafka/bin/, конфигурация — в /opt/kafka/config/server.properties.
Перед первым запуском службы нужно отформатировать хранилище.
 

Конфигурационный файл

Откройте конфигурационный файл /opt/kafka/config/server.properties на каждом узле:

sudo nano /opt/kafka/config/server.properties

Общие параметры (одинаковы на всех трёх узлах):

process.roles=broker,controller
controller.quorum.voters=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
log.dirs=/opt/kafka/logs/kraft-combined-logs
num.partitions=3
default.replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false

где:

• process.roles=broker,controller — узел одновременно хранит данные и управляет кластером. Для учебного кластера это допустимо; для продуктивной среды контроллеры рекомендуется вынести на отдельные узлы .
• controller.quorum.voters — список всех контроллеров кластера в формате id@host:port. Именно этот параметр (а не bootstrap.servers) задаёт статический кворум для KRaft .
• listeners=PLAINTEXT://0.0.0.0:9092,... — привязка ко всем интерфейсам, чтобы узел был доступен как локально, так и из сети.
• advertised.listeners — адрес, который брокер сообщает клиентам и другим узлам. Не используйте localhost, если клиенты подключаются с других машин: они получат этот адрес и попытаются подключиться к своей собственной машине. Укажите IP или DNS-имя, разрешаемое с клиентских хостов .
• default.replication.factor=3 и min.insync.replicas=2 — обеспечивают отказоустойчивость: тема по умолчанию создаётся с тремя репликами, запись подтверждается минимум двумя из них.

Параметры, различающиеся по узлам:

Параметр	kafka1	kafka2	kafka3
node.id	1	2	3
advertised.listeners	PLAINTEXT://kafka1:9092	PLAINTEXT://kafka2:9092	PLAINTEXT://kafka3:9092

 

Генерация UUID кластера

UUID генерируется один раз на любом узле и используется на всех трёх:

/opt/kafka/bin/kafka-storage.sh random-uuid

Пример вывода:

A19aRWX7S22fXgiC0_ZK5A

Скопируйте полученное значение — оно понадобится при форматировании хранилища на каждом узле.
 

Форматирование хранилища

На каждом узле выполните форматирование с одним и тем же UUID. Замените <UUID> на значение, полученное ранее:

sudo -u kafka /opt/kafka/bin/kafka-storage.sh format \
  -t <UUID> \
  -c /opt/kafka/config/server.properties

Форматирование выполняется от имени пользователя kafka, чтобы файлы метаданных сразу получили правильного владельца.

Пример успешного вывода:

Formatting metadata directory /opt/kafka/logs/kraft-combined-logs with metadata.version 4.3-IV0.

 

Права на каталог данных

Служба работает под пользователем kafka, поэтому каталог данных должен принадлежать ему. Назначьте владельца:

sudo chown -R kafka:kafka /opt/kafka/logs/kraft-combined-logs

Проверьте владельца каталога:

ls -ld /opt/kafka/logs/kraft-combined-logs

Пример вывода:

drwxr-xr-x. 3 kafka kafka 4096 окт 21 15:29 /opt/kafka/logs/kraft-combined-logs

В столбцах владельца и группы должно быть указано kafka kafka.

 

Запуск службы

Для запуска службы выполните команду на всех трёх узлах:

sudo systemctl enable kafka --now

Для проверки состояния службы выполните команду:

sudo systemctl status kafka

В статусе должно быть отображено active (running).

 

Настройка Firewalld

При включённом firewalld откройте порт:

sudo firewall-cmd --add-port=9092/tcp --permanent
sudo firewall-cmd --add-port=9093/tcp --permanent
sudo firewall-cmd --reload

 

Проверка прослушивания порта

На каждом узле проверьте, что порты 9092 (клиенты) и 9093 (контроллеры) слушаются:

sudo ss -tlnp | grep -E ':(9092|9093)'

Ожидаемый вывод должен содержать LISTEN и процесс java для обоих портов.

Пример вывода:

ISTEN 0      50                 *:9092             *:*    users:(("java",pid=9435,fd=170))                         
LISTEN 0      50                 *:9093             *:*    users:(("java",pid=9435,fd=135)) 

 

Проверка кворума контроллеров

Выполните на любом узле:

/opt/kafka/bin/kafka-metadata-quorum.sh \
 --bootstrap-server kafka1:9092 \
 describe --status

Убедитесь, что в выводе присутствуют все три контроллера и выбран лидер. Пример вывода:

ClusterId:              A19aRWX7S22fXgiC0_ZK5A
LeaderId:               1
LeaderEpoch:            15
HighWatermark:          72
MaxFollowerLag:         73
MaxFollowerLagTimeMs:   0
CurrentVoters:          [{"id": 1, "endpoints": ["CONTROLLER://kafka1:9093"]}, {"id": 2, "endpoints": ["CONTROLLER://kafka2:9093"]}, {"id": 3, "endpoints": ["CONTROLLER://kafka3:9093"]}]
CurrentObservers:       []

где:

• clusterId — идентификатор кластера;
• leaderId — id контроллера-лидера кворума;
• leaderEpoch — номер эпохи лидера, растёт при каждой смене лидера;
• highWatermark — номер последней записи журнала метаданных;
• maxFollowerLag — максимальное отставание ведомого от лидера в записях;
• maxFollowerLagTimeMs — время отставания в мс;
• currentVoters — голосующие узлы кворума, они выбирают лидера и обеспечивают отказоустойчивость;
• currentObservers — узлы без права голоса, следящие за журналом метаданных.

Кластер KRaft сформировался.

 

Проверка отказоустойчивого кластера

Создание темы с репликацией

На любом узле создайте тему с тремя разделами и фактором репликации 3:

/opt/kafka/bin/kafka-topics.sh --create \
  --topic demo-events \
  --bootstrap-server kafka1:9092 \
  --partitions 3 \
  --replication-factor 3

Ожидаемый вывод:

Created topic demo-events.

 

Просмотр сведений о теме

Просмотр сведений о теме:

/opt/kafka/bin/kafka-topics.sh --describe \
 --topic demo-events \
 --bootstrap-server kafka1:9092

Пример вывода:

Topic: demo-events    PartitionCount: 3    ReplicationFactor: 3
    Topic: demo-events    Partition: 0    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
    Topic: demo-events    Partition: 1    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    Topic: demo-events    Partition: 2    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2

где:

• PartitionCount — количество разделов;
• ReplicationFactor —  коэффициент репликации (для трёх узлов — 3);
• для каждого раздела в Replicas указаны три разных node.id (1, 2, 3);
• Isr — содержит все три реплики, они синхронизированы;
• лидеры разделов распределены по разным узлам (в примере: 1, 2, 3).

 

Проверка отправки и получения сообщений

Запустите консольного производителя и отправьте сообщение в созданную тему:

/opt/kafka/bin/kafka-console-producer.sh \
  --bootstrap-server kafka1:9092 \
  --topic demo-events

После запуска откроется интерактивный режим. Введите несколько строк и нажмите Enter. Например:

Сообщение 1 
Сообщение 2 
Сообщение 3
Сообщение 4 
Сообщение 5 

В другом терминале запустите потребителя:

/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka1:9092 \
  --topic demo-events \
  --from-beginning

Убедитесь, что все отправленные сообщения получены.

 

Тест отказа узла

Остановите Kafka на одном из узлов (например, kafka2):

sudo systemctl stop kafka

Проверьте, что тема остаётся доступной для записи и чтения с оставшихся узлов. Создайте нового производителя и отправьте сообщение — при acks=all и min.insync.replicas=2 запись должна пройти, пока живы минимум два брокера из трёх.

Проверьте состояние реплик:

/opt/kafka/bin/kafka-topics.sh --describe \
  --topic demo-events \
  --bootstrap-server kafka1:9092

В Isr для разделов, лидером которых был остановленный узел, должно остаться две реплики. Метрика UnderReplicatedPartitions будет равна 2 — это ожидаемо при отказе одного узла. После восстановления узла (sudo systemctl start kafka) реплики догонят лидера, и Isr вернётся к трём значениям.

 

Настройка клиента

Если Java отсутствует, установите (потребуются права администратора):

sudo dnf install java-17-openjdk

Для установки программы через терминал выполните команду:

sudo dnf install apache-kafka

Добавьте записи в /etc/hosts:

sudo nano /etc/hosts

Добавьте строки:

192.168.114.151 kafka1
192.168.114.153 kafka2
192.168.114.78  kafka3

С клиентского хоста выполните команду, указав в --bootstrap-server все три узла:

/opt/kafka/bin/kafka-topics.sh --list \
  --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092

Пример вывода:

__consumer_offsets
demo-events

Это означает, что клиент подключился к кластеру и видит список тем.

 

Графический интерфейс Kafdrop

Для просмотра состояния тем, брокеров и групп потребителей можно развернуть веб-интерфейс Kafdrop. 

 

Веб-интерфейс

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

1. Убедитесь, что Docker установлен. Подробную информацию по установке docker и настройке доступа к среде контейнеризации см. в нашей инструкции «Установка и настройка docker».

2. Официальный образ Kafdrop находится в репозитории obsidiandynamics/kafdrop на Docker Hub. Запустите Kafdrop на одном из узлов (например, kafka1), указав все три брокера:

sudo docker run -d --restart unless-stopped --network host \
  -e KAFKA_BROKERCONNECT=kafka1:9092,kafka2:9092,kafka3:9092 \
  -e CMD_ARGS="--message.sendEnabled=true" \
  --name kafdrop \
  obsidiandynamics/kafdrop

Флаг --network host позволяет контейнеру использовать сеть хоста, поэтому имена kafka1, kafka2, kafka3 должны разрешаться на этом хосте.

3. Проверьте, что контейнер запустился:

sudo docker ps
sudo docker logs <ID_контейнера>

4. Если на узле с Kafdrop используется firewalld, откройте порт 9000, чтобы интерфейс был доступен не только локально, но и с других машин:

sudo systemctl start firewalld
sudo systemctl enable firewalld
sudo firewall-cmd --add-port=9000/tcp --permanent
sudo firewall-cmd --reload

5. Откройте браузер и перейдите по адресу: http://<IP_сервера>:9000 (если Kafdrop запущен на kafka1 http://kafka1:9000). 

 

Просмотр сообщений

На главной странице Kafdrop нажмите на название темы (например, demo-events):

На странице сведений о теме нажмите кнопку «View Messages».

Нажмите на кнопку «All Messages».

Отобразится список сообщений.

Если тема содержит несколько разделов, можно переключаться на разные разделы для просмотра или использовать функцию поиска для фильтрации сообщений по ключевым словам.

 

Отправка сообщений

Kafdrop по умолчанию разрешает только просмотр сообщений. Для отправки текстовых сообщений непосредственно из интерфейса при запуске контейнера необходимо добавить параметр --message.sendEnabled=true.

В интерфейсе Kafdrop отправка сообщений выполняется следующим образом:

1. Откройте страницу сведений о нужной теме. Нажмите кнопку «Add Message»:

В открывшейся панели заполните:

• Topic Partition (0) — раздел темы, в который уйдёт сообщение. 0 — это первый раздел.
• Key (Message key, optional) — ключ сообщения. Необязательное поле. В Kafka ключ определяет, в какой раздел попадёт сообщение (по хешу ключа). Если ключ не нужен — оставьте пустым.
• Value — само тело сообщения. Пример: {"key": "value"}. Можно заменить на JSON или любой текст.
• Headers (optional) — заголовки сообщения (пары ключ-значение). Необязательное поле. Используются для метаданных (например, content-type: application/json). Можно не заполнять.

Подтвердите отправку. 

После успешной отправки сообщение сразу записывается в тему и становится доступным для просмотра.

 

Удаление контейнера

Для удаления контейнера выполните команду:

sudo docker rm -f kafdrop

Дата последнего изменения: 27.10.2026

Если вы нашли ошибку, пожалуйста, выделите текст и нажмите Ctrl+Enter.