3) Установка Kafka по топологии Kafka3-F2

В этом руководстве кластер Kafka будет развёртываться на серверах: worker4 worker5 worker6 - в вашем случае это могут быть другие сервера.

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

3.1) Предварительные требование

Необходимо иметь три сервера, на которые будем устанавливать кластер кафки.

Необходимо иметь ssh-доступ к этим серверам. В нашем случае доступ к серверам будет осуществляться с помощью команд:

ssh worker4
ssh worker5
ssh worker6

Также в дальнейшем используется, что эти сервера имеют следующие IP-адреса:

worker4 - 192.168.17.134
worker5 - 192.168.17.135 
worker6 - 192.168.17.136

В вашем случае это может быть иначе, учтите это в дальнейшем.

3.2) Установка OpenJDK

Zookeeper написан на языке программирования Java и поэтому он нуждается в JVM-е, и её нужно установить. Для этого запустите команды:

ssh worker4 sudo apt install -y openjdk-17-jre
ssh worker5 sudo apt install -y openjdk-17-jre
ssh worker6 sudo apt install -y openjdk-17-jre

3.3) Скачивание и проверка дистрибутивов

Также зайдите на своём компьютере в директорию dist

cd dist

Если её нет, то создайте. И скачайте последнюю версию программы со страницы:

https://kafka.apache.org/downloads

Например, такими командами:

wget https://downloads.apache.org/kafka/3.4.0/kafka_2.12-3.4.0.tgz
wget https://downloads.apache.org/kafka/3.4.0/kafka_2.12-3.4.0.tgz.sha512

Проверяем скаченный дистрибутив на отсутствие ошибок скачивания:

cat kafka_2.13-3.4.0.tgz.sha512
sha512sum kafka_2.13-3.4.0.tgz

Проверьте хэш-суммы - они должны быть идентичны.

3.4) Копирование дистрибутивов на сервера

Предполагается, что на серверах уже есть директория dist в домашней директории пользователя. Если нет, то создайте:

ssh worker4 mkdir dist
ssh worker5 mkdir dist
ssh worker6 mkdir dist

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

scp kafka_2.13-3.4.0.tgz worker4:dist/
scp kafka_2.13-3.4.0.tgz worker5:dist/
scp kafka_2.13-3.4.0.tgz worker6:dist/

Теперь его нужно установить. Далее будет описано как установить его на один сервер. Вам нужно будут это сделать для всех серверов, кластера Kafka.

3.5) Установка на один сервер

Заходим на сервер и в папку dist:

ssh worker4
cd dist

Все программы в системах Unix, которые устанавливаются вручную, устанавливаются в директорию /opt. (Ubuntu является Linux-ом, а Linux является Unix-ом)

Убедитесь, что она существует. Если нет, то создайте её.

sudo mkdir -p /opt/

Вы находитесь в директории ~/dist. Распакуйте дистрибутив:

tar xvf kafka_2.13-3.4.0.tgz

Появиться директория:

kafka_2.13-3.4.0

Перенесите её в opt:

sudo mv ./kafka_2.13-3.4.0 /opt/

Далее перейдём в папку /opt

cd /opt

Поменяйте владельца файлов на root:

sudo chown -Rv root: ./kafka_2.13-3.4.0

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

sudo ln -s ./kafka_2.13-3.4.0 kafka

Теперь переходим в папку kafka

cd /opt/kafka/

Создаём пользователя, из под которого будет работать сервер:

sudo useradd kafka

Создаём директорию, где будут храниться данные и логи:

sudo mkdir -p /data/kafka/data
sudo mkdir -p /data/kafka/logs

У вас эта директория может оказаться где-то в другом месте. Если так, то в дальнейшем не забудьте менять эту директорию на нужную вам.

Передадим их в собственность пользователя kafka:

sudo chown -Rv kafka: /data/kafka

Перенаправим логи:

sudo ln -s /data/kafka/logs logs

Теперь нужно настроить конфиг Kafka:

sudo mcedit -b config/server.properties

И меняем его текст на следующий:

broker.id=0
listeners=PLAINTEXT://192.168.17.134:9092
#advertised.listeners=PLAINTEXT://your.host.name:9092
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
auto.create.topics.enable=true
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

log.dirs=/data/kafka/data
unclean.leader.election.enable=true
num.partitions=48
num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=2
transaction.state.log.replication.factor=2
transaction.state.log.min.isr=2

#log.flush.interval.messages=10000
#log.flush.interval.ms=1000
log.retention.hours=-1
#log.retention.bytes=1073741824
#log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

zookeeper.connect=192.168.17.134:2181,192.168.17.135:2181,192.168.17.136:2181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

Нужно поменять broker.id=0 по таблице:

Также в параметре listeners=PLAINTEXT://192.168.17.134:9092 настроить IP-адрес текущего сервера

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

sudo -u kafka bin/kafka-server-start.sh config/server.properties

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

Ctrl+C и сервер остановиться.

3.6) Настройка systemd

Теперь нужно настроить сервис в операционной системе. Для этого создайте файл:

sudo mcedit -b /etc/systemd/system/kafka.service

И вставьте в него текст:

[Unit]
Description=Kafka Daemon
Documentation=https://kafka.apache.org/
Requires=network.target
After=network.target

[Service]    
Type=forking
WorkingDirectory=/opt/kafka
User=kafka
Group=kafka
Environment=KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
ExecStart=/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
TimeoutSec=30
Restart=on-failure

[Install]
WantedBy=default.target

После этого запустите:

sudo systemctl daemon-reload

И активируйте новый сервис:

sudo systemctl enable kafka

Ну а теперь запустите сам сервер как сервис в операционной системе:

sudo systemctl start kafka

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

sudo systemctl status kafka

Должно вылететь что-то типа:

● kafka.service - Kafka Daemon
Loaded: loaded (/etc/systemd/system/kafka.service; disabled; vendor preset: enabled)
Active: active (running) since Thu 2023-03-16 03:27:31 UTC; 4s ago
Docs: https://kafka.apache.org/
Process: 25408 ExecStart=/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties 
(code=exited, status=0/SUCCESS)
Main PID: 25747 (java)
Tasks: 73 (limit: 9402)
Memory: 318.8M
CPU: 3.768s
CGroup: /system.slice/kafka.service
└─25747 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 
-XX:+ExplicitGCInvokesConcurrent -XX:MaxInlineLevel=15 -Djava.awt.headless=true "-Xlog:gc*>

Mar 16 03:27:31 worker4 systemd[1]: Starting Kafka Daemon...
Mar 16 03:27:31 worker4 systemd[1]: Started Kafka Daemon.