Подключение и работа с кластером Kafka (тариф Base)

Последнее изменение: 30 июля 2026

Раздел содержит описание способов подключения к управлению кластером Kafka. Поддерживаются два типа подключения:

• SASL_PLAINTEXT - незашифрованное;
• SASL_SSL - зашифрованное (через SSL-сертификат).

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

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

Для начала работы с кластером Kafka необходимо выполнить предварительную настройку:

1. Установить клиент Kafka - скачивается дистрибутив Kafka (например, kafka_2.13-4.2.0) и распаковывается в удобную директорию;
2. Установить среду выполнения Java - для работы необходимо установить среду выполнения, с помощью команды: sudo apt install openjdk-17-jre-headless. Рекомендуется использовать версию OpenJDK 17 или выше;
3. Доступ к брокерам - адреса брокеров (bootstrap servers) предоставляются после развертывания кластера;
4. Учетные данные:
   • Имя пользователя (username) - стандартное значение client;
   • Пароль - предоставляется после развертывания кластера.
5. Корневой сертификат (для SSL-подключения) - сертификат для зашифрованного подключения предоставляется файлом при заказе сервиса Kafka.

Структура клиентского дистрибутива

После распаковки архива Kafka в директории bin доступны основные скрипты для управления:

Скрипт	Назначение
kafka-topics.sh	Управление топиками (создание, удаление, просмотр)
kafka-console-producer.sh	Отправка сообщений в топик
kafka-console-consumer.sh	Чтение сообщений из топика
kafka-consumer-groups.sh	Управление консьюмер-группами
kafka-configs.sh	Изменение конфигурации топиков и других объектов
kafka-acls.sh	Управление списками доступа (ACL)

В разделе ниже приведены примеры использования скриптов.

Подробное описание операций с данными и другими скриптами приведено в официальной документации Kafka. Дополнительную информацию можно получить, выполнив любой скрипт с флагом --help:

./kafka-topics.sh --help

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

Для подключения к Kafka используется файл настроек client.properties, который содержит параметры аутентификации, протокола и его механизмов (рекомендуются SCRAM_SHA_512 и SCRAM_SHA_256).

Незашифрованное подключение (SASL_PLAINTEXT)

Создается файл client.properties со следующим содержимым:

sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="client" \
    password="вашпароль";

Примечание

Для незашифрованного подключения (SASL_PLAINTEXT) используется порт 9091;

Зашифрованное подключение (SASL_SSL)

Для зашифрованного подключения (в production-средах рекомендуется использовать именно его) дополнительно требуется корневой сертификат и хранилище доверенных сертификатов (truststore).

Примечание

Для зашифрованного подключения (SASL_SSL) используется порт 9092.

Шаг 1. Создание truststore

С помощью утилиты keytool выполняется импорт сертификата в хранилище:

keytool -importcert -storetype PKCS12 -keystore /путь/к/трастстору/truststore.jks -alias myalias -file ca.crt -storepass любойвашпароль -keypass любойвашпароль

Параметры:

• keystore - путь к создаваемому хранилищу;
• alias - псевдоним сертификата в хранилище;
• file - путь к загруженному корневому сертификату;
• storepass - пароль для доступа к хранилищу;
• keypass - пароль для доступа к ключу.

Шаг 2. Настройка client.properties

В файл client.properties добавляются параметры SSL:

sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
    username="client" \
    password="вашпароль";
ssl.truststore.location=/путь/к/трастстору/truststore.jks
ssl.truststore.password=парольотхранилища

Работа с кластером Kafka

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

./kafka-topics.sh --bootstrap-server <IP машины 0>:9092,<IP машины 1>:9092,<IP машины X>:9092 --command-config ../config/client.properties --create --topic <имя топика> --partitions <количество партиций> --replication-factor <значение> --config min.insync.replicas=<значение>

Ознакомиться с рекомендациями по настройке топиков можно в разделе Настройка топиков.

По умолчанию топик создаётся с --partitions 10 --replication-factor 3 --config min.insync.replicas=2, поэтому, если нет необходимости создавать топик со специфическими настройками, эти флаги в команде можно не указывать.

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

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

./kafka-topics.sh --bootstrap-server <IP_брокера_0>:9092,<IP_брокера_1>:9092,<IP_брокера_2>:9092 --command-config ~/client.properties --list

