Теория по Kafka:

Общие сущности:

Kafka (Кафка)	это брокер сообщений - распределенная платформа для обработки и передачи потоковых данных. Она предоставляет возможности для публикации, подписки и обработки потоковых данных в реальном времени
Producer (Производитель)	это клиент, который публикует (отправляет) сообщения в топики Kafka. Продюсеры генерируют данные и отправляют их на брокеры
Consumer (Потребитель)	это клиент, который подписывается на топики Kafka и считывает (получает) сообщения из них. консьюмеры обрабатывают данные, полученные от продюсеров
Broker (Брокер)	это сервер, который хранит и управляет данными в Kafka. Брокеры принимают данные от продюсеров, хранят их на диске и предоставляют их консьюмерам

---

Кафка выступает в качестве брокера:

---

Сущности Kafka:

Kafka Broker (Kafka Server/Kafka Node)	это сервер, который принимает, хранит и выдаёт сообщения
Zookeeper (Координатор)	кластер кафка - это хранилище данных с состоянием кластера, конфигурации и мета-данные
Controller (специальный брокер)	обеспечивает консистентность данных - это компонент, отвечающий за управление состоянием разделов и реплик в кластере Kafka. Он выполняет административные задачи, такие как выбор лидера, переназначение разделов и обработка сбоев брокеров. Контроллер работает в виде потока в одном из брокеров кластера Kafka и координирует и управляет общим состоянием кластера. Он обеспечивает, чтобы каждый раздел имел лидера и правильное распределение реплик для обеспечения отказоустойчивости и масштабируемости системы
Kafka Message (Record/Event)	сообщение Kafka - это основная единица данных, которая передается и обрабатывается. Сообщение представляет собой упорядоченную последовательность байтов и может содержать любую информацию, которую продюсер отправляет в топик Kafka. Имеет структуру: key-value pair
Kafka Topic (Топик)	это набор данных, которые отправляет продюсер и будет считывать консьюмер (в том же порядке, котором были отправленны). RabbitMQ так не умеет (там отправка/приём идёт только по приоритету). Так же топик может иметь несколько партиций
Partition (Партиция)	находятся внутри топиков, ускоряет чтение и запись данных путём параллелизации
Broker File System	это брокеровская файловая система - это место (broker), где хранятся данные топиков (папка ./logs)
Offset (оффсет)	это номер сообщения в партиции
Producer (Производитель)	это высокопроизводительный отправитель сообщений (клиент, который отправляет сообщения в топики Kafka. Продюсеры генерируют данные и отправляют их на брокеры
Consumer (Потребитель)	это высокопроизводительный получатель сообщений (клиент, который подписывается на топики Kafka и получает сообщения из них. консьюмеры обрабатывают данные, полученные от продюсеров)
Consumer Group (Группа потребителей)	это несколько консьюмеров, которые работают вместе для обработки сообщений из одного или нескольких топиков. Каждое сообщение в топике будет обработано только одним консьюмером в группе
Lag (Задержка)	это разница между последним доступным сообщением в партиции и текущим положением консьюмера. Задержка показывает, насколько консьюмер отстает от продюсера
Consumer Offsets (Консьюмеры оффсетов)	это информация о прогрессе консьюмера в чтении сообщений из топиков. Консьюмеры оффсетов отслеживают, на каком сообщении они остановились, чтобы продолжить чтение с этой позиции при следующем запуске

Kafka Message/сообщение Kafka:

Структура хранения топиков/партиций:

---

Как удалить данные из топика?

---

Репликация:

---

Конфигурации и логирование:

$ cd /var/log/kafka           # Логи kafka
$ ./kafka-topics.sh --version # Посмотреть версию kafka (в папке /opt/kafka/bin/)
$ cat /etc/schema-registry/schema-registry.properties # Конфигурация схемы реестров
$ cat /opt/kafka/config/server.properties             # Конфигурация kafka

Работа с сервисами

$ systemctl status kafka.service  # Просмотр хелсчека сервиса
$ journalctl -u kafka.service     # Все логи сервиса
$ journalctl -eu kafka.service    # Подробные логи сервиса
$ cd /var/log/kafka               # Локальные логи сервиса
$ cd /opt/kafka/config/craft      # Конфиги кафки (брокер, контроллер и конфиг сервера)
$ journalctl --no-tail --no-pager -p 7 -u kafka.service  # Подробные логи сервиса
$ journalctl --all --since "2024-05-22 09:00:00" --until "2024-05-22 10:00:00" --no-tail --no-pager -u kafka.service  # Логи сервиса с указанием времени события

[BROKER]:

$ cd /opt/kafka/bin/  # Директория с скриптами
$ ./kafka-topics.sh --bootstrap-server localhost:9092 --describe # Проверка количества подключенных брокеров кафка к ноде (в папке /opt/kafka/bin/)
$ ./kafka-server-start.sh /opt/kafka/config/server.properties    # Рестарт брокера

[ZOOKEEPER]:

$ echo stat | nc {ip}:9092               # Просмотр доступа к зукиперу
$ ./zookeeper-shell.sh localhost:2181    # Подключение к зукиперу
$ get /brokers/topics/{topic_name}       # Просмотр брокера в консоли зукипера
$ ./opt/zookeeper/bin/zkServer.sh status # Статус zookeeper
$ cat /opt/zookeeper/conf/zoo.cfg        # Конфигурация zookeeper
$ ./zkServer.sh version                  # Версия zookeeper (в папке /opt/zookeeper/bin/)

[TOPICS]:

$ cd /opt/kafka/bin/
$ ./kafka-topics.sh --list --bootstrap-server localhost:9092 # Вывести список всех топиков (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --describe --topic {topic_name} --bootstrap-server localhost:9092 # Посмотреть информацию о топике (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --alter --topic {topic_name} --bootstrap-server localhost:9092 --config {config_name}={config_value} # Изменение конфигурации топика (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --create --topic {topic_name} --bootstrap-server localhost:9092 # Создание топика по умолчанию (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --zookeeper zoo1.example.com:2181/kafka-cluster --create --topic {topic_name} --replication-factor 3 --partitions 8 # Создание топика с конкретными параметрами (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic {topic_name} # Удаление топика (в папке /opt/kafka/bin/)
$ ./kafka-topics.sh --bootstrap-server localhost:9092 --delete-all # Удаление всех топиков (в папке /opt/kafka/bin/) - Если выдаёт ошибку, то пишем скрипт:
--------------------------------------------------------------------------------- script_delete_all_topic.sh
#!/bin/bash
# Список всех топиков
topics=$(.kafka-topics.sh --list --bootstrap-server localhost:9092)
# Удаление топиков
for topic in $topics; do
    ./kafka-topics.sh --delete --topic "$topic" --bootstrap-server localhost:9092
    echo "Deleted topic: $topic"
done
---------------------------------------------------------------------------------
$ ./kafka-topics.sh --bootstrap-server localhost:9092 --describe                                        # Посмотреть конфигурацию всех топиков и брокеров (в папке /opt/kafka/bin/)
$ echo "Text message" | bin/kafka-console-producer.sh --broker-list localhost:9092 --topic {topic_name} # Отправка сообщения в топик
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic {topic_name} --from-beginning # Подключение к топику и чтение сообщения