Данную статью посвящаю тем, кто начинает изучать Kafka и собирается установить ее для тестирования в среде Windows. Так как, практика — лучший способ глубокого понимания теории, то приступим.
 
Устанавливать будем именно Kafka, а не Confluent:

• Начинать лучше от простого к сложному

• Для Kafka много разных компонентов от сторонних разработчиков. Пробуйте свою конфигурацию.

• Отсутствует необходимость в платных подписках.

• Лучшее понимание внутренней работы системы.

С учетом того, что разработчики Kafka с версии 4.3 исключили из дистрибутива ZooKeeper, устанавливать мы будем версию Kafka с внутренним консенсус-протоколом KRaft (Kafka Raft).

Также развернем Kafka с тремя брокерами, как в настоящем продакшн, чтобы как можно лучше понять особенности работы этого брокера. Данное решение обеспечит:

• Отказоустойчивость (при падении одного брокера кластер останется работоспособным).

• Балансировку нагрузки (партиции распределятся между узлами).

• Практическое понимание работы распределенного брокера сообщений.

Сразу обращу ваше внимание на важный момент: Kafka успешно запускается в среде Windows, и вы сможете создать свой первый топик. Однако при дальнейшем использовании могут возникнуть проблемы с совместимостью. Например, Windows может заблокировать log- и index- файлы Kafka при попытке удалить топик.

Чтобы избежать проблем с совместимостью при работе Kafka в Windows, воспользуемся Windows Subsystem for Linux (WSL). Это решение позволяет запустить Kafka в нативной Linux-среде прямо в Windows, что обеспечивает стабильную работу всех функций системы, включая создание и удаление топиков, а также корректную работу с файлами журналов:

• WSL актуален для Windows 10 и выше

• Позволяет избежать критических ошибок

• Обеспечивает полноценную работу всех компонентов Kafka

• Не требует установки виртуальных машин

• Даст возможность получить практику работы с Linux

Устанавливаем WSL

Итак, приступим. Начнем с установки WSL2.

Откройте PowerShell от имени администратора и выполните:

wsl -install

Указываем пользователя, пароль.

По окончании установки у вас появится возможность выбрать терминал Ubuntu в приложениях Windows (меню «Пуск») или в терминале Windows PowerShell. Также вы сможете работать с файловой системой Ubuntu, используя стандартные инструменты Windows, например проводник Windows.

Доступ к файловой системе Linux осуществляется через каталог \\wsl$\Ubuntu

Если вы только начинаете свое знакомство с Linux и работаете с командной оболочкой Bash, вам регулярно будет встречаться команда sudo. Эта команда предоставляет временные права администратора (root) для выполнения конкретных задач, подобно тому, как в Windows вы используете учётную запись с правами администратора для внесения изменений в систему.

Далее установим доступные обновления для нашей Linux-среды, для чего выполните команду:

sudo apt update && sudo apt upgrade -y

Устанавливаем Java

Apache Kafka написана на Java и Scala и для ее работы, а также работы компонентов Kafka нам необходимо установить Java. Kafka официально поддерживает несколько версий Java:

• Java 8 (устаревшая, не рекомендуется к использованию)

• Java 11 (LTS-версия)

• Java 17 (LTS-версия)

Мы с вами установим Java17, как наиболее современную LTS-версию:

sudo apt install openjdk-17-jdk -y

Проверим установку:

java -version

В ответ получим примерно такой результат:

Завершив настройку WSL, переходим к следующему этапу — установке и настройке Kafka.

Устанавливаем Apache Kafka

Устанавливать предлагаю версию 4.0.0. На самом деле выбор конкретной версии не принципиален, вы можете поменять ее на другую. Важно, чтобы версия поддерживала работу с внутренним консенсус-протоколом Kraft.

Скачиваем архив с Kafka

sudo wget https://dlcdn.apache.org/kafka/4.0.0/kafka_2.13-4.0.0.tgz -O /tmp/kafka.tgz

Затем распакуем и переместим файлы в /opt/kafka:

sudo tar -xzf /tmp/kafka.tgz -C /opt/    # Распаковываем архив в /opt/
sudo mv /opt/kafka_2.13-4.0.0 /opt/kafka # Перемещаем в /opt/kafka
sudo chown -R $USER:$USER /opt/kafka     # Даем права нашему пользователю