Параметры:

• bootstrap-server - список брокеров кластера (адреса и порты); command-config - путь к файлу с настройками клиента;
• list - вывод списка топиков.

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

Выполняется команда с указанием брокеров, файла конфигурации и топика:

./kafka-console-producer.sh --bootstrap-server <IP_брокера_0>:9092,<IP_брокера_1>:9092,<IP_брокера_2>:9092 --command-config~/client.properties --topic test-topic

После выполнения команды сообщения вводятся построчно. Для завершения используется сочетание клавиш Ctrl+C.

Чтение сообщений из топика

Выполняется команда с указанием топика и, при необходимости, consumer group:

./kafka-console-consumer.sh --bootstrap-server <IP_брокера_0>:9092,<IP_брокера_1>:9092,<IP_брокера_2>:9092 --command-config~/client.properties --topic test-topic --from-beginning --group my-new-consumer-group

Параметры:

• from-beginning- чтение всех сообщений с начала (при отсутствии параметра чтение начинается с новых сообщений);
• group - название consumer group, которая создастся автоматически.

Управление пользователями и доступом

Рекомендации к созданию пароля

Длина пароля (рекомендуемая):

• для учётной записи пользователя - не менее 12 знаков;
• для учётных записей администраторов, технических и служебных учётных записей - не менее 16 знаков.

Сложность пароля: рекомендуется использовать уникальный пароль, содержащий символы как минимум трёх из четырёх указанных ниже групп (при отсутствии технических ограничений):

• буквы латинского алфавита в верхнем регистре (A-Z);
• буквы латинского алфавита в нижнем регистре (a-z);
• цифры (0-9);
• специальные символы и знаки пунктуации (например, !@#$%^&*(),.?).

Периодичность смены: рекомендуемая периодичность смены пароля - не реже одного раза в год.

Создание пользователей

Для удобства можно написать скрипт, который создает пользователя сразу для двух методов шифрования - SHA-256 и SHA-512:

#!/bin/bash
username="имя_пользователя"
password="сгенерированный_пароль"
bootstrap="<IP_брокера_0>:9092,<IP_брокера_1>:9092,<IP_брокера_X>:9092"

/opt/kafka/bin/kafka-configs.sh \
          --bootstrap-server $bootstrap \
          --alter \
          --add-config SCRAM-SHA-256=[password="$password"] \
          --command-config /opt/kafka/config/client.properties \
          --entity-type users \
          --entity-name $username

/opt/kafka/bin/kafka-configs.sh \
          --bootstrap-server $bootstrap \
          --alter \
          --add-config SCRAM-SHA-512=[password="$password"] \
          --command-config /opt/kafka/config/client.properties \
          --entity-type users \
          --entity-name $username

Запуск скрипта:

sudo ./kafka-user-add.sh

Просмотр данных о пользователе

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

sudo ./kafka-console-consumer.sh --bootstrap-server <IP_брокеров>:9092 --describe --command-config ../config/client.properties --entity-type users

Управление consumer groups

Consumer Group создается автоматически при обращении к ней. Например, при чтении сообщений из топика с указанием этой consumer group:

sudo ./kafka-console-consumer.sh --bootstrap-server <IP_брокеров>:9092 --command-config ../config/client.properties --topic test-topic --group test-group --from-beginning

Просмотр списка ACL

Действия с ACL из командной строки осуществляются скриптом kafka-acls.sh. Пример просмотра списка ACL:

sudo ./kafka-acls.sh --bootstrap-server <IP_брокера_0>:9092,<IP_брокера_1>:9092,<IP_брокера_X>:9092 --command-config ../config/client.properties --list

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

Изменение объема RAM

Для изменения объема оперативной памяти, выделенной под Kafka, направляется обращение в техническую поддержку с указанием требуемого объема RAM.

Изменение количества брокеров

Выполняется также через обращение в техническую поддержку.

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

После добавления брокера выполняется перераспределение существующих партиций с учетом нового узла с помощью утилиты kafka-reassign-partitions.sh:

./kafka-reassign-partitions.sh --bootstrap-server <адреса_брокеров> --generate --topics-to-move-json-file topics.json --broker-list "0,1,2" --execute

Примечание

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