Создание и настройка кластера

Следующий этап — создание и настройка кластера, состоящего из трёх узлов.

Для развёртывания кластера Apache Kafka необходимо настроить три сервера (брокера): server1, server2 и server3. Каждому брокеру потребуется индивидуальный конфигурационный файл.

Создадим директорию kraft в папке /opt/kafka/config/ и разместим в ней следующие конфигурационные файлы: server1.properties, server2.properties, server3.properties:

mkdir -p /opt/kafka/config1/kraft/
sudo touch /opt/kafka/config1/kraft/server1.properties
sudo touch /opt/kafka/config1/kraft/server2.properties
sudo touch /opt/kafka/config1/kraft/server3.properties

Проверьте, все файлы должны быть сохранены в кодировке UTF-8 с форматом конца строки LF.

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

Чтобы определить IP-адрес WSL, выполните следующую команду:

hostname -I

Замените <ваш_IP> на IP-адрес, полученный командой hostname -I:

process.roles=broker,controller
node.id=1
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT:// <ваш_IP>:9092
controller.quorum.voters=1@<ваш_IP>:9093,2@<ваш_IP>:9095,3@<ваш_IP>:9097
controller.listener.names=CONTROLLER
log.dirs=/var/lib/kafka/broker1
num.partitions=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

process.roles=broker,controller
node.id=2
listeners=PLAINTEXT://:9094,CONTROLLER://:9095
advertised.listeners=PLAINTEXT:// <ваш_IP>:9094
controller.quorum.voters=1@<ваш_IP>:9093,2@<ваш_IP>:9095,3@<ваш_IP>:9097
controller.listener.names=CONTROLLER
log.dirs=/var/lib/kafka/broker2
num.partitions=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

process.roles=broker,controller
node.id=3
listeners=PLAINTEXT://:9096,CONTROLLER://:9097
advertised.listeners=PLAINTEXT:// <ваш_IP>:9096
controller.quorum.voters=1@<ваш_IP>:9093,2@<ваш_IP>:9095,3@<ваш_IP>:9097
controller.listener.names=CONTROLLER
log.dirs=/var/lib/kafka/broker3
num.partitions=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

Параметры, указанные в конфигурационных файлах, можно разделить на несколько групп:

1.Режим работы узла

process.roles – определяет роли узла. Возможны комбинации: только broker, только controller, или обе роли (в режиме KRaft):

• broker – отвечает за хранение данных и обработку клиентских запросов

• controller – управляет метаданными кластера (лидеры партиций и другие метаданные)

node.id – уникальный идентификатор узла в кластере (например, ваш брокер имеет ID=1).

2.Сетевые настройки

listeners – интерфейсы и порты для подключений:

• PLAINTEXT://:9092 – порт для клиентских подключений (продюсеры/консьюмеры), без шифрования.

• CONTROLLER://:9093 – порт для внутренней коммуникации контроллеров (Raft-трафик).

advertised.listeners – адрес, который брокер сообщает клиентам (должен быть доступен извне).

controller.listener.names – указывает, какой listener используется для Raft-коммуникации (CONTROLLER).

3.Кворум контроллеров (Kraft)

controller.quorum.voters - определяет состав Raft-кворума для управления метаданными:

• Формат: node_id@host:port. В нашем случае указаны 3 узла с одним IP, но разными портами (9093, 9095, 9097) – это эмулирует кластер на одной машине (например, для тестирования).

4.Хранение данных

log.dirs - директория для хранения: данных топиков, логов коммитов (Kraft), методанных кластера

5.Настройки топиков

num.partitions – количество партиций по умолчанию для новых топиков.

offsets.topic.replication.factor – фактор репликации для внутреннего топика __consumer_offsets (хранение смещений потребителей).

transaction.state.log.replication.factor – фактор репликации для топика транзакций.

transaction.state.log.min.isr –  минимальное число реплик (In-Sync Replicas) для записи в топик транзакций.

6.Как это работает вместе?

• Узел запускается как брокер + контроллер (ID=1).

• Клиенты подключаются к порту 9092 брокера ID=1 (через advertised.listeners).

• Контроллеры общаются между собой через порт 9093 (Raft-протокол).

• Данные хранятся в /var/lib/kafka/broker1.

• Кворум из 3 узлов (на одном IP, но разных портах) обеспечивает отказоустойчивость метаданных.

Запуск кластера Kafka

Надеюсь, предыдущие шаги не вызвали у вас сложности и можно приступить к финальному шагу – запуску кластера Kafka

1.Подготовка переменных

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

KAFKA_HOME=/opt/kafka
BROKER1_CONF=$KAFKA_HOME/config/kraft/server1.properties
BROKER2_CONF=$KAFKA_HOME/config/kraft/server2.properties
BROKER3_CONF=$KAFKA_HOME/config/kraft/server3.properties
CLUSTER_ID=my-kafka-cluster-123 #уникальный идентификатор кластера

2.Форматирование хранилища Kafka для каждого брокера

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

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

# Брокер 1
sudo $KAFKA_HOME/bin/kafka-storage.sh format \
  -t $CLUSTER_ID \
  -c $BROKER1_CONF \
  --ignore-formatted

# Брокер 2
sudo $KAFKA_HOME/bin/kafka-storage.sh format \
  -t $CLUSTER_ID \
  -c $BROKER2_CONF \
  --ignore-formatted

# Брокер 3
sudo $KAFKA_HOME/bin/kafka-storage.sh format \
  -t $CLUSTER_ID \
  -c $BROKER3_CONF \
  --ignore-formatted

3.Создание каталогов для хранения логов внутри WSL

Чтобы Kafka работала корректно — особенно в окружении с несколькими брокерами — важно заранее организовать чёткую файловую структуру. Ниже приведены ключевые директории и их назначение.

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

Структура хранения логов и топиков

В окне того же терминала выполняем:

mkdir -p /var/lib/kafka/{broker1,broker2,broker3}    # data_dir для брокеров
mkdir -p /var/log/kafka/{broker1,broker2,broker3}    # логи Kafka
mkdir -p /opt/kafka/logs			            	 #логи пользователя

4.Настройка прав доступа

Далее назначим права доступа на директории Kafka, чтобы процесс мог работать от имени вашего пользователя, не требуя постоянного использования sudo.

sudo chown -R $USER:$USER /var/lib/kafka/
sudo chown -R $USER:$USER /var/log/kafka/ 
sudo chown -R $USER:$USER /opt/kafka/logs

5.Запуск брокеров

И наконец — пожалуй, один из самых ответственных и долгожданных моментов: запуск кластера Kafka.
Команду выполняем в том же терминале, где заданы переменные:

sudo chown -R $USER:$USER /var/lib/kafka/
sudo chown -R $USER:$USER /var/log/kafka/ 
sudo chown -R $USER:$USER /opt/kafka/logs

Скрипты для запуска кластера

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

Форматирование кластера

Для повторного форматирования и очистки создадим в папке /opt/kafka/ текстовый файл format-cluster.sh. Помним про необходимость придерживаться кодировки UTF-8 и форматом конца строки LF (Unix-стиль).

sudo touch /opt/kafka/format-cluster.sh

 Скопируйте в файл содержимое скрипта.

#!/bin/bash

# Конфигурация
KAFKA_HOME="/opt/kafka"
CLUSTER_ID="my-kafka-cluster-123"
DATA_DIR="/var/lib/kafka"
LOG_DIR="$KAFKA_HOME/logs"
LOCK_DIR="/var/lock/kafka"

# Функция очистки
cleanup() {
    echo "[1/5] Очистка старых данных..."
    pkill -f "kafka-server-start" || true
    rm -rf "$DATA_DIR"/{broker1,broker2,broker3}
    rm -rf "$LOCK_DIR"/{broker1,broker2,broker3}
    mkdir -p "$DATA_DIR"/{broker1,broker2,broker3}
    mkdir -p "$LOCK_DIR"/{broker1,broker2,broker3}
    mkdir -p "$LOG_DIR"
}

# Функция форматирования
format_storage() {
    echo "[2/5] Форматирование хранилищ..."
    for id in {1..3}; do
        echo "Форматирование брокера $id"
        "$KAFKA_HOME/bin/kafka-storage.sh" format \
            -t "$CLUSTER_ID" \
            -c "$KAFKA_HOME/config/kraft/server$id.properties" \
            --ignore-formatted || {
                echo "[ОШИБКА] Форматирование брокера $id не удалось"
                exit 1
            }
    done
}

cleanup
format_storage

Дайте права на выполнение:

sudo chmod +x /opt/kafka/format-cluster.sh

После этого можете запускать ваш скрипт:

/opt/kafka/format-cluster.sh

Осторожно! Скрипт удалит все данные кластера без возможности восстановления. Используйте только в тестовых целях и при необходимости полного сброса.

Запуск кластера

Аналогичным образом создайте в директории /opt/kafka/ скрипт run-cluster.sh:

sudo touch /opt/kafka/run-cluster.sh

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

/opt/kafka/run-cluster.sh

Дайте права на выполнение:

sudo chmod +x /opt/kafka/run-cluster.sh

После этого можете запускать ваш скрипт:

/opt/kafka/run-cluster.sh

Управление топиками

Далее приведу примеры основных операций по созданию и администрированию топиков. Основная утилита Kafka для работы с топиками — kafka-topics.sh.
Для начала достаточно освоить три сценария её использования:

• Создание топика

• Просмотр списка топиков

• Получение информации о конкретном топике

Полный список доступных флагов и опций можно получить, выполнив в терминале команду:

/opt/kafka/bin/kafka-topics.sh --help

Обратите внимание, знак обратного слеша (\) в командной строке используется как символ продолжения строки. Он позволяет разбить длинную команду на несколько физических строк для лучшей читаемости. В противном случае все флаги команды нужно указать в одну строку. Каждый флаг команды начинается с (--)

Пример:

# С использованием \
команда \
--флаг1 \
--флаг

# Без использования \
команда --флаг1 --флаг2

Создать топик

Команда для создания нового топика с заданными параметрами

/opt/kafka/bin/kafka-topics.sh \
 --create \
 --bootstrap-server localhost:9092 \
 --topic test-topic \
 --partitions 1 \
 --replication-factor 3 \

• create — указывает на операцию создания нового топика

• bootstrap-server — адрес и порт сервера Kafka для подключения (например, localhost:9092)

• topic — имя создаваемого топика (в примере: test-topic)

• partitions — количество партиций в топике (1 в данном случае)

• replication-factor — фактор репликации (количество копий данных, 3 в примере)

Посмотреть список топиков

Получение списка всех существующих топиков

/opt/kafka/bin/kafka-topics.sh \
 --list \
 --bootstrap-server localhost:9092 

• list — команда для вывода списка всех существующих топиков

• bootstrap-server — адрес сервера Kafka для подключения

Информация о топике

Получение детальной информации о конкретном топике

/opt/kafka/bin/kafka-topics.sh \
 --bootstrap-server localhost:9092 \
 --describe --topic test-topic

• bootstrap-server — адрес сервера Kafka

• describe — команда для получения детальной информации о топике

• topic — имя топика, для которого требуется информация

Работа с данными

Теперь перейдём к примерам работы с данными в Kafka. Чтобы протестировать обмен сообщениями через созданный кластер, запустим консольного продюсера и консьюмера. В арсенале Kafka для этого есть утилиты:
kafka-console-producer.sh и kafka-console-consumer.sh.

Полный список флагов этих утилит можно посмотреть, запустив их с опцией --help. Пока же рассмотрим только базовые возможности.

Запуск продюсера

/opt/kafka/bin/kafka-console-producer.sh \
 --bootstrap-server localhost:9092 \
 --topic test-topic

• bootstrap-server — адрес и порт Kafka-брокера для подключения (например, localhost:9092).

• topic <имя топика> — топик, в который будут отправляться сообщения

Запуск потребителя

/opt/kafka/bin/kafka-console-consumer.sh \
 --bootstrap-server localhost:9092 \
 --topic test-topic \
 --from-beginning 

• bootstrap-server — адрес и порт Kafka-брокера для подключения (например, localhost:9092).

• topic <имя топика> — топик, на сообщения которого нужно подписаться потребителю

• from-beginning - необязательный флаг, который указывает потребителю читать все сообщения с самого начала топика. Без этого флага потребитель начинает чтение с последнего оффсета (т.е. только новые сообщения, приходящие после запуска).

Теперь вы можете отправить сообщение из консоли продюсера (производителя) в консоль консьюмера (потребителя).

Устанавливаем UI интерфейс

Чтобы сделать кластер удобным в использовании и приблизить его по функциональности к облачным решениям, добавим веб-интерфейс на основе Kafka UI.

Существует множество инструментов для управления Kafka (например, Conduktor, Kafka Tool, Redpanda Console), но мы выберем Kafka UI, так как он:

• предоставляет широкие возможности мониторинга и управления

• не требует лицензии

• легко интегрируется через Docker или отдельный сервис

Для этого с GitHub скачиваем Java-приложение Kafka UI, например в папку /opt/kafka/ui/: https://github.com/provectus/kafka-ui/releases

sudo mkdir -p /opt/kafka/ui && sudo wget -O /opt/kafka/ui/kafka-ui-api-v0.7.2.jar https://github.com/provectus/kafka-ui/releases/download/v0.7.2/kafka-ui-api-v0.7.2.jar

Далее повторим уже знакомые нам действия для создания скрипта, который запустит наш Kafka UI.

В корневой директории /opt/kafka/ создайте скрипт run-kafka-ui.sh:

sudo touch /opt/kafka/run-kafka-ui.sh

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

#!/bin/bash

KAFKA_UI_JAR="/opt/kafka/ui/kafka-ui-api-v0.7.2.jar"
KAFKA_BROKERS="localhost:9092,localhost:9094,localhost:9096"
WEB_PORT="8080"

echo "Kafka UI будет доступен по адресу: http://localhost:$WEB_PORT"

java -jar "$KAFKA_UI_JAR" \
  --kafka.clusters.0.name=WSL_Kafka_Cluster \
  --kafka.clusters.0.bootstrapServers="$KAFKA_BROKERS" \
  --server.port="$WEB_PORT"

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

sudo chmod +x /opt/kafka/run-kafka-ui.sh

Теперь вы можете запустить Kafka UI одной командой:

/opt/kafka/run-kafka-ui.sh

После запуска сервиса перейдите в браузере по адресу: http://localhost:8080
Если всё настроено верно, откроется веб-интерфейс Kafka UI

Остановка брокеров

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

Для этой задачи Kafka предоставляет специальный инструмент — скрипт kafka-server-stop.sh. Он безопасно останавливает брокер, выполняя синхронизацию и сохранение текущего состояния.

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

KAFKA_HOME=/opt/kafka
BROKER1_CONF=$KAFKA_HOME/config/kraft/server1.properties
BROKER2_CONF=$KAFKA_HOME/config/kraft/server2.properties
BROKER3_CONF=$KAFKA_HOME/config/kraft/server3.properties

$KAFKA_HOME/bin/kafka-server-stop.sh $BROKER3_CONF
sleep 3
$KAFKA_HOME/bin/kafka-server-stop.sh $BROKER2_CONF
sleep 3
$KAFKA_HOME/bin/kafka-server-stop.sh $BROKER1_CONF

Скрипт для остановки брокера

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

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

Следующие шаги вам уже хорошо знакомы:

В корневой директории /opt/kafka/ создайте скрипт kafka-server-stop.sh:

/opt/kafka/run-kafka-ui.sh

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

/opt/kafka/run-kafka-ui.sh

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

/opt/kafka/run-kafka-ui.sh

Теперь вы можете остановить Kafka одной командой:

/opt/kafka/run-kafka-ui.sh

Заключение

Вот и все!

Мы разобрали основные шаги для установки Apache Kafka на Windows в режиме KRaft без использования ZooKeeper.

Теперь у вас есть рабочая лаборатория, где можно экспериментировать: создавать топики, запускать продюсеров и консьюмеров, тестировать отказоустойчивость и изучать поведение кластера.

Дальше вы можете:

• Углубить знания: изучить работу с разными типами данных (Avro, Protobuf), поэкспериментировать с настройками очистки логов (log retention) и сжатия.

• Расширить экосистему: попробовать подключить такие инструменты, как ksqlDB, Kafka Connect или UI-просмотрщики типа AKHQ, для большего удобства.

Не бойтесь менять конфигурации, пробовать новые инструменты и дополнять стек — именно так формируется настоящее понимание Kafka. Помните: никакая документация не заменит практики.

В следующей статье подробно разберём с вами, как работают продюсеры, консьюмеры и сами сообщения — от отправки до доставки